RocketMQ专题  ->  应用篇-事务消息
上一篇:应用篇-延时消息
下一篇:应用篇-批量消息

1、问题引入

这里的一个需求场景是:工行用户A向建行用户B转账1万元。

我们可以使用同步消息来处理该需求场景:

  1. 工行系统发送一个给B增款1万元的同步消息M给Broker
  2. 消息被Broker成功接收后,向工行系统发送成功ACK
  3. 工行系统收到成功ACK后从用户A中扣款1万元
  4. 建行系统从Broker中获取到消息M
  5. 建行系统消费消息M,即向用户B中增加1万元

这其中是有问题的:若第3步中的扣款操作失败,但消息已经成功发送到了Broker。对于MQ来说,只要消息写入成功,那么这个消息就可以被消费。此时建行系统中用户B增加了1万元。出现了数据不一致问题。

2、解决思路

解决思路是,让第1、2、3步具有原子性,要么全部成功,要么全部失败。即消息发送成功后,必须要保证扣款成功。如果扣款失败,则回滚发送成功的消息。而该思路即使用事务消息。这里要使用分布式事务解决方案。

使用事务消息来处理该需求场景:

  1. 事务管理器TM向事务协调器TC发起指令,开启全局事务

  2. 工行系统发一个给B增款1万元的事务消息M给TC

  3. TC会向Broker发送半事务消息prepareHalf,将消息M预提交到Broker。此时的建行系统是看不到Broker中的消息M的

  4. Broker会将预提交执行结果Report给TC。

  5. 如果预提交失败,则TC会向TM上报预提交失败的响应,全局事务结束;如果预提交成功,TC会调用工行系统的回调操作,去完成工行用户A的预扣款1万元的操作

  6. 工行系统会向TC发送预扣款执行结果,即本地事务的执行状态

  7. TC收到预扣款执行结果后,会将结果上报给TM

    预扣款执行结果存在三种可能性:

    1. // 描述本地事务执行状态
    2. public enum LocalTransactionState {
    3. COMMIT_MESSAGE, // 本地事务执行成功
    4. ROLLBACK_MESSAGE, // 本地事务执行失败
    5. UNKNOW, // 不确定,表示需要进行回查以确定本地事务的执行结果
    6. }
  8. TM会根据上报结果向TC发出不同的确认指令

    • 若预扣款成功(本地事务状态为COMMIT_MESSAGE),则TM向TC发送Global Commit指令
    • 若预扣款失败(本地事务状态为ROLLBACK_MESSAGE),则TM向TC发送Global Rollback指令
    • 若现未知状态(本地事务状态为UNKNOW),则会触发工行系统的本地事务状态回查操作。回查操作会将回查结果,即COMMIT_MESSAGE或ROLLBACK_MESSAGE Report给TC。TC将结果上报给TM,TM会再向TC发送最终确认指令Global Commit或Global Rollback
  9. TC在接收到指令后会向Broker与工行系统发出确认指令

    • TC接收的若是Global Commit指令,则向Broker与工行系统发送Branch Commit指令。此时Broker中的消息M才可被建行系统看到;此时的工行用户A中的扣款操作才真正被确认
    • TC接收到的若是Global Rollback指令,则向Broker与工行系统发送Branch Rollback指令。此时Broker中的消息M将被撤销;工行用户A中的扣款操作将被回滚

    以上方案就是为了确保 消息投递与 扣款操作 能够在一个事务中,要成功都成功,有一个失败,则全部回滚。

    以上方案并不是一个典型的XA模式。因为XA模式中的分支事务是异步的,而事务消息方案中的消息预提交与预扣款操作间是同步的。

3、基础

3.1、分布式事务

对于分布式事务,通俗地说就是,一次操作由若干分支操作组成,这些分支操作分属不同应用,分布在不同服务器上。分布式事务需要保证这些分支操作要么全部成功,要么全部失败。分布式事务与普通事务一样,就是为了保证操作结果的一致性。

3.2、事务消息

RocketMQ提供了类似X/Open XA的分布式事务功能,通过事务消息能达到分布式事务的最终一致。XA是一种分布式事务解决方案,一种分布式事务处理模式。

3.3、半事务消息

暂不能投递的消息,发送方已经成功地将消息发送到了Broker,但是Broker未收到最终确认指令,此时该消息被标记成“暂不能投递”状态,即不能被消费者看到。处于该种状态下的消息即半事务消息。

3.4、本地事务状态

Producer回调操作执行的结果为本地事务状态,其会发送给TC,而TC会再发送给TM。TM会根据TC发送来的本地事务状态来决定全局事务确认指令。

  1. // 描述本地事务执行状态
  2. public enum LocalTransactionState {
  3. COMMIT_MESSAGE, // 本地事务执行成功
  4. ROLLBACK_MESSAGE, // 本地事务执行失败
  5. UNKNOW, // 不确定,表示需要进行回查以确定本地事务的执行结果
  6. }

3.5、消息回查

消息回查,即重新查询本地事务的执行状态。本例就是重新到DB中查看预扣款操作是否执行成功。

注意,消息回查不是重新执行回调操作。回调操作是进行预扣款操作,而消息回查则是查看预扣款操作执行的结果。

引发消息回查的原因最常见的有两个:

1)回调操作返回UNKNWON

2)TC没有接收到TM的最终全局事务确认指令

3.5、RocketMQ中的消息回查设置

关于消息回查,有三个常见的属性设置。它们都在broker加载的配置文件中设置,例如:

  • transactionTimeout=20,指定TM在20秒内应将最终确认状态发送给TC,否则引发消息回查。默认为60秒
  • transactionCheckMax=5,指定最多回查5次,超过后将丢弃消息并记录错误日志。默认15次。
  • transactionCheckInterval=10,指定设置的多次消息回查的时间间隔为10秒。默认为60秒

4、XA模式三剑客

4.1、XA协议

XA(Unix Transaction)是一种分布式事务解决方案,一种分布式事务处理模式,是基于XA协议的。XA协议由Tuxedo(Transaction for Unix has been Extended for Distributed Operation,分布式操作扩展之后的Unix事务系统)首先提出的,并交给X/Open组织,作为资源管理器与事务管理器的接口标准。

XA模式中有三个重要组件:TC、TM、RM。

4.2、TC

Transaction Coordinator,事务协调者。维护全局和分支事务的状态,驱动全局事务提交或回滚。

RocketMQ中Broker充当着TC。

4.3、TM

Transaction Manager,事务管理器。定义全局事务的范围:开始全局事务、提交或回滚全局事务。它实际是全局事务的发起者。

RocketMQ中事务消息的Producer充当着TM。

4.4、RM

Resource Manager,资源管理器。管理分支事务处理的资源,与TC交谈以注册分支事务和报告分支事务的状态,并驱动分支事务提交或回滚。

RocketMQ中事务消息的Producer及Broker均是RM。

5、XA模式架构

XA模式是一个典型的2PC,其执行原理如下:

  1. TM向TC发起指令,开启一个全局事务。

  2. 根据业务要求,各个RM会逐个向TC注册分支事务,然后TC会逐个向RM发出预执行指令。

  3. 各个RM在接收到指令后会在进行本地事务预执行。

  4. RM将预执行结果Report给TC。当然,这个结果可能是成功,也可能是失败。

  5. TC在接收到各个RM的Report后会将汇总结果上报给TM,根据汇总结果TM会向TC发出确认指令。

    • 若所有结果都是成功响应,则向TC发送Global Commit指令。
    • 只要有结果是失败响应,则向TC发送Global Rollback指令。
  6. TC在接收到指令后再次向RM发送确认指令。

    事务消息方案并不是一个典型的XA模式。因为XA模式中的分支事务是异步的,而事务消息方案中的消息预提交与预扣款操作间是同步的。

6、注意

  • 事务消息不支持延时消息
  • 对于事务消息要做好幂等性检查,因为事务消息可能不止一次被消费(因为存在回滚后再提交的情况)

7、代码举例

7.1、定义工行事务监听器

  1. import org.apache.commons.lang3.StringUtils;
  2. import org.apache.rocketmq.client.producer.LocalTransactionState;
  3. import org.apache.rocketmq.client.producer.TransactionListener;
  4. import org.apache.rocketmq.common.message.Message;
  5. import org.apache.rocketmq.common.message.MessageExt;
  6. public class ICBCTransactionListener implements TransactionListener {
  7. // 回调操作方法
  8. // 消息预提交成功就会触发该方法的执行,用于完成本地事务
  9. @Override
  10. public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
  11. System.out.println("预提交消息成功:" + msg);
  12. // 假设接收到TAGA的消息就表示扣款操作成功,TAGB的消息表示扣款失败,
  13. // TAGC表示扣款结果不清楚,需要执行消息回查
  14. if (StringUtils.equals("TAGA", msg.getTags())) {
  15. return LocalTransactionState.COMMIT_MESSAGE;
  16. } else if (StringUtils.equals("TAGB", msg.getTags())) {
  17. return LocalTransactionState.ROLLBACK_MESSAGE;
  18. } else if (StringUtils.equals("TAGC", msg.getTags())) {
  19. return LocalTransactionState.UNKNOW;
  20. }
  21. return LocalTransactionState.UNKNOW;
  22. }
  23. // 消息回查方法
  24. // 引发消息回查的原因最常见的有两个:
  25. // 1)回调操作返回UNKNWON
  26. // 2)TC没有接收到TM的最终全局事务确认指令
  27. @Override
  28. public LocalTransactionState checkLocalTransaction(MessageExt msg) {
  29. System.out.println("执行消息回查" + msg.getTags());
  30. return LocalTransactionState.COMMIT_MESSAGE;
  31. }
  32. }

7.2、定义事物消息生产者

  1. import org.apache.rocketmq.client.producer.SendResult;
  2. import org.apache.rocketmq.client.producer.TransactionMQProducer;
  3. import org.apache.rocketmq.common.message.Message;
  4. import java.util.concurrent.*;
  5. public class TransactionProducer {
  6. public static void main(String[] args) throws Exception {
  7. TransactionMQProducer producer = new TransactionMQProducer("tpg");
  8. producer.setNamesrvAddr("192.168.216.128:9876");
  9. /**
  10. * 定义一个线程池
  11. * @param corePoolSize 线程池中核心线程数量
  12. * @param maximumPoolSize 线程池中最多线程数
  13. * @param keepAliveTime 这是一个时间,当线程池中线程数量大于核心线程数量是多余空闲线程的存活时长
  14. * @param unit 时间单位
  15. * @param workQueue 临时存放任务的队列,其参数就是队列的长度
  16. * @param threadFactory 线程工厂
  17. */
  18. ExecutorService executorService = new ThreadPoolExecutor(
  19. 2,
  20. 5,
  21. 100,
  22. TimeUnit.SECONDS,
  23. new ArrayBlockingQueue<Runnable>(2000),
  24. new ThreadFactory() {
  25. @Override
  26. public Thread newThread(Runnable r) {
  27. Thread thread = new Thread(r);
  28. thread.setName("client-transaction-msg-check-thread");
  29. return thread;
  30. }
  31. });
  32. // 为生产者指定一个线程池
  33. producer.setExecutorService(executorService);
  34. // 为生产者添加事务监听器
  35. producer.setTransactionListener(new ICBCTransactionListener());
  36. producer.start();
  37. String[] tags = {"TAGA", "TAGB", "TAGC"};
  38. for (int i = 0; i < 3; i++) {
  39. byte[] body = ("Hi," + i).getBytes();
  40. Message msg = new Message("TTopic", tags[i], body);
  41. // 发送事务消息
  42. // 第二个参数用于指定在执行本地事务时要使用的业务参数
  43. SendResult sendResult = producer.sendMessageInTransaction(msg, null);
  44. System.out.println("发送结果为:" + sendResult.getSendStatus());
  45. }
  46. }
  47. }

7.3、定义消费者

直接使用普通消息的SomeConsumer作为消费者即可

  1. import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
  2. import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
  3. import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
  4. import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
  5. import org.apache.rocketmq.client.exception.MQClientException;
  6. import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
  7. import org.apache.rocketmq.common.message.MessageExt;
  8. import java.util.List;
  9. public class SomeConsumer {
  10. public static void main(String[] args) throws MQClientException {
  11. // 定义一个pull消费者
  12. // DefaultLitePullConsumer consumer = new DefaultLitePullConsumer("cg");
  13. // 定义一个push消费者
  14. DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("cg");
  15. // 指定nameServer
  16. consumer.setNamesrvAddr("rocketmqOS:9876");
  17. // 指定从第一条消息开始消费
  18. consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET
  19. );
  20. // 指定消费topic与tag
  21. consumer.subscribe("TTopic", "*");
  22. // 指定采用“广播模式”进行消费,默认为“集群模式”
  23. // consumer.setMessageModel(MessageModel.BROADCASTING);
  24. // 注册消息监听器
  25. consumer.registerMessageListener(new MessageListenerConcurrently() {
  26. // 一旦broker中有了其订阅的消息就会触发该方法的执行,其返回值为当前consumer消费的状态
  27. @Override
  28. public ConsumeConcurrentlyStatus
  29. consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
  30. // 逐条消费消息
  31. for (MessageExt msg : msgs) {
  32. System.out.println(msg);
  33. }
  34. //返回消费状态:消费成功
  35. return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
  36. }
  37. });
  38. // 开启消费者消费
  39. consumer.start();
  40. System.out.println("Consumer Started");
  41. }
  42. }

继续收门徒,亲手带,月薪 4W 以下的可以来找我

最新资料

  1. 尚硅谷 Java 学科全套教程(总 207.77GB)
  2. 2021 最新版 Java 微服务学习线路图 + 视频
  3. 阿里技术大佬整理的《Spring 学习笔记.pdf》
  4. 阿里大佬的《MySQL 学习笔记高清.pdf》
  5. 2021 版 java 高并发常见面试题汇总.pdf
  6. Idea 快捷键大全.pdf