【RocketMQ 生产者和消费者】- 事务消息的使用

本文章基于 RocketMQ 4.9.31. 前言【RocketMQ】- 源码系列目录【RocketMQ 生产者消费者】- 同步、异步、单向发送消费消息【RocketMQ 生产者和消费者】- 消费者启动源码【RocketMQ 生产者和消费者】- 消费者重平衡1【RocketMQ 生产者和消费者】- 消费者重平衡2- 分配策略【RocketMQ 生产者和消费者】- 消费者重平衡3- 消费者 ID 对负载均衡的影响【RocketMQ 生产者和消费者】- 消费者的订阅关系一致性【RocketMQ 生产者和消费者】- 消费者发起消息拉取请求 PullMessageService【RocketMQ 生产者和消费者】- broker 是如何处理消费者消息拉取的 Netty 请求的【RocketMQ 生产者和消费者】- broker 处理消息拉取请求【RocketMQ 生产者和消费者】- 消费者处理消息拉取结果【RocketMQ 生产者和消费者】- 消费者处理消息拉取结果【RocketMQ 生产者和消费者】- ConsumeMessageConcurrentlyService 并发消费消息【RocketMQ 生产者和消费者】- ConsumeMessageOrderlyService 顺序消费消息【RocketMQ 生产者和消费者】- sendMessageBack 发送重试消息【RocketMQ 生产者和消费者】- 延时消息的使用【RocketMQ 生产者和消费者】- 延时消息原理解析-ScheduleMessageService上两篇文章我们分析了延时消息的基本使用和原理这篇文章就来看下事务消息的基本使用。2. 事务消息首先介绍下什么是事务消息也可以看官网的介绍事务消息发送。事务消息可以用在消息队列中确保消息传递和业务操作原子性、一致性尤其适用于分布式系统环境特别是消息上下游服务相互依赖的场景。比如下面的两个场景订单支付比如上游支付订单之后扣减本地库存更新订单状态接着通知下游服务比如物流系统做后续的处理这种情况下可以先发送一条事务消息然后执行本地事务逻辑扣减库存更新订单状态做完之后根据本地事务消息的结果决定这条消息能不能被下游的服务看到如果本地事务执行失败那么下游也不需要继续处理。金融系统中的转账业务转账同时扣减转出账户余额和增加转入账户余额。用事务消息在发送转账成功消息之前执行本地账户余额更新事务确保消息发送与账户余额变更操作的一致性。上面说的本地事务可以是操作数据库、操作缓存等等反正就是一个方法根据业务来处理下面就是大致的流程这个图也是直接用官方的图了。上面涉及到几个名词下面解释下半事务消息半事务消息是指暂不能投递的消息当生产者发送半事务消息到 Broker后Broker 收到消息会将其先存储起来但不会马上将该消息投递给消费者此时消费者对这条消息不可见。只有生产者对该消息进行二次确认即提交或回滚操作后Broker 才会根据确认结果进行相应处理。本地事务本地事务就是本地的逻辑当生产者执行完本地事务之后根据执行结果返回对应的状态码broker 会根据对应结果来处理这条半事务消息。本地事务有三种状态UNKNOW表示中间状态代表需要通过事务回查来确定最终的执行结果COMMIT_MESSAGE表示本地事务执行成功此消息可以被消费者消费ROLLBACK_MESSAGE表示事务回滚这条消息会被删掉不会被消费者消费。事务回查如果由于网络闪断、生产者应用重启等原因导致某条事务消息的二次确认丢失Broker 端会通过扫描发现某条消息长期处于“半事务消息”时需要主动向消息生产者询问该消息的最终状态Commit 或是 Rollback如果业务阻塞也可以通过这种方式给一个兜底的结果在 broker.conf 可以配置transactionCheckInterval回查的时间间隔根据自己实际业务决定。下面把整个流程串起来生产者将半事务消息发送至 broker。broker 将这条消息持久化这条消息属于半事务消息还不可被消费者消费。生产者通过executeLocalTransaction开始执行本地事务逻辑。本地事务执行完成后生产者向 broker 提交二次确认结果Commit 或是 Rollback服务端走不同的处理逻辑二次确认结果为COMMIT_MESSAGE服务端将半事务消息标记为可投递并投递给消费者。二次确认结果为ROLLBACK_MESSAGE服务端将回滚事务不会将半事务消息投递给消费者。二次确认结果为UNKNOW服务端将回滚事务不会将半事务消息投递给消费者。在断网或者是生产者应用重启的特殊情况下broker 迟迟收不到事务的二次提交结果又或者如果事务提交结果给的是UNKNOW那么服务端会对生产者发起事务回查。为了避免单个消息被检查太多次而导致半队列消息累积默认将单个消息的检查次数限制为 15 次但是用户可以通过 Broker 配置文件的transactionCheckMax参数来修改检查次数。如果回查次数到达上限了那么 broker 就丢掉这条消息然后在默认情况下同时打印错误日志。不过我们可以通过重写AbstractTransactionalMessageCheckListener、 类来修改回查上限之后要做什么。生产者收到消息回查后通过checkLocalTransaction检查对应消息的本地事务执行的最终结果。根据检查出来的结果进行二次提交这时候又回到了 4 的逻辑。3. 示例下面看下事务消息的使用首先在 broker 配置下消息回查的时间。transactionCheckInterval20000然后设置下生产者生产者一共发送 10 条消息下标从 0 开始计算然后每一条消息发送的 tag 不一样。publicclassTransactionProducer{publicstaticvoidmain(String[]args)throwsMQClientException,InterruptedException{TransactionListenertransactionListenernewTransactionListenerImpl();// 事务消息生产者TransactionMQProducerproducernewTransactionMQProducer(transactionMQProducer);// 执行本地回查的线程池ExecutorServiceexecutorServicenewThreadPoolExecutor(2,5,100,TimeUnit.SECONDS,newArrayBlockingQueueRunnable(2000),newThreadFactory(){OverridepublicThreadnewThread(Runnabler){ThreadthreadnewThread(r);thread.setName(client-transaction-msg-check-thread);returnthread;}});producer.setNamesrvAddr(localhost:9876);producer.setExecutorService(executorService);// 设置事务监听器producer.setTransactionListener(transactionListener);// 设置本地事务回查时间间隔producer.start();String[]tagsnewString[]{TagA,TagB,TagC,TagD,TagE};for(inti0;i10;i){try{// 消息发送MessagemsgnewMessage(TopicTest1234,tags[i%tags.length],KEYi,(Hello RocketMQ i).getBytes(RemotingHelper.DEFAULT_CHARSET));SendResultsendResultproducer.sendMessageInTransaction(msg,null);System.out.printf(now() %s%n,sendResult);Thread.sleep(10);}catch(MQClientException|UnsupportedEncodingExceptione){e.printStackTrace();}}for(inti0;i100000;i){Thread.sleep(1000);}producer.shutdown();}}接下来初始化事务监听器TransactionListenerImpl实现 TransactionListener 接口在里面定义executeLocalTransaction本地事务执行的逻辑以及checkLocalTransaction本地事务回查的逻辑。publicclassTransactionListenerImplimplementsTransactionListener{privateAtomicIntegertransactionIndexnewAtomicInteger(0);privateConcurrentHashMapString,IntegerlocalTransnewConcurrentHashMap();/** * 执行本地事务 * param msg Half(prepare) message * param arg Custom business parameter * return */OverridepublicLocalTransactionStateexecuteLocalTransaction(Messagemsg,Objectarg){// 执行本地事务, 根据不同下标在下面的 checkLocalTransaction 方法走不同的逻辑intvaluetransactionIndex.getAndIncrement();intstatusvalue%3;localTrans.put(msg.getTransactionId(),status);// 全部返回 UNKNOW, 意味者要通过事务回查 checkLocalTransaction 方法去检查本地事务的提交结果returnLocalTransactionState.UNKNOW;}/** * 本地事务回查 * param msg Check message * return */OverridepublicLocalTransactionStatecheckLocalTransaction(MessageExtmsg){System.out.println(now() checkLocalTransaction 执行本地事务回查, 当前消息事务 ID: msg.getTransactionId());IntegerstatuslocalTrans.get(msg.getTransactionId());if(null!status){switch(status){case0:returnLocalTransactionState.UNKNOW;case1:returnLocalTransactionState.COMMIT_MESSAGE;case2:returnLocalTransactionState.ROLLBACK_MESSAGE;default:returnLocalTransactionState.COMMIT_MESSAGE;}}returnLocalTransactionState.COMMIT_MESSAGE;}}executeLocalTransaction执行本地事务的时候会先通过 transactionIndex 1 算出这条消息是哪个下标然后在事务回查checkLocalTransaction的时候将所有消息分为三类来处理。0LocalTransactionState.UNKNOW1LocalTransactionState.COMMIT_MESSAGE2LocalTransactionState.ROLLBACK_MESSAGE大致流程就是执行本地事务的时候LocalTransactionState.UNKNOWbroker 会对每一条消息都进行回查然后在回查里面会根据不同的下标来返回不同的结果结合 executeLocalTransaction 方法可以大概看出消息0、3、6、9会不断回查1、4、7回查之后发现是COMMIT_MESSAGE消息提交消费者可以消费到这几条消息2、5、8回查之后是ROLLBACK_MESSAGE这几条消息会被丢掉后续回查也不会对这几条消息发起回查了。然后是消费者消费者比较简单就是订阅TopicTest1234下面的所有 tags 进行消费。publicclassTransactionConsumer{publicstaticvoidmain(String[]args)throwsInterruptedException,MQClientException{DefaultMQPushConsumerconsumernewDefaultMQPushConsumer(testGroupConsumer);consumer.setNamesrvAddr(localhost:9876);consumer.setMessageModel(MessageModel.CLUSTERING);consumer.subscribe(TopicTest1234,*);consumer.registerMessageListener(newMessageListenerConcurrently(){OverridepublicConsumeConcurrentlyStatusconsumeMessage(ListMessageExtmsgs,ConsumeConcurrentlyContextcontext){for(inti0;imsgs.size();i){MessageExtmsgmsgs.get(i);System.out.printf(%s, %s Receive New Messages: %s %n,now(),Thread.currentThread().getName(),newString(msg.getBody(),StandardCharsets.UTF_8));context.setAckIndex(i);}returnConsumeConcurrentlyStatus.CONSUME_SUCCESS;}});consumer.start();System.out.printf(Consumer Started.%n);}publicstaticStringnow(){returnnewSimpleDateFormat(yyyy-MM-dd HH:mm:ss:SSS).format(newDate());}}输出结果如下首先是消费者可以看到确实只消费了 1、4、7 三条消息。然后是生产者生产者发送消息之后由于本地事务执行都是返回的 LocalTransactionState.UNKNOW所以经过一定时间后会消息回查这里为什么不是 20s后面源码会分析。经过事务回查之后发现2、5、8返回了ROLLBACK_MESSAGE于是后续不会再回查这几条消息而0、3、6、9由于还是返回UNKNOW于是就一直不断回查但是回查次数有上限所以也不会一直回查不过时间太长了就不贴出所有输出。4. 使用限制这里的限制也是代码给的文档里面标注出来的下面贴出来事务消息不支持延时消息和批量消息。提交给用户的目标主题消息可能会失败它的高可用性通过 RocketMQ 本身的高可用性机制来保证如果希望确保事务消息不丢失、并且事务完整性得到保证建议使用同步的双重写入机制。事务消息的生产者 ID 不能与其他类型消息的生产者 ID 共享。与其他类型的消息不同事务消息允许反向查询、MQ 服务器能通过它们的生产者 ID 查询到消费者。5. 小结好了到这里事务消息的使用和大致流程已经介绍完下一篇文章就要进入源码分析环节。如有错误欢迎指出