Rocketmq: New Programming Model of TransactionMQProducer

[liujian16]

Under current design, LocalTransaction is executed after half message is successfully sent. A LocalTransaction may takes long time to complete, and while it is running , TransactionCheckerListener may be called to report the status of the transaction. It's very hard or even impossible for the TransactionCheckerListener to distinguish whether the transaction is on-going or is rollbacked.

I'd propose a new programming model of TransactionMQProducer.

1. Start local transaction
2. Execute local transaction logic
3. Call TransactionMQProducer to send half message and record SendResult
4. Commit or Rollback local transaction based on SendResult.
5. Tell MQ the local transaction state.

According to this design, there is no extra work to do after half message is sent, TransactionCheckerListener can be confident the transaction is rollbacked if it can find a record in the database. And Step 5 can be easily implemented using Spring Transaction bound event.

[2259289435]

请教下你怎么设计的，现在发送后立马就会触发回调，但此时本地数据库事务其实还没有提交，我是通过修改源码，把发送完触发回调的逻辑干掉了，然后开放了手工提交和手工回滚的接口出来，通过spring的事务回调来手工提交和回滚，然后另外一个回调是用来兜底判断，那个没问题，因为手工提交和回滚都是oneway的，有可能不成功。

[email protected]

From: liujian16
Date: 2020-03-12 16:08
To: apache/rocketmq
CC: Subscribed
Subject: [apache/rocketmq] New Programming Model of TransactionMQProducer (#1838)
Under current design, LocalTransaction is executed after half message is successfully sent. A LocalTransaction may takes long time to complete, and while it is running , TransactionCheckerListener may be called to report the status of the transaction. It's very hard or even impossible for the TransactionCheckerListener to distinguish whether the transaction is on-going or is rollbacked.
I'd propose a new programming model of TransactionMQProducer.
Start local transaction
execute local transaction logic
Call TransactionMQProducer to send half message and record SendResult
Commit or Rollback local transaction based on SendResult.
Tell MQ the local transaction state.
According to this design, there is no extract work to do after half message is sent, TransactionCheckerListener can be confident the transaction is rollbacked if it can find a record in the database. And Step 5 can be easily implemented using Spring Transaction bound event.
—
You are receiving this because you are subscribed to this thread.
Reply to this email directly, view it on GitHub, or unsubscribe.

[liujian16]

我的提案就是想支持你的这种用法。 你说“发送后立马就会触发回调”，是因为现在的设计是让你在这个回调里开启事务，完成业务逻辑。 但是如果事务执行时间长了，兜底的回调就很难处理了。@2259289435

[2259289435]

嗯嗯，我现在的场景还有事务消息还有可能嵌套事务消息，也有可能几个事务消息平行，所以感觉这个非兜底回调的用法很不方便，所以现在基本都是改造后用，没按原先这种设计意图用。

[email protected]

From: liujian16
Date: 2020-03-12 16:43
To: apache/rocketmq
CC: 2259289435; Comment
Subject: Re: [apache/rocketmq] New Programming Model of TransactionMQProducer (#1838)
我的提案就是想支持你的这种用法。 你说“发送后立马就会触发回调”，是因为现在的设计是让你在这个回调里开启事务，完成业务逻辑。 但是如果事务执行时间长了，兜底的回调就很难处理了。
—
You are receiving this because you commented.
Reply to this email directly, view it on GitHub, or unsubscribe.

[liujian16]

Combined with some Spring Boot magic, TransactionMQProducer can be used just like ordinary producer.

class FooService{
    @Autowired
    SpringTransactionMQProducer producer;

    @Transactional
    public void doSomething(){
           // 1. do you business
           businessLogic();
           // 2. send out message, that's it
           Message message = buildMessage(...);
           producer.sendTransactionMessage(message);
    }
}

The following is what make the magic happens provide that we make some change on TransactionMQProducer.

@Component
class SpringTransactionMQProducer{
    @Autowired
    TransactionMQProducer producer;
   @Autowired
    private ApplicationEventPublisher applicationEventPublisher;
    public sendTransactionMessage(Message message){
          SendResult sendResult = producer.sendMessage(message);
          applicationEventPublisher.publishEvent(new TransactionMessageSendEvent(this, sendResult));
    }

   @TransactionalEventListener(phase=AFTER_COMMIT)
    public void commitMessage(TransactionMessageSendEvent event){
         producer.commitMessage(event.getSendResult());
    }
   @TransactionalEventListener(phase=AFTER_ROLLBACK)
    public void rollbackMessage(TransactionMessageSendEvent event){             
         producer.rollbackMessage(event.getSendResult());
    }

}

class TransactionMessageSendEvent extends ApplicationEvent{
    SendResult sendResult;
    public TransactionMessageSendEvent(Object source, SendResult sendResult){
        super(source);
         this.sendResult = sendResult;
    }
    public SendResult getSendResult(){
         return sendResult;
    }
}

[2259289435]

1.关键代码移除 TransactionMQProducer.sendMessageInTransaction --> defaultMQProducerImpl.sendMessageInTransaction // localTransactionState = transactionListener.executeLocalTransaction(msg, arg); try {
// this.endTransaction(sendResult, localTransactionState, localException);
} catch (Exception e) {
log.warn("local transaction execute " + localTransactionState + ", but end broker transaction failed", e);
}2.新增方法 TransactionMQProducer.endTransaction --> defaultMQProducerImpl.endTransaction 3. 业务层 通过 hooks 拦截，如果是半消息不给终态的方法，并且是在某个业务事务中，添加事务回调，例如 spring 的 TransactionSynchronizationManager.registerSynchronization，然后在回调内手工提交终态
跟 spring 无关，只是因为项目内用 spring 来管理事务的比较多，但 rocketmq 可以不用考虑这块， 建议是 TransactionMQProducer 新增一个方法 sendHalfMessage，这个方法里面不会去调用 executeLocalTransaction，半消息需手动指定终态，保持原有设计，然后通过把底层的 defaultMQProducerImpl.endTransaction 公开到 TransactionMQProducer 这一层即可。坏处新增了一个 api 方法，事务消息这块多了一个概念，虽然真正使用的时候可以通过封装让业务无感，但对开源的通用中间件，我觉得应该用更好的方案，虽然现在业务中我是这么干的。

[email protected]

From: liujian16
Date: 2020-03-12 20:53
To: apache/rocketmq
CC: 2259289435; Mention
Subject: Re: [apache/rocketmq] New Programming Model of TransactionMQProducer (#1838)
Combined with some Spring Boot magic, TransactionMQProducer can be used just like ordinary producer.
class FooService{
@Autowired
SpringTransactionMQProducer producer;

@Transactional
public void doSomething(){
       // 1. do you business
       businessLogic();
       // 2. send out message, that it
       Message message = buildMessage(...);
       producer.sendTransactionMessage(message);
}

}
The following is what make the magic happens provide that we make some change on TransactionMQProducer.
@Component
class SpringTransactionMQProducer{
@Autowired
TransactionMQProducer producer;
@Autowired
private ApplicationEventPublisher applicationEventPublisher;
public sendTransactionMessage(Message message){
SendResult sendResult = producer.sendMessage(message);
applicationEventPublisher.publishEvent(new TransactionMessageSendEvent(this, sendResult));
}

@TransactionalEventListener(phase=AFTER_COMMIT)
public void commitMessage(TransactionMessageSendEvent event){
producer.commitMessage(event.getSendResult());
}
@TransactionalEventListener(phase=AFTER_ROLLBACK)
public void rollbackMessage(TransactionMessageSendEvent event){
producer.rollbackMessage(event.getSendResult());
}
}

class TransactionMessageSendEvent extends ApplicationEvent{
SendResult sendResult;
public TransactionMessageSendEvent(Object source, SendResult sendResult){
super(source);
this.sendResult = sendResult;
}
public SendResult getSendResult(){
return sendResult;
}
}
—
You are receiving this because you were mentioned.
Reply to this email directly, view it on GitHub, or unsubscribe.

[duhenglucky]

https://github.com/openmessaging/openmessaging-java/blob/master/openmessaging-api/src/main/java/io/openmessaging/producer/Producer.java
The API shown below is provided in OpenMessaging, any comments are welcome: )

[wlliqipeng]

This Issue will be close first, but please feel free to reopen it if you have any other issues.

[xyz327]

我也是这样做的。使用org.springframework.transaction.support.TransactionSynchronization 来绑定到spring的事务管理上。
修改发送半消息逻辑。直接拿到sendresult 绑定到上下文

1. 在beforeCommit将消息通过获取本事务的数据库链接来写入数据库
2. 在afterCompletion根据状态去把上下文的sendresult commit/rollback 掉
3. 由事务消息回查兜底

但是现在感觉偶尔会出现发送了半消息.没有确认然后也没有回查.
在 RMQ_SYS_TRANS_HALF_TOPIC 里面可以查到对应的消息，但是回查确没有对应的日志

[2259289435]

我这边4.6.0没问题，已经上线稳定运行很久了。没有回查你检查下是不是check那边，你这偶发看是不是压测可以复现

[email protected]

From: Xi Zhou
Date: 2020-04-29 14:28
To: apache/rocketmq
CC: 2259289435; Mention
Subject: Re: [apache/rocketmq] New Programming Model of TransactionMQProducer (#1838)
我也是这样做的。使用org.springframework.transaction.support.TransactionSynchronization 来绑定到spring的事务管理上。
修改发送半消息逻辑。直接拿到sendresult 绑定到上下文
在beforeCommit将消息通过获取本事务的数据库链接来写入数据库
在afterCompletion根据状态去把上下文的sendresult commit/rollback 掉
由事务消息回查兜底
但是现在感觉偶尔会出现发送了半消息.没有确认然后也没有回查.
在 RMQ_SYS_TRANS_HALF_TOPIC 里面可以查到对应的消息，但是回查确没有对应的日志
—
You are receiving this because you were mentioned.
Reply to this email directly, view it on GitHub, or unsubscribe.

[diwayou]

https://github.com/openmessaging/openmessaging-java/blob/master/openmessaging-api/src/main/java/io/openmessaging/producer/Producer.java
The API shown below is provided in OpenMessaging, any comments are welcome: )

I didn't find the prepare method implementation in the code https://github.com/apache/rocketmq/blob/master/openmessaging/src/main/java/io/openmessaging/rocketmq/producer/ProducerImpl.java, rocketmq have implemented this? i implement a demo with spring transaction manager like
`
@Slf4j
public class TransactionProducer {

private RocketMQTemplate rocketMQTemplate;

private DefaultMQProducerImpl producer;

private MessageConverter messageConverter;

public TransactionProducer(RocketMQTemplate rocketMQTemplate) {
    this.rocketMQTemplate = rocketMQTemplate;

    Field field = FieldUtils.getField(TransactionMQProducer.class, "defaultMQProducerImpl", true);

    this.producer = (DefaultMQProducerImpl) ReflectionUtils.getField(field, rocketMQTemplate.getProducer());
    if (this.producer == null) {
        throw new RuntimeException("不能发送事务消息");
    }

    this.messageConverter = rocketMQTemplate.getMessageConverter();
}

public SendResult send(String destination, org.springframework.messaging.Message<?> rocketMessage) throws MQClientException {
    Message msg = RocketMQUtil.convertToRocketMessage(messageConverter,
            StandardCharsets.UTF_8.toString(),
            destination, Objects.requireNonNull(messageConverter.toMessage(rocketMessage.getPayload(), rocketMessage.getHeaders())));
    // ignore DelayTimeLevel parameter
    if (msg.getDelayTimeLevel() != 0) {
        MessageAccessor.clearProperty(msg, MessageConst.PROPERTY_DELAY_TIME_LEVEL);
    }

    Validators.checkMessage(msg, producer.getDefaultMQProducer());

    SendResult sendResult;
    MessageAccessor.putProperty(msg, MessageConst.PROPERTY_TRANSACTION_PREPARED, "true");
    MessageAccessor.putProperty(msg, MessageConst.PROPERTY_PRODUCER_GROUP, producer.getDefaultMQProducer().getProducerGroup());
    try {
        sendResult = producer.send(msg);
    } catch (Exception e) {
        throw new MQClientException("send message Exception", e);
    }

    LocalTransactionState localTransactionState = LocalTransactionState.UNKNOW;
    switch (sendResult.getSendStatus()) {
        case FLUSH_DISK_TIMEOUT:
        case FLUSH_SLAVE_TIMEOUT:
        case SLAVE_NOT_AVAILABLE:
            localTransactionState = LocalTransactionState.ROLLBACK_MESSAGE;
            break;
        default:
            break;
    }

    final org.apache.rocketmq.client.producer.SendResult finalSendResult = sendResult;
    TransactionSynchronization transactionSynchronization = new TransactionSynchronizationAdapter() {
        @Override
        public void afterCompletion(int status) {
            LocalTransactionState localState;
            if (status == TransactionSynchronization.STATUS_ROLLED_BACK){
                localState = LocalTransactionState.ROLLBACK_MESSAGE;
            } else if (status == TransactionSynchronization.STATUS_UNKNOWN) {
                localState = LocalTransactionState.UNKNOW;
            } else {
                localState = LocalTransactionState.COMMIT_MESSAGE;
            }

            try {
                producer.endTransaction(finalSendResult, localState, null);
            } catch (Exception e) {
                log.warn("local transaction execute " + localState + ", but end broker transaction failed", e);
            }
        }
    };

    TransactionSynchronizationManager.registerSynchronization(transactionSynchronization);

    if (localTransactionState.equals(LocalTransactionState.ROLLBACK_MESSAGE)) {
        throw new RuntimeException("rollback message because send mq fail");
    }

    TransactionSendResult transactionSendResult = new TransactionSendResult();
    transactionSendResult.setSendStatus(sendResult.getSendStatus());
    transactionSendResult.setMessageQueue(sendResult.getMessageQueue());
    transactionSendResult.setMsgId(sendResult.getMsgId());
    transactionSendResult.setQueueOffset(sendResult.getQueueOffset());
    transactionSendResult.setTransactionId(sendResult.getTransactionId());
    transactionSendResult.setLocalTransactionState(localTransactionState);

    return transactionSendResult;
}

}
`