You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@rocketmq.apache.org by GitBox <gi...@apache.org> on 2020/03/12 08:07:54 UTC
[GitHub] [rocketmq] liujian16 opened a new issue #1838: New Programming
Model of TransactionMQProducer
liujian16 opened a new issue #1838: New Programming Model of TransactionMQProducer
URL: https://github.com/apache/rocketmq/issues/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.
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 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.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [rocketmq] 2259289435 commented on issue #1838: New Programming
Model of TransactionMQProducer
Posted by GitBox <gi...@apache.org>.
2259289435 commented on issue #1838: New Programming Model of TransactionMQProducer
URL: https://github.com/apache/rocketmq/issues/1838#issuecomment-598066085
请教下你怎么设计的，现在发送后立马就会触发回调，但此时本地数据库事务其实还没有提交，我是通过修改源码，把发送完触发回调的逻辑干掉了，然后开放了手工提交和手工回滚的接口出来，通过spring的事务回调来手工提交和回滚，然后另外一个回调是用来兜底判断，那个没问题，因为手工提交和回滚都是oneway的，有可能不成功。
2259289435@qq.com
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.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [rocketmq] wlliqipeng closed issue #1838: New Programming Model of
TransactionMQProducer
Posted by GitBox <gi...@apache.org>.
wlliqipeng closed issue #1838: New Programming Model of TransactionMQProducer
URL: https://github.com/apache/rocketmq/issues/1838
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [rocketmq] liujian16 commented on issue #1838: New Programming
Model of TransactionMQProducer
Posted by GitBox <gi...@apache.org>.
liujian16 commented on issue #1838: New Programming Model of TransactionMQProducer
URL: https://github.com/apache/rocketmq/issues/1838#issuecomment-598168943
Combined with some Spring Boot magic, TransactionMQProducer can be used just like ordinary producer.
```java
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.
```java
@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());
}
}
```
```java
class TransactionMessageSendEvent extends ApplicationEvent{
SendResult sendResult;
public TransactionMessageSendEvent(Object source, SendResult sendResult){
super(source);
this.sendResult = sendResult;
}
public SendResult getSendResult(){
return sendResult;
}
}
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [rocketmq] liujian16 edited a comment on issue #1838: New
Programming Model of TransactionMQProducer
Posted by GitBox <gi...@apache.org>.
liujian16 edited a comment on issue #1838: New Programming Model of TransactionMQProducer
URL: https://github.com/apache/rocketmq/issues/1838#issuecomment-598070108
我的提案就是想支持你的这种用法。 你说“发送后立马就会触发回调”，是因为现在的设计是让你在这个回调里开启事务，完成业务逻辑。 但是如果事务执行时间长了，兜底的回调就很难处理了。@2259289435
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [rocketmq] 2259289435 commented on issue #1838: New Programming
Model of TransactionMQProducer
Posted by GitBox <gi...@apache.org>.
2259289435 commented on issue #1838: New Programming Model of TransactionMQProducer
URL: https://github.com/apache/rocketmq/issues/1838#issuecomment-598526490
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 方法，事务消息这块多了一个概念，虽然真正使用的时候可以通过封装让业务无感，但对开源的通用中间件，我觉得应该用更好的方案，虽然现在业务中我是这么干的。
2259289435@qq.com
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.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [rocketmq] duhenglucky commented on issue #1838: New Programming
Model of TransactionMQProducer
Posted by GitBox <gi...@apache.org>.
duhenglucky commented on issue #1838: New Programming Model of TransactionMQProducer
URL: https://github.com/apache/rocketmq/issues/1838#issuecomment-600957986
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: )
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [rocketmq] liujian16 edited a comment on issue #1838: New
Programming Model of TransactionMQProducer
Posted by GitBox <gi...@apache.org>.
liujian16 edited a comment on issue #1838: New Programming Model of TransactionMQProducer
URL: https://github.com/apache/rocketmq/issues/1838#issuecomment-598168943
Combined with some Spring Boot magic, TransactionMQProducer can be used just like ordinary producer.
```java
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.
```java
@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());
}
}
```
```java
class TransactionMessageSendEvent extends ApplicationEvent{
SendResult sendResult;
public TransactionMessageSendEvent(Object source, SendResult sendResult){
super(source);
this.sendResult = sendResult;
}
public SendResult getSendResult(){
return sendResult;
}
}
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [rocketmq] liujian16 commented on issue #1838: New Programming
Model of TransactionMQProducer
Posted by GitBox <gi...@apache.org>.
liujian16 commented on issue #1838: New Programming Model of TransactionMQProducer
URL: https://github.com/apache/rocketmq/issues/1838#issuecomment-598070108
我的提案就是想支持你的这种用法。 你说“发送后立马就会触发回调”，是因为现在的设计是让你在这个回调里开启事务，完成业务逻辑。 但是如果事务执行时间长了，兜底的回调就很难处理了。
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [rocketmq] wlliqipeng commented on issue #1838: New Programming
Model of TransactionMQProducer
Posted by GitBox <gi...@apache.org>.
wlliqipeng commented on issue #1838: New Programming Model of TransactionMQProducer
URL: https://github.com/apache/rocketmq/issues/1838#issuecomment-608365487
This Issue will be close first, but please feel free to reopen it if you have any other issues.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [rocketmq] 2259289435 commented on issue #1838: New Programming
Model of TransactionMQProducer
Posted by GitBox <gi...@apache.org>.
2259289435 commented on issue #1838: New Programming Model of TransactionMQProducer
URL: https://github.com/apache/rocketmq/issues/1838#issuecomment-598073459
嗯嗯，我现在的场景还有事务消息还有可能嵌套事务消息，也有可能几个事务消息平行，所以感觉这个非兜底回调的用法很不方便，所以现在基本都是改造后用，没按原先这种设计意图用。
2259289435@qq.com
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.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services