RocketMQ源码解析十三（事务消息流程）

港控/mmm° 2022-09-12 07:47 96阅读 0赞

RocketMQ版本4.6.0，记录自己看源码的过程

代码示例

  1. @Configuration
  2. public class UserProducerConfiguration { 
  4.     @Autowired
  5.     private UserService userService;
  7.     @Bean
  8.     public TransactionMQProducer producer() throws MQClientException { 
  9.         TransactionMQProducer producer = new TransactionMQProducer("user_producer_group");
  10.         producer.setNamesrvAddr("localhost:9876");
  11.         producer.setTransactionListener(new TransactionListener() { 
  12.             @Override
  13.             public LocalTransactionState executeLocalTransaction(Message message, Object o) { 
  14.                 // 根据本地事务执行成与否判断 事务消息是否需要commit与 rollback
  15.                 ObjectMapper objectMapper = new ObjectMapper();
  16.                 LocalTransactionState state;
  17.                 try { 
  18.                     User user = objectMapper.readValue(message.getBody(), User.class);
  19.                     userService.insert(user);
  20.                     state = LocalTransactionState.COMMIT_MESSAGE;
  21.                 } catch (Exception e) { 
  22.                     e.printStackTrace();
  23.                     state = LocalTransactionState.ROLLBACK_MESSAGE;
  24.                 }
  25.                 return state;
  26.             }
  28.             @Override
  29.             public LocalTransactionState checkLocalTransaction(MessageExt messageExt) { 
  30.                 ObjectMapper objectMapper = new ObjectMapper();
  31.                 LocalTransactionState state = LocalTransactionState.UNKNOW;
  32.                 User user;
  33.                 try { 
  34.                     user = objectMapper.readValue(messageExt.getBody(), User.class);
  35.                 } catch (IOException e) { 
  36.                     e.printStackTrace();
  37.                     return state;
  38.                 }
  39.                 try { 
  40.                     // 查询数据库是否有这条记录
  41.                     user = userService.select(user.getId());
  43.                     if (user != null) { 
  44.                         state = LocalTransactionState.COMMIT_MESSAGE;
  45.                     } else { 
  46.                         state = LocalTransactionState.ROLLBACK_MESSAGE;
  47.                     }
  48.                 } catch (Exception e) { 
  49.                     e.printStackTrace();
  50.                 }
  51.                 return state;
  52.             }
  53.         });
  54.         producer.start();
  55.         return producer;
  56.     }
  57. }
  59. @Service
  60. public class UserServiceImpl implements UserService { 
  62.     @Autowired
  63.     private UserMapper userMapper;
  64.     @Autowired
  65.     private TransactionMQProducer producer;
  67.     @Override
  68.     public void insert() throws Exception { 
  69.         User user = User.builder().id(UUID.randomUUID().toString()).name("test").build();
  70.         Message msg = new Message("userTopic", "tagA", user.getId(),
  71.                 new ObjectMapper().writeValueAsString(user).getBytes(RemotingHelper.DEFAULT_CHARSET));
  72.         // 发送事务消息
  73.         SendResult sendResult = producer.sendMessageInTransaction(msg, null);
  74.     }
  76.     @Override
  77.     public boolean insert(User user) { 
  78.         return userMapper.insert(user);
  79.     }
  81.     @Override
  82.     public User select(String id) { 
  83.         return userMapper.select(id);
  84.     }
  86.     public boolean update(User user) { 
  87.         return userMapper.update(user);
  88.     }
  89. }

事务消息设计思想

在执行本地数据落库之前，会由生产者先发送一条半消息（prepare，这条消息对消费者是不可见的），发送成功表示MQ是可用的，然后会回调生产者的事件监听程序，执行本地事务使数据落库，之后根据本地事务执行情况向MQ发送commit或rollback，提交后，消费者就可以消费到消息了。
如果在commit或rollback时失败了，MQ服务器会开启一个定时任务，每隔60s会回查那些prepare状态下的消息，检查本地事务是否成功，根据状态再次发送commit或rollback，保证数据的一致性。
在这里插入图片描述

事务消息发送流程

DefaultMQProducerImpl

  1. /** * 发送事务消息 */
  2. public TransactionSendResult sendMessageInTransaction(final Message msg,
  3.     final LocalTransactionExecuter localTransactionExecuter, final Object arg)
  4.     throws MQClientException { 
  5.     TransactionListener transactionListener = getCheckListener();
  6.     if (null == localTransactionExecuter && null == transactionListener) { 
  7.         throw new MQClientException("tranExecutor is null", null);
  8.     }
  10.     // 忽略延时
  11.     if (msg.getDelayTimeLevel() != 0) { 
  12.         MessageAccessor.clearProperty(msg, MessageConst.PROPERTY_DELAY_TIME_LEVEL);
  13.     }
  15.     Validators.checkMessage(msg, this.defaultMQProducer);
  17.     SendResult sendResult = null;
  18.     // 事务消息属性设为true
  19.     MessageAccessor.putProperty(msg, MessageConst.PROPERTY_TRANSACTION_PREPARED, "true");
  20.     MessageAccessor.putProperty(msg, MessageConst.PROPERTY_PRODUCER_GROUP, this.defaultMQProducer.getProducerGroup());
  21.     try { 
  22.         // 同步发送，跟发送普通消息一样
  23.         sendResult = this.send(msg);
  24.     } catch (Exception e) { 
  25.         throw new MQClientException("send message Exception", e);
  26.     }
  28.     LocalTransactionState localTransactionState = LocalTransactionState.UNKNOW;
  29.     Throwable localException = null;
  30.     // 处理发送结果
  31.     switch (sendResult.getSendStatus()) { 
  32.         // 发送成功
  33.         case SEND_OK: { 
  34.             try { 
  35.                 if (sendResult.getTransactionId() != null) { 
  36.                     msg.putUserProperty("__transactionId__", sendResult.getTransactionId());
  37.                 }
  38.                 String transactionId = msg.getProperty(MessageConst.PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX);
  39.                 if (null != transactionId && !"".equals(transactionId)) { 
  40.                     msg.setTransactionId(transactionId);
  41.                 }
  42.                 if (null != localTransactionExecuter) { 
  43.                     localTransactionState = localTransactionExecuter.executeLocalTransactionBranch(msg, arg);
  44.                 } else if (transactionListener != null) { 
  45.                     log.debug("Used new transaction API");
  46.                     // 执行本地事务
  47.                     localTransactionState = transactionListener.executeLocalTransaction(msg, arg);
  48.                 }
  49.                 if (null == localTransactionState) { 
  50.                     localTransactionState = LocalTransactionState.UNKNOW;
  51.                 }
  53.                 if (localTransactionState != LocalTransactionState.COMMIT_MESSAGE) { 
  54.                     log.info("executeLocalTransactionBranch return {}", localTransactionState);
  55.                     log.info(msg.toString());
  56.                 }
  57.             } catch (Throwable e) { 
  58.                 log.info("executeLocalTransactionBranch exception", e);
  59.                 log.info(msg.toString());
  60.                 localException = e;
  61.             }
  62.         }
  63.         break;
  64.         // 刷盘或同步slave失败则需要回滚，不会执行本地事务
  65.         case FLUSH_DISK_TIMEOUT:
  66.         case FLUSH_SLAVE_TIMEOUT:
  67.         case SLAVE_NOT_AVAILABLE:
  68.             localTransactionState = LocalTransactionState.ROLLBACK_MESSAGE;
  69.             break;
  70.         default:
  71.             break;
  72.     }
  74.     try { 
  75.         // 结束事务，根据本地事务决定消息事务是提交还是回滚，单向请求
  76.         this.endTransaction(sendResult, localTransactionState, localException);
  77.     } catch (Exception e) { 
  78.         log.warn("local transaction execute " + localTransactionState + ", but end broker transaction failed", e);
  79.     }
  81.     TransactionSendResult transactionSendResult = new TransactionSendResult();
  82.     transactionSendResult.setSendStatus(sendResult.getSendStatus());
  83.     transactionSendResult.setMessageQueue(sendResult.getMessageQueue());
  84.     transactionSendResult.setMsgId(sendResult.getMsgId());
  85.     transactionSendResult.setQueueOffset(sendResult.getQueueOffset());
  86.     transactionSendResult.setTransactionId(sendResult.getTransactionId());
  87.     transactionSendResult.setLocalTransactionState(localTransactionState);
  88.     return transactionSendResult;
  89. }

可以看到发送的逻辑也是用发送普通消息的逻辑，只是在发送之前会先清除延时以及设置事务消息属性。
然后根据发送的结果进行分别处理：
1、如果发送成功，则执行监听器中的本地事务方法executeLocalTransaction，返回本地事务状态。

按我的理解，如果executeLocalTransaction方法，或者说执行发送事务消息的方法与本地事务（service中的方法）处于同一个事务内，则不管本地事务成功与否，都应该返回UNKNOW，交由broker回查来决定提交或回滚。因为同一个事务内，如果本地事务执行成功，但还未提交，向broker发送提交确认完后该机器宕机了，此时本地事务回滚，但发出去的确认消息无法收回，消费者会收到消息，导致数据不一致。所以发送UNKNOW，由broker来回查，没问题，消息确认，有问题，消息回滚。
如果没有处于同一个事务内，则直接根据本地事务的执行情况返回COMMIT_MESSAGE或ROLLBACK_MESSAGE

2、如果发送失败，则将本地事务状态设置为ROLLBACK_MESSAGE。
最后结束事务，根据上面的本地事务决定消息事务是提交还是回滚，单向请求。

接着看下broker端如果处理prepare状态的半消息。

Broker处理事务消息也是用SendMessageProcessor处理器处理。

  1. /** * 处理发送消息请求 */
  2. private RemotingCommand sendMessage(final ChannelHandlerContext ctx,
  3.                                     final RemotingCommand request,
  4.                                     final SendMessageContext sendMessageContext,
  5.                                     final SendMessageRequestHeader requestHeader) throws RemotingCommandException { 
  7.     // 省略。。。
  9.     // 校验是否不允许发送事务消息
  10.     if (traFlag != null && Boolean.parseBoolean(traFlag)) { 
  11.         // 不允许发送事务消息
  12.         if (this.brokerController.getBrokerConfig().isRejectTransactionMessage()) { 
  13.             response.setCode(ResponseCode.NO_PERMISSION);
  14.             response.setRemark(
  15.                 "the broker[" + this.brokerController.getBrokerConfig().getBrokerIP1()
  16.                     + "] sending transaction message is forbidden");
  17.             return response;
  18.         }
  19.         // 处理事务消息
  20.         putMessageResult = this.brokerController.getTransactionalMessageService().prepareMessage(msgInner);
  21.     } else { 
  22.         // 使用MessageStore组件将消息存储在本地文件，只存储CommitLog文件，
  23.         // ConsumerQueue文件和IndexFile文件会由后台线程异步存储
  24.         putMessageResult = this.brokerController.getMessageStore().putMessage(msgInner);
  25.     }
  27.     // 处理消息存储结果
  28.     return handlePutMessageResult(putMessageResult, response, request, msgInner, responseHeader, sendMessageContext, ctx, queueIdInt);
  30. }

broker会判断是否是事务消息，事务消息则由事务消息服务处理。
TransactionalMessageServiceImpl

  1. @Override
  2. public PutMessageResult prepareMessage(MessageExtBrokerInner messageInner) { 
  3.     return transactionalMessageBridge.putHalfMessage(messageInner);
  4. }

TransactionalMessageBridge

  1. public PutMessageResult putHalfMessage(MessageExtBrokerInner messageInner) { 
  2.     // 将消息存入事务消息专属的topic中，存储操作跟普通消息一样
  3.     return store.putMessage(parseHalfMessageInner(messageInner));
  4. }
  6. /** * 转换事务消息，将原topic和queueId放到属性中，替换为topic为RMQ_SYS_TRANS_HALF_TOPIC， * queueId为0 */
  7. private MessageExtBrokerInner parseHalfMessageInner(MessageExtBrokerInner msgInner) { 
  8.     MessageAccessor.putProperty(msgInner, MessageConst.PROPERTY_REAL_TOPIC, msgInner.getTopic());
  9.     MessageAccessor.putProperty(msgInner, MessageConst.PROPERTY_REAL_QUEUE_ID,
  10.         String.valueOf(msgInner.getQueueId()));
  11.     msgInner.setSysFlag(
  12.         MessageSysFlag.resetTransactionValue(msgInner.getSysFlag(), MessageSysFlag.TRANSACTION_NOT_TYPE));
  13.     msgInner.setTopic(TransactionalMessageUtil.buildHalfTopic());
  14.     msgInner.setQueueId(0);
  15.     msgInner.setPropertiesString(MessageDecoder.messageProperties2String(msgInner.getProperties()));
  16.     return msgInner;
  17. }

这里是事务消息与非事务消息发送流程的主要区别，如果是事务消息则备份消息的原主题与原消息消费队列，然后将主题变更为RMQ_ SYS_ TRANS_ HALF_ TOPIC，消费队列变更为0，然后消息按照普通消息存储在commitlog文件进而转发到RMQ_SYS_TRANS_HALF_ TOPIC主题对应的消息消费队列。也就是说，事务消息在未提交之前并不会存人消息原有主题，自然也不会被消费者消费。——摘自《RocketMQ技术内幕》

提交或回滚消息

上面说到，半消息发送完成后，会根据发送状态执行本地事务，再根据本地事务状态结束事务（提交或回滚消息）。
DefaultMQProducerImpl

  1. public void endTransaction(
  2.     final SendResult sendResult,
  3.     final LocalTransactionState localTransactionState,
  4.     final Throwable localException) throws RemotingException, MQBrokerException, InterruptedException, UnknownHostException { 
  5.     final MessageId id;
  6.     if (sendResult.getOffsetMsgId() != null) { 
  7.         id = MessageDecoder.decodeMessageId(sendResult.getOffsetMsgId());
  8.     } else { 
  9.         id = MessageDecoder.decodeMessageId(sendResult.getMsgId());
  10.     }
  11.     String transactionId = sendResult.getTransactionId();
  12.     final String brokerAddr = this.mQClientFactory.findBrokerAddressInPublish(sendResult.getMessageQueue().getBrokerName());
  13.     EndTransactionRequestHeader requestHeader = new EndTransactionRequestHeader();
  14.     requestHeader.setTransactionId(transactionId);
  15.     requestHeader.setCommitLogOffset(id.getOffset());
  16.     // 根据本地状态设置提交或回滚
  17.     switch (localTransactionState) { 
  18.         case COMMIT_MESSAGE:
  19.             requestHeader.setCommitOrRollback(MessageSysFlag.TRANSACTION_COMMIT_TYPE);
  20.             break;
  21.         case ROLLBACK_MESSAGE:
  22.             requestHeader.setCommitOrRollback(MessageSysFlag.TRANSACTION_ROLLBACK_TYPE);
  23.             break;
  24.         case UNKNOW:
  25.             requestHeader.setCommitOrRollback(MessageSysFlag.TRANSACTION_NOT_TYPE);
  26.             break;
  27.         default:
  28.             break;
  29.     }
  31.     requestHeader.setProducerGroup(this.defaultMQProducer.getProducerGroup());
  32.     requestHeader.setTranStateTableOffset(sendResult.getQueueOffset());
  33.     requestHeader.setMsgId(sendResult.getMsgId());
  34.     String remark = localException != null ? ("executeLocalTransactionBranch exception: " + localException.toString()) : null;
  35.     // 提交或回滚或不处理事务消息
  36.     this.mQClientFactory.getMQClientAPIImpl().endTransactionOneway(brokerAddr, requestHeader, remark,
  37.         this.defaultMQProducer.getSendMsgTimeout());
  38. }

Broker端处理结束事务的处理器是EndTransactionProcessor

  1. @Override
  2. public RemotingCommand processRequest(ChannelHandlerContext ctx, RemotingCommand request) throws
  3.     RemotingCommandException { 
  4.     final RemotingCommand response = RemotingCommand.createResponseCommand(null);
  5.     final EndTransactionRequestHeader requestHeader =
  6.         (EndTransactionRequestHeader)request.decodeCommandCustomHeader(EndTransactionRequestHeader.class);
  7.     LOGGER.debug("Transaction request:{}", requestHeader);
  8.     if (BrokerRole.SLAVE == brokerController.getMessageStoreConfig().getBrokerRole()) { 
  9.         response.setCode(ResponseCode.SLAVE_NOT_AVAILABLE);
  10.         LOGGER.warn("Message store is slave mode, so end transaction is forbidden. ");
  11.         return response;
  12.     }
  14.     // 省略不重要代码。。
  16.     OperationResult result = new OperationResult();
  17.     // 提交消息
  18.     if (MessageSysFlag.TRANSACTION_COMMIT_TYPE == requestHeader.getCommitOrRollback()) { 
  19.         // 根据requestHeader中的物理偏移量从commitLog中取出prepared状态的消息
  20.         result = this.brokerController.getTransactionalMessageService().commitMessage(requestHeader);
  21.         if (result.getResponseCode() == ResponseCode.SUCCESS) { 
  22.             RemotingCommand res = checkPrepareMessage(result.getPrepareMessage(), requestHeader);
  23.             if (res.getCode() == ResponseCode.SUCCESS) { 
  24.                 // 根据PrepareMessage复制一份消息，topic为原主题
  25.                 MessageExtBrokerInner msgInner = endMessageTransaction(result.getPrepareMessage());
  26.                 msgInner.setSysFlag(MessageSysFlag.resetTransactionValue(msgInner.getSysFlag(), requestHeader.getCommitOrRollback()));
  27.                 msgInner.setQueueOffset(requestHeader.getTranStateTableOffset());
  28.                 msgInner.setPreparedTransactionOffset(requestHeader.getCommitLogOffset());
  29.                 msgInner.setStoreTimestamp(result.getPrepareMessage().getStoreTimestamp());
  30.                 // 将消息投递到原主题队列中
  31.                 RemotingCommand sendResult = sendFinalMessage(msgInner);
  32.                 if (sendResult.getCode() == ResponseCode.SUCCESS) { 
  33.                     // 删除PrepareMessage，表示消息处理过了
  34.                     this.brokerController.getTransactionalMessageService().deletePrepareMessage(result.getPrepareMessage());
  35.                 }
  36.                 return sendResult;
  37.             }
  38.             return res;
  39.         }
  40.     } else if (MessageSysFlag.TRANSACTION_ROLLBACK_TYPE == requestHeader.getCommitOrRollback()) { 
  41.         // 回滚消息，与提交消息的唯一区别是不会将消息转发给原主题
  42.         result = this.brokerController.getTransactionalMessageService().rollbackMessage(requestHeader);
  43.         if (result.getResponseCode() == ResponseCode.SUCCESS) { 
  44.             RemotingCommand res = checkPrepareMessage(result.getPrepareMessage(), requestHeader);
  45.             if (res.getCode() == ResponseCode.SUCCESS) { 
  46.                 // 删除PrepareMessage，表示消息处理过了
  47.                 this.brokerController.getTransactionalMessageService().deletePrepareMessage(result.getPrepareMessage());
  48.             }
  49.             return res;
  50.         }
  51.     }
  52.     response.setCode(result.getResponseCode());
  53.     response.setRemark(result.getResponseRemark());
  54.     return response;
  55. }

删除PrepareMessage，表示消息处理过了

  1. /** * 删除PrepareMessage，其实是将原预处理消息存入一个新的主题RMQ_SYS_TRANS_OP_HALF_TOPIC， * 代表该消息已被处理 */
  2. @Override
  3. public boolean deletePrepareMessage(MessageExt msgExt) { 
  4.     if (this.transactionalMessageBridge.putOpMessage(msgExt, TransactionalMessageUtil.REMOVETAG)) { 
  5.         log.debug("Transaction op message write successfully. messageId={}, queueId={} msgExt:{}", msgExt.getMsgId(), msgExt.getQueueId(), msgExt);
  6.         return true;
  7.     } else { 
  8.         log.error("Transaction op message write failed. messageId is {}, queueId is {}", msgExt.getMsgId(), msgExt.getQueueId());
  9.         return false;
  10.     }
  11. }

回查事务状态

如果执行完本地事务返回本地事务状态为UN_KNOW时，结束事务时将不做任何处理，或者在发送提交或回滚时失败，此时就需要通过事务回查机制定时的去应用程序检查本地的事务状态来决定是否提交或回滚。
MQ通过TransactionalMessageCheckService线程服务定时去检测RMQ_SYS_TRANS_HALF_TOPIC主题中的消息，回查消息的事务状态。

  1. @Override
  2. public void run() { 
  3.     log.info("Start transaction check service thread!");
  4.     // 检测间隔，默认60s
  5.     long checkInterval = brokerController.getBrokerConfig().getTransactionCheckInterval();
  6.     while (!this.isStopped()) { 
  7.         // 最终调用onWaitEnd方法
  8.         this.waitForRunning(checkInterval);
  9.     }
  10.     log.info("End transaction check service thread!");
  11. }
  13. @Override
  14. protected void onWaitEnd() { 
  15.     // 表示事务的过期时间，一个消息的存储时间 + 该值 大于系统当前时间，才对该消息执行事务状态回查
  16.     long timeout = brokerController.getBrokerConfig().getTransactionTimeOut();
  17.     // 最多检测几次，超过检测次数，则会回滚
  18.     int checkMax = brokerController.getBrokerConfig().getTransactionCheckMax();
  19.     long begin = System.currentTimeMillis();
  20.     log.info("Begin to check prepare message, begin time:{}", begin);
  21.     this.brokerController.getTransactionalMessageService().check(timeout, checkMax, this.brokerController.getTransactionalMessageCheckListener());
  22.     log.info("End to check prepare message, consumed time:{}", System.currentTimeMillis() - begin);
  23. }

TransactionalMessageServiceImpl

  1. @Override
  2. public void check(long transactionTimeout, int transactionCheckMax,
  3.     AbstractTransactionalMessageCheckListener listener) { 
  4.     try { 
  5.         String topic = MixAll.RMQ_SYS_TRANS_HALF_TOPIC;
  6.         // 获取事务消息topic下的所有消息队列
  7.         Set<MessageQueue> msgQueues = transactionalMessageBridge.fetchMessageQueues(topic);
  8.         if (msgQueues == null || msgQueues.size() == 0) { 
  9.             log.warn("The queue of topic is empty :" + topic);
  10.             return;
  11.         }
  12.         log.debug("Check topic={}, queues={}", topic, msgQueues);
  13.         // 遍历所有消息队列
  14.         for (MessageQueue messageQueue : msgQueues) { 
  15.             long startTime = System.currentTimeMillis();
  16.             // 根据messageQueue从缓存获取操作队列，如果没有，则创建一个，topic为RMQ_SYS_TRANS_OP_HALF_TOPIC，
  17.             // queueId与messageQueue的queueId一样
  18.             MessageQueue opQueue = getOpQueue(messageQueue);
  19.             // 获取两个队列的offset
  20.             long halfOffset = transactionalMessageBridge.fetchConsumeOffset(messageQueue);
  21.             long opOffset = transactionalMessageBridge.fetchConsumeOffset(opQueue);
  22.             log.info("Before check, the queue={} msgOffset={} opOffset={}", messageQueue, halfOffset, opOffset);
  23.             if (halfOffset < 0 || opOffset < 0) { 
  24.                 log.error("MessageQueue: {} illegal offset read: {}, op offset: {},skip this queue", messageQueue,
  25.                     halfOffset, opOffset);
  26.                 continue;
  27.             }
  29.             List<Long> doneOpOffset = new ArrayList<>();
  30.             HashMap<Long, Long> removeMap = new HashMap<>();
  31.             PullResult pullResult = fillOpRemoveMap(removeMap, opQueue, opOffset, halfOffset, doneOpOffset);
  32.             if (null == pullResult) { 
  33.                 log.error("The queue={} check msgOffset={} with opOffset={} failed, pullResult is null",
  34.                     messageQueue, halfOffset, opOffset);
  35.                 continue;
  36.             }
  37.             // single thread
  38.             int getMessageNullCount = 1;
  39.             long newOffset = halfOffset;
  40.             long i = halfOffset;
  41.             while (true) { 
  42.                 if (System.currentTimeMillis() - startTime > MAX_PROCESS_TIME_LIMIT) { 
  43.                     log.info("Queue={} process time reach max={}", messageQueue, MAX_PROCESS_TIME_LIMIT);
  44.                     break;
  45.                 }
  46.                 if (removeMap.containsKey(i)) { 
  47.                     log.info("Half offset {} has been committed/rolled back", i);
  48.                     Long removedOpOffset = removeMap.remove(i);
  49.                     doneOpOffset.add(removedOpOffset);
  50.                 } else { 
  51.                     GetResult getResult = getHalfMsg(messageQueue, i);
  52.                     MessageExt msgExt = getResult.getMsg();
  53.                     if (msgExt == null) { 
  54.                         if (getMessageNullCount++ > MAX_RETRY_COUNT_WHEN_HALF_NULL) { 
  55.                             break;
  56.                         }
  57.                         if (getResult.getPullResult().getPullStatus() == PullStatus.NO_NEW_MSG) { 
  58.                             log.debug("No new msg, the miss offset={} in={}, continue check={}, pull result={}", i,
  59.                                 messageQueue, getMessageNullCount, getResult.getPullResult());
  60.                             break;
  61.                         } else { 
  62.                             log.info("Illegal offset, the miss offset={} in={}, continue check={}, pull result={}",
  63.                                 i, messageQueue, getMessageNullCount, getResult.getPullResult());
  64.                             i = getResult.getPullResult().getNextBeginOffset();
  65.                             newOffset = i;
  66.                             continue;
  67.                         }
  68.                     }
  70.                     if (needDiscard(msgExt, transactionCheckMax) || needSkip(msgExt)) { 
  71.                         listener.resolveDiscardMsg(msgExt);
  72.                         newOffset = i + 1;
  73.                         i++;
  74.                         continue;
  75.                     }
  76.                     if (msgExt.getStoreTimestamp() >= startTime) { 
  77.                         log.debug("Fresh stored. the miss offset={}, check it later, store={}", i,
  78.                             new Date(msgExt.getStoreTimestamp()));
  79.                         break;
  80.                     }
  82.                     long valueOfCurrentMinusBorn = System.currentTimeMillis() - msgExt.getBornTimestamp();
  83.                     long checkImmunityTime = transactionTimeout;
  84.                     String checkImmunityTimeStr = msgExt.getUserProperty(MessageConst.PROPERTY_CHECK_IMMUNITY_TIME_IN_SECONDS);
  85.                     if (null != checkImmunityTimeStr) { 
  86.                         checkImmunityTime = getImmunityTime(checkImmunityTimeStr, transactionTimeout);
  87.                         if (valueOfCurrentMinusBorn < checkImmunityTime) { 
  88.                             if (checkPrepareQueueOffset(removeMap, doneOpOffset, msgExt)) { 
  89.                                 newOffset = i + 1;
  90.                                 i++;
  91.                                 continue;
  92.                             }
  93.                         }
  94.                     } else { 
  95.                         if ((0 <= valueOfCurrentMinusBorn) && (valueOfCurrentMinusBorn < checkImmunityTime)) { 
  96.                             log.debug("New arrived, the miss offset={}, check it later checkImmunity={}, born={}", i,
  97.                                 checkImmunityTime, new Date(msgExt.getBornTimestamp()));
  98.                             break;
  99.                         }
  100.                     }
  101.                     List<MessageExt> opMsg = pullResult.getMsgFoundList();
  102.                     boolean isNeedCheck = (opMsg == null && valueOfCurrentMinusBorn > checkImmunityTime)
  103.                         || (opMsg != null && (opMsg.get(opMsg.size() - 1).getBornTimestamp() - startTime > transactionTimeout))
  104.                         || (valueOfCurrentMinusBorn <= -1);
  106.                     if (isNeedCheck) { 
  107.                         if (!putBackHalfMsgQueue(msgExt, i)) { 
  108.                             continue;
  109.                         }
  110.                         // 消息回查
  111.                         listener.resolveHalfMsg(msgExt);
  112.                     } else { 
  113.                         pullResult = fillOpRemoveMap(removeMap, opQueue, pullResult.getNextBeginOffset(), halfOffset, doneOpOffset);
  114.                         log.debug("The miss offset:{} in messageQueue:{} need to get more opMsg, result is:{}", i,
  115.                             messageQueue, pullResult);
  116.                         continue;
  117.                     }
  118.                 }
  119.                 newOffset = i + 1;
  120.                 i++;
  121.             }
  122.             if (newOffset != halfOffset) { 
  123.                 transactionalMessageBridge.updateConsumeOffset(messageQueue, newOffset);
  124.             }
  125.             long newOpOffset = calculateOpOffset(doneOpOffset, opOffset);
  126.             if (newOpOffset != opOffset) { 
  127.                 transactionalMessageBridge.updateConsumeOffset(opQueue, newOpOffset);
  128.             }
  129.         }
  130.     } catch (Exception e) { 
  131.         e.printStackTrace();
  132.         log.error("Check error", e);
  133.     }
  135. }

AbstractTransactionalMessageCheckListener

  1. public void resolveHalfMsg(final MessageExt msgExt) { 
  2.     executorService.execute(new Runnable() { 
  3.         @Override
  4.         public void run() { 
  5.             try { 
  6.                 sendCheckMessage(msgExt);
  7.             } catch (Exception e) { 
  8.                 LOGGER.error("Send check message error!", e);
  9.             }
  10.         }
  11.     });
  12. }
  14. public void sendCheckMessage(MessageExt msgExt) throws Exception { 
  15.     CheckTransactionStateRequestHeader checkTransactionStateRequestHeader = new CheckTransactionStateRequestHeader();
  16.     checkTransactionStateRequestHeader.setCommitLogOffset(msgExt.getCommitLogOffset());
  17.     checkTransactionStateRequestHeader.setOffsetMsgId(msgExt.getMsgId());
  18.     checkTransactionStateRequestHeader.setMsgId(msgExt.getUserProperty(MessageConst.PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX));
  19.     checkTransactionStateRequestHeader.setTransactionId(checkTransactionStateRequestHeader.getMsgId());
  20.     checkTransactionStateRequestHeader.setTranStateTableOffset(msgExt.getQueueOffset());
  21.     msgExt.setTopic(msgExt.getUserProperty(MessageConst.PROPERTY_REAL_TOPIC));
  22.     msgExt.setQueueId(Integer.parseInt(msgExt.getUserProperty(MessageConst.PROPERTY_REAL_QUEUE_ID)));
  23.     msgExt.setStoreSize(0);
  24.     String groupId = msgExt.getProperty(MessageConst.PROPERTY_PRODUCER_GROUP);
  25.     Channel channel = brokerController.getProducerManager().getAvaliableChannel(groupId);
  26.     if (channel != null) { 
  27.         // 发送回查请求
  28.         brokerController.getBroker2Client().checkProducerTransactionState(groupId, channel, checkTransactionStateRequestHeader, msgExt);
  29.     } else { 
  30.         LOGGER.warn("Check transaction failed, channel is null. groupId={}", groupId);
  31.     }
  32. }

生产端使用ClientRemotingProcessor处理请求

  1. public RemotingCommand checkTransactionState(ChannelHandlerContext ctx,
  2.     RemotingCommand request) throws RemotingCommandException { 
  3.     final CheckTransactionStateRequestHeader requestHeader =
  4.         (CheckTransactionStateRequestHeader) request.decodeCommandCustomHeader(CheckTransactionStateRequestHeader.class);
  5.     final ByteBuffer byteBuffer = ByteBuffer.wrap(request.getBody());
  6.     final MessageExt messageExt = MessageDecoder.decode(byteBuffer);
  7.     if (messageExt != null) { 
  8.         if (StringUtils.isNotEmpty(this.mqClientFactory.getClientConfig().getNamespace())) { 
  9.             messageExt.setTopic(NamespaceUtil
  10.                 .withoutNamespace(messageExt.getTopic(), this.mqClientFactory.getClientConfig().getNamespace()));
  11.         }
  12.         String transactionId = messageExt.getProperty(MessageConst.PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX);
  13.         if (null != transactionId && !"".equals(transactionId)) { 
  14.             messageExt.setTransactionId(transactionId);
  15.         }
  16.         final String group = messageExt.getProperty(MessageConst.PROPERTY_PRODUCER_GROUP);
  17.         if (group != null) { 
  18.             MQProducerInner producer = this.mqClientFactory.selectProducer(group);
  19.             if (producer != null) { 
  20.                 final String addr = RemotingHelper.parseChannelRemoteAddr(ctx.channel());
  21.                 // 检查本地事务状态
  22.                 producer.checkTransactionState(addr, messageExt, requestHeader);
  23.             } else { 
  24.                 log.debug("checkTransactionState, pick producer by group[{}] failed", group);
  25.             }
  26.         } else { 
  27.             log.warn("checkTransactionState, pick producer group failed");
  28.         }
  29.     } else { 
  30.         log.warn("checkTransactionState, decode message failed");
  31.     }
  33.     return null;
  34. }

DefaultMQProducerImpl

  1. @Override
  2. public void checkTransactionState(final String addr, final MessageExt msg,
  3.     final CheckTransactionStateRequestHeader header) { 
  4.     Runnable request = new Runnable() { 
  5.         private final String brokerAddr = addr;
  6.         private final MessageExt message = msg;
  7.         private final CheckTransactionStateRequestHeader checkRequestHeader = header;
  8.         private final String group = DefaultMQProducerImpl.this.defaultMQProducer.getProducerGroup();
  10.         @Override
  11.         public void run() { 
  12.             TransactionCheckListener transactionCheckListener = DefaultMQProducerImpl.this.checkListener();
  13.             TransactionListener transactionListener = getCheckListener();
  14.             if (transactionCheckListener != null || transactionListener != null) { 
  15.                 LocalTransactionState localTransactionState = LocalTransactionState.UNKNOW;
  16.                 Throwable exception = null;
  17.                 try { 
  18.                     if (transactionCheckListener != null) { 
  19.                         localTransactionState = transactionCheckListener.checkLocalTransactionState(message);
  20.                     } else if (transactionListener != null) { 
  21.                         log.debug("Used new check API in transaction message");
  22.                         // 回查本地事务状态
  23.                         localTransactionState = transactionListener.checkLocalTransaction(message);
  24.                     } else { 
  25.                         log.warn("CheckTransactionState, pick transactionListener by group[{}] failed", group);
  26.                     }
  27.                 } catch (Throwable e) { 
  28.                     log.error("Broker call checkTransactionState, but checkLocalTransactionState exception", e);
  29.                     exception = e;
  30.                 }
  32.                 this.processTransactionState(
  33.                     localTransactionState,
  34.                     group,
  35.                     exception);
  36.             } else { 
  37.                 log.warn("CheckTransactionState, pick transactionCheckListener by group[{}] failed", group);
  38.             }
  39.         }
  41.         private void processTransactionState(
  42.             final LocalTransactionState localTransactionState,
  43.             final String producerGroup,
  44.             final Throwable exception) { 
  45.             final EndTransactionRequestHeader thisHeader = new EndTransactionRequestHeader();
  46.             thisHeader.setCommitLogOffset(checkRequestHeader.getCommitLogOffset());
  47.             thisHeader.setProducerGroup(producerGroup);
  48.             thisHeader.setTranStateTableOffset(checkRequestHeader.getTranStateTableOffset());
  49.             thisHeader.setFromTransactionCheck(true);
  51.             String uniqueKey = message.getProperties().get(MessageConst.PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX);
  52.             if (uniqueKey == null) { 
  53.                 uniqueKey = message.getMsgId();
  54.             }
  55.             thisHeader.setMsgId(uniqueKey);
  56.             thisHeader.setTransactionId(checkRequestHeader.getTransactionId());
  57.             switch (localTransactionState) { 
  58.                 case COMMIT_MESSAGE:
  59.                     thisHeader.setCommitOrRollback(MessageSysFlag.TRANSACTION_COMMIT_TYPE);
  60.                     break;
  61.                 case ROLLBACK_MESSAGE:
  62.                     thisHeader.setCommitOrRollback(MessageSysFlag.TRANSACTION_ROLLBACK_TYPE);
  63.                     log.warn("when broker check, client rollback this transaction, {}", thisHeader);
  64.                     break;
  65.                 case UNKNOW:
  66.                     thisHeader.setCommitOrRollback(MessageSysFlag.TRANSACTION_NOT_TYPE);
  67.                     log.warn("when broker check, client does not know this transaction state, {}", thisHeader);
  68.                     break;
  69.                 default:
  70.                     break;
  71.             }
  73.             String remark = null;
  74.             if (exception != null) { 
  75.                 remark = "checkLocalTransactionState Exception: " + RemotingHelper.exceptionSimpleDesc(exception);
  76.             }
  78.             try { 
  79.                 // 根据回查本地事务的状态提交或回滚消息，单向请求
  80.                 DefaultMQProducerImpl.this.mQClientFactory.getMQClientAPIImpl().endTransactionOneway(brokerAddr, thisHeader, remark,
  81.                     3000);
  82.             } catch (Exception e) { 
  83.                 log.error("endTransactionOneway exception", e);
  84.             }
  85.         }
  86.     };
  88.     // 将回查任务提交给线程池执行
  89.     this.checkExecutor.submit(request);
  90. }

通过调用自定义的监听器回查方法，检测本地事务状态，再根据回查本地事务的状态提交或回滚消息，单向请求。
在这里插入图片描述

参考资料

《儒猿技术窝——从 0 开始带你成为消息中间件实战高手》

发表评论

表情：
评论列表 （有 0 条评论，96人围观）

还没有评论，来说两句吧...

相关阅读