RocketMQ 源码解析：事务消息

栏目: 后端 · 发布时间: 6年前

内容简介：RocketMQ 源码解析：事务消息

原文地址： RocketMQ源码解析：事务消息

RocketMQ 带注释 地址： YunaiV/incubator-rocketmq

:smiling_imp:本系列每 1-2 周更新一篇，欢迎订阅、关注、收藏 GitHub。

- 2.1 Producer 发送事务消息
- 2.2 Broker 处理结束事务请求
- 2.3 Broker 生成 ConsumeQueue
- 3.1 Broker 发起【事务消息回查】
  - 3.1.1 官方V3.1.4：基于文件系统
    - 3.1.1.1 存储消息到 CommitLog
    - 3.1.1.2 写【事务消息】状态存储（TranStateTable）
    - 3.1.1.3 【事务消息】回查
    - 3.1.1.4 初始化【事务消息】状态存储（TranStateTable）
  - 3.1.2 官方V4.0.0：基于数据库
- 3.2 Producer 接收【事务消息回查】

1. 概述

必须必须必须前置阅读内容：

《事务消息（阿里云）》

2. 事务消息发送

2.1 Producer 发送事务消息

活动图如下（结合 核心代码 理解）：

实现代码如下：

1: // :arrow_down::arrow_down::arrow_down:【DefaultMQProducerImpl.java】
  2: /**
  3:  * 发送事务消息
  4:  *
  5:  * @param msg 消息
  6:  * @param tranExecuter 【本地事务】执行器
  7:  * @param arg 【本地事务】执行器参数
  8:  * @return 事务发送结果
  9:  * @throws MQClientException 当 Client 发生异常时
 10:  */
 11: public TransactionSendResult sendMessageInTransaction(final Message msg, final LocalTransactionExecuter tranExecuter, final Object arg)
 12:     throws MQClientException {
 13:     if (null == tranExecuter) {
 14:         throw new MQClientException("tranExecutor is null", null);
 15:     }
 16:     Validators.checkMessage(msg, this.defaultMQProducer);
 17: 
 18:     // 发送【Half消息】
 19:     SendResult sendResult;
 20:     MessageAccessor.putProperty(msg, MessageConst.PROPERTY_TRANSACTION_PREPARED, "true");
 21:     MessageAccessor.putProperty(msg, MessageConst.PROPERTY_PRODUCER_GROUP, this.defaultMQProducer.getProducerGroup());
 22:     try {
 23:         sendResult = this.send(msg);
 24:     } catch (Exception e) {
 25:         throw new MQClientException("send message Exception", e);
 26:     }
 27: 
 28:     // 处理发送【Half消息】结果
 29:     LocalTransactionState localTransactionState = LocalTransactionState.UNKNOW;
 30:     Throwable localException = null;
 31:     switch (sendResult.getSendStatus()) {
 32:         // 发送【Half消息】成功，执行【本地事务】逻辑
 33:         case SEND_OK: {
 34:             try {
 35:                 if (sendResult.getTransactionId() != null) { // 事务编号。目前开源版本暂时没用到，猜想ONS在使用。
 36:                     msg.putUserProperty("__transactionId__", sendResult.getTransactionId());
 37:                 }
 38: 
 39:                 // 执行【本地事务】逻辑
 40:                 localTransactionState = tranExecuter.executeLocalTransactionBranch(msg, arg);
 41:                 if (null == localTransactionState) {
 42:                     localTransactionState = LocalTransactionState.UNKNOW;
 43:                 }
 44: 
 45:                 if (localTransactionState != LocalTransactionState.COMMIT_MESSAGE) {
 46:                     log.info("executeLocalTransactionBranch return {}", localTransactionState);
 47:                     log.info(msg.toString());
 48:                 }
 49:             } catch (Throwable e) {
 50:                 log.info("executeLocalTransactionBranch exception", e);
 51:                 log.info(msg.toString());
 52:                 localException = e;
 53:             }
 54:         }
 55:         break;
 56:         // 发送【Half消息】失败，标记【本地事务】状态为回滚
 57:         case FLUSH_DISK_TIMEOUT:
 58:         case FLUSH_SLAVE_TIMEOUT:
 59:         case SLAVE_NOT_AVAILABLE:
 60:             localTransactionState = LocalTransactionState.ROLLBACK_MESSAGE;
 61:             break;
 62:         default:
 63:             break;
 64:     }
 65: 
 66:     // 结束事务：提交消息 COMMIT / ROLLBACK
 67:     try {
 68:         this.endTransaction(sendResult, localTransactionState, localException);
 69:     } catch (Exception e) {
 70:         log.warn("local transaction execute " + localTransactionState + ", but end broker transaction failed", e);
 71:     }
 72: 
 73:     // 返回【事务发送结果】
 74:     TransactionSendResult transactionSendResult = new TransactionSendResult();
 75:     transactionSendResult.setSendStatus(sendResult.getSendStatus());
 76:     transactionSendResult.setMessageQueue(sendResult.getMessageQueue());
 77:     transactionSendResult.setMsgId(sendResult.getMsgId());
 78:     transactionSendResult.setQueueOffset(sendResult.getQueueOffset());
 79:     transactionSendResult.setTransactionId(sendResult.getTransactionId());
 80:     transactionSendResult.setLocalTransactionState(localTransactionState);
 81:     return transactionSendResult;
 82: }
 83: 
 84: /**
 85:  * 结束事务：提交消息 COMMIT / ROLLBACK
 86:  *
 87:  * @param sendResult 发送【Half消息】结果
 88:  * @param localTransactionState 【本地事务】状态
 89:  * @param localException 执行【本地事务】逻辑产生的异常
 90:  * @throws RemotingException 当远程调用发生异常时
 91:  * @throws MQBrokerException 当 Broker 发生异常时
 92:  * @throws InterruptedException 当线程中断时
 93:  * @throws UnknownHostException 当解码消息编号失败是
 94:  */
 95: public void endTransaction(//
 96:     final SendResult sendResult, //
 97:     final LocalTransactionState localTransactionState, //
 98:     final Throwable localException) throws RemotingException, MQBrokerException, InterruptedException, UnknownHostException {
 99:     // 解码消息编号
100:     final MessageId id;
101:     if (sendResult.getOffsetMsgId() != null) {
102:         id = MessageDecoder.decodeMessageId(sendResult.getOffsetMsgId());
103:     } else {
104:         id = MessageDecoder.decodeMessageId(sendResult.getMsgId());
105:     }
106: 
107:     // 创建请求
108:     String transactionId = sendResult.getTransactionId();
109:     final String brokerAddr = this.mQClientFactory.findBrokerAddressInPublish(sendResult.getMessageQueue().getBrokerName());
110:     EndTransactionRequestHeader requestHeader = new EndTransactionRequestHeader();
111:     requestHeader.setTransactionId(transactionId);
112:     requestHeader.setCommitLogOffset(id.getOffset());
113:     switch (localTransactionState) {
114:         case COMMIT_MESSAGE:
115:             requestHeader.setCommitOrRollback(MessageSysFlag.TRANSACTION_COMMIT_TYPE);
116:             break;
117:         case ROLLBACK_MESSAGE:
118:             requestHeader.setCommitOrRollback(MessageSysFlag.TRANSACTION_ROLLBACK_TYPE);
119:             break;
120:         case UNKNOW:
121:             requestHeader.setCommitOrRollback(MessageSysFlag.TRANSACTION_NOT_TYPE);
122:             break;
123:         default:
124:             break;
125:     }
126:     requestHeader.setProducerGroup(this.defaultMQProducer.getProducerGroup());
127:     requestHeader.setTranStateTableOffset(sendResult.getQueueOffset());
128:     requestHeader.setMsgId(sendResult.getMsgId());
129:     String remark = localException != null ? ("executeLocalTransactionBranch exception: " + localException.toString()) : null;
130: 
131:     // 提交消息 COMMIT / ROLLBACK。！！！通信方式为：Oneway！！！
132:     this.mQClientFactory.getMQClientAPIImpl().endTransactionOneway(brokerAddr, requestHeader, remark, this.defaultMQProducer.getSendMsgTimeout());
133: }

2.2 Broker 处理结束事务请求

查询请求的消息，进行 提交 / 回滚 。实现代码如下：

1: // :arrow_down::arrow_down::arrow_down:【EndTransactionProcessor.java】
  2: public RemotingCommand processRequest(ChannelHandlerContext ctx, RemotingCommand request) throws RemotingCommandException {
  3:     final RemotingCommand response = RemotingCommand.createResponseCommand(null);
  4:     final EndTransactionRequestHeader requestHeader = (EndTransactionRequestHeader) request.decodeCommandCustomHeader(EndTransactionRequestHeader.class);
  5: 
  6:     // 省略代码 =》打印日志（只处理 COMMIT / ROLLBACK）
  7: 
  8:     // 查询提交的消息
  9:     final MessageExt msgExt = this.brokerController.getMessageStore().lookMessageByOffset(requestHeader.getCommitLogOffset());
 10:     if (msgExt != null) {
 11:         // 省略代码 =》校验消息
 12: 
 13:         // 生成消息
 14:         MessageExtBrokerInner msgInner = this.endMessageTransaction(msgExt);
 15:         msgInner.setSysFlag(MessageSysFlag.resetTransactionValue(msgInner.getSysFlag(), requestHeader.getCommitOrRollback()));
 16:         msgInner.setQueueOffset(requestHeader.getTranStateTableOffset());
 17:         msgInner.setPreparedTransactionOffset(requestHeader.getCommitLogOffset());
 18:         msgInner.setStoreTimestamp(msgExt.getStoreTimestamp());
 19:         if (MessageSysFlag.TRANSACTION_ROLLBACK_TYPE == requestHeader.getCommitOrRollback()) {
 20:             msgInner.setBody(null);
 21:         }
 22: 
 23:         // 存储生成消息
 24:         final MessageStore messageStore = this.brokerController.getMessageStore();
 25:         final PutMessageResult putMessageResult = messageStore.putMessage(msgInner);
 26: 
 27:         // 处理存储结果
 28:         if (putMessageResult != null) {
 29:             switch (putMessageResult.getPutMessageStatus()) {
 30:                 // Success
 31:                 case PUT_OK:
 32:                 case FLUSH_DISK_TIMEOUT:
 33:                 case FLUSH_SLAVE_TIMEOUT:
 34:                 case SLAVE_NOT_AVAILABLE:
 35:                     response.setCode(ResponseCode.SUCCESS);
 36:                     response.setRemark(null);
 37:                     break;
 38:                 // Failed
 39:                 case CREATE_MAPEDFILE_FAILED:
 40:                     response.setCode(ResponseCode.SYSTEM_ERROR);
 41:                     response.setRemark("create maped file failed.");
 42:                     break;
 43:                 case MESSAGE_ILLEGAL:
 44:                 case PROPERTIES_SIZE_EXCEEDED:
 45:                     response.setCode(ResponseCode.MESSAGE_ILLEGAL);
 46:                     response.setRemark("the message is illegal, maybe msg body or properties length not matched. msg body length limit 128k, msg properties length limit 32k.");
 47:                     break;
 48:                 case SERVICE_NOT_AVAILABLE:
 49:                     response.setCode(ResponseCode.SERVICE_NOT_AVAILABLE);
 50:                     response.setRemark("service not available now.");
 51:                     break;
 52:                 case OS_PAGECACHE_BUSY:
 53:                     response.setCode(ResponseCode.SYSTEM_ERROR);
 54:                     response.setRemark("OS page cache busy, please try another machine");
 55:                     break;
 56:                 case UNKNOWN_ERROR:
 57:                     response.setCode(ResponseCode.SYSTEM_ERROR);
 58:                     response.setRemark("UNKNOWN_ERROR");
 59:                     break;
 60:                 default:
 61:                     response.setCode(ResponseCode.SYSTEM_ERROR);
 62:                     response.setRemark("UNKNOWN_ERROR DEFAULT");
 63:                     break;
 64:             }
 65: 
 66:             return response;
 67:         } else {
 68:             response.setCode(ResponseCode.SYSTEM_ERROR);
 69:             response.setRemark("store putMessage return null");
 70:         }
 71:     } else {
 72:         response.setCode(ResponseCode.SYSTEM_ERROR);
 73:         response.setRemark("find prepared transaction message failed");
 74:         return response;
 75:     }
 76: 
 77:     return response;
 78: }

2.3 Broker 生成 ConsumeQueue

事务消息，提交（ COMMIT ）后才生成 ConsumeQueue 。

1: // :arrow_down::arrow_down::arrow_down:【DefaultMessageStore.java】
  2: public void doDispatch(DispatchRequest req) {
  3:     // 非事务消息 或 事务提交消息 建立 消息位置信息 到 ConsumeQueue
  4:     final int tranType = MessageSysFlag.getTransactionValue(req.getSysFlag());
  5:     switch (tranType) {
  6:         case MessageSysFlag.TRANSACTION_NOT_TYPE: // 非事务消息
  7:         case MessageSysFlag.TRANSACTION_COMMIT_TYPE: // 事务消息COMMIT
  8:             DefaultMessageStore.this.putMessagePositionInfo(req.getTopic(), req.getQueueId(), req.getCommitLogOffset(), req.getMsgSize(),
  9:                 req.getTagsCode(), req.getStoreTimestamp(), req.getConsumeQueueOffset());
 10:             break;
 11:         case MessageSysFlag.TRANSACTION_PREPARED_TYPE: // 事务消息PREPARED
 12:         case MessageSysFlag.TRANSACTION_ROLLBACK_TYPE: // 事务消息ROLLBACK
 13:             break;
 14:     }
 15:     // 省略代码 =》 建立 索引信息 到 IndexFile
 16: }

3. 事务消息回查

【事务消息回查】功能曾经开源过，目前（V4.0.0）暂未开源。如下是该功能的开源情况：

版本	【事务消息回查】
官方V3.0.4 ~ V3.1.4	基于文件系统实现	已开源
官方V3.1.5 ~ V4.0.0	基于数据库实现	未完全开源

我们来看看两种情况下是怎么实现的。

3.1 Broker 发起【事务消息回查】

3.1.1 官方V3.1.4：基于文件系统

仓库地址： https://github.com/YunaiV/rocketmq-3.1.9/tree/release_3.1.4

相较于普通消息，【事务消息】多依赖如下三个组件：

TransactionStateService ：事务状态服务，负责对【事务消息】进行管理，包括存储与更新事务消息状态、回查事务消息状态等等。
TranStateTable ：【事务消息】状态存储。基于 MappedFileQueue 实现，默认存储路径为 ~/store/transaction/statetable ，每条【事务消息】状态存储结构如下：

第几位	字段	说明	数据类型	字节数
1	offset	CommitLog 物理存储位置	Long	8
2	size	消息长度	Int	4
3	timestamp	消息存储时间，单位：秒	Int	4
4	producerGroupHash	producerGroup 求 HashCode	Int	4
5	state	事务状态	Int	4

TranRedoLog ： TranStateTable 重放日志，每次 写操作 TranStateTable 记录重放日志。当 Broker 异常关闭时，使用 TranRedoLog 恢复 TranStateTable 。基于 ConsumeQueue 实现， Topic 为 TRANSACTION_REDOLOG_TOPIC_XXXX ，默认存储路径为 ~/store/transaction/redolog 。

简单手绘逻辑图如下:smiling_imp:：

3.1.1.1 存储消息到 CommitLog

存储【half消息】到 CommitLog 时，消息队列位置（ queueOffset ）使用 TranStateTable 最大物理位置（可写入物理位置）。这样，消息可以索引到自己对应的 TranStateTable 的位置和记录。

核心代码如下：

1: // :arrow_down::arrow_down::arrow_down:【DefaultAppendMessageCallback.java】
  2: class DefaultAppendMessageCallback implements AppendMessageCallback {
  3:     public AppendMessageResult doAppend(final long fileFromOffset, final ByteBuffer byteBuffer,  final int maxBlank, final Object msg) {
  4:         // ...省略代码
  5: 
  6:         // 事务消息需要特殊处理 
  7:         final int tranType = MessageSysFlag.getTransactionValue(msgInner.getSysFlag());
  8:         switch (tranType) {
  9:         case MessageSysFlag.TransactionPreparedType: // 消息队列位置（queueOffset）使用 TranStateTable 最大物理位置（可写入物理位置） 
 10:             queueOffset = CommitLog.this.defaultMessageStore.getTransactionStateService().getTranStateTableOffset().get();
 11:             break;
 12:         case MessageSysFlag.TransactionRollbackType:
 13:             queueOffset = msgInner.getQueueOffset();
 14:             break;
 15:         case MessageSysFlag.TransactionNotType:
 16:         case MessageSysFlag.TransactionCommitType:
 17:         default:
 18:             break;
 19:         }
 20: 
 21:         // ...省略代码
 22: 
 23:         switch (tranType) {
 24:         case MessageSysFlag.TransactionPreparedType:
 25:             // 更新 TranStateTable 最大物理位置（可写入物理位置） 
 26:             CommitLog.this.defaultMessageStore.getTransactionStateService().getTranStateTableOffset().incrementAndGet();
 27:             break;
 28:         case MessageSysFlag.TransactionRollbackType:
 29:             break;
 30:         case MessageSysFlag.TransactionNotType:
 31:         case MessageSysFlag.TransactionCommitType:
 32:             // 更新下一次的ConsumeQueue信息
 33:             CommitLog.this.topicQueueTable.put(key, ++queueOffset);
 34:             break;
 35:         default:
 36:             break;
 37:         }
 38: 
 39:         // 返回结果
 40:         return result;
 41:     }
 42: }

3.1.1.2 写【事务消息】状态存储（TranStateTable）

处理【Half消息】时，新增【事务消息】状态存储（ TranStateTable ）。
处理【Commit / Rollback消息】时，更新【事务消息】状态存储（ TranStateTable ） COMMIT / ROLLBACK。
每次 写操作【 事务消息】状态存储（ TranStateTable ），记录重放日志（ TranRedoLog ）。

核心代码如下：

1: // :arrow_down::arrow_down::arrow_down:【DispatchMessageService.java】
  2: private void doDispatch() {
  3:     if (!this.requestsRead.isEmpty()) {
  4:         for (DispatchRequest req : this.requestsRead) {
  5: 
  6:             // ...省略代码
  7: 
  8:             // 2、写【事务消息】状态存储（TranStateTable）
  9:             if (req.getProducerGroup() != null) {
 10:                 switch (tranType) {
 11:                 case MessageSysFlag.TransactionNotType:
 12:                     break;
 13:                 case MessageSysFlag.TransactionPreparedType:
 14:                     // 新增 【事务消息】状态存储（TranStateTable）
 15:                     DefaultMessageStore.this.getTransactionStateService().appendPreparedTransaction(
 16:                         req.getCommitLogOffset(), req.getMsgSize(), (int) (req.getStoreTimestamp() / 1000), req.getProducerGroup().hashCode());
 17:                     break;
 18:                 case MessageSysFlag.TransactionCommitType:
 19:                 case MessageSysFlag.TransactionRollbackType:
 20:                     // 更新 【事务消息】状态存储（TranStateTable） COMMIT / ROLLBACK
 21:                     DefaultMessageStore.this.getTransactionStateService().updateTransactionState(
 22:                         req.getTranStateTableOffset(), req.getPreparedTransactionOffset(), req.getProducerGroup().hashCode(), tranType);
 23:                     break;
 24:                 }
 25:             }
 26:             // 3、记录 TranRedoLog
 27:             switch (tranType) {
 28:             case MessageSysFlag.TransactionNotType:
 29:                 break;
 30:             case MessageSysFlag.TransactionPreparedType:
 31:                 // 记录 TranRedoLog
 32:                 DefaultMessageStore.this.getTransactionStateService().getTranRedoLog().putMessagePostionInfoWrapper(
 33:                         req.getCommitLogOffset(), req.getMsgSize(), TransactionStateService.PreparedMessageTagsCode,
 34:                         req.getStoreTimestamp(), 0L);
 35:                 break;
 36:             case MessageSysFlag.TransactionCommitType:
 37:             case MessageSysFlag.TransactionRollbackType:
 38:                 // 记录 TranRedoLog
 39:                 DefaultMessageStore.this.getTransactionStateService().getTranRedoLog().putMessagePostionInfoWrapper(
 40:                         req.getCommitLogOffset(), req.getMsgSize(), req.getPreparedTransactionOffset(),
 41:                         req.getStoreTimestamp(), 0L);
 42:                 break;
 43:             }
 44:         }
 45: 
 46:         // ...省略代码
 47:     }
 48: }
 49: // :arrow_down::arrow_down::arrow_down:【TransactionStateService.java】
 50: /**
 51:  * 新增事务状态
 52:  *
 53:  * @param clOffset commitLog 物理位置
 54:  * @param size 消息长度
 55:  * @param timestamp 消息存储时间
 56:  * @param groupHashCode groupHashCode
 57:  * @return 是否成功
 58:  */
 59: public boolean appendPreparedTransaction(//
 60:         final long clOffset,//
 61:         final int size,//
 62:         final int timestamp,//
 63:         final int groupHashCode//
 64: ) {
 65:     MapedFile mapedFile = this.tranStateTable.getLastMapedFile();
 66:     if (null == mapedFile) {
 67:         log.error("appendPreparedTransaction: create mapedfile error.");
 68:         return false;
 69:     }
 70: 
 71:     // 首次创建，加入定时任务中
 72:     if (0 == mapedFile.getWrotePostion()) {
 73:         this.addTimerTask(mapedFile);
 74:     }
 75: 
 76:     this.byteBufferAppend.position(0);
 77:     this.byteBufferAppend.limit(TSStoreUnitSize);
 78: 
 79:     // Commit Log Offset
 80:     this.byteBufferAppend.putLong(clOffset);
 81:     // Message Size
 82:     this.byteBufferAppend.putInt(size);
 83:     // Timestamp
 84:     this.byteBufferAppend.putInt(timestamp);
 85:     // Producer Group Hashcode
 86:     this.byteBufferAppend.putInt(groupHashCode);
 87:     // Transaction State
 88:     this.byteBufferAppend.putInt(MessageSysFlag.TransactionPreparedType);
 89: 
 90:     return mapedFile.appendMessage(this.byteBufferAppend.array());
 91: }
 92: 
 93: /**
 94:  * 更新事务状态
 95:  *
 96:  * @param tsOffset tranStateTable 物理位置
 97:  * @param clOffset commitLog 物理位置
 98:  * @param groupHashCode groupHashCode
 99:  * @param state 事务状态
100:  * @return 是否成功
101:  */
102: public boolean updateTransactionState(
103:         final long tsOffset,
104:         final long clOffset,
105:         final int groupHashCode,
106:         final int state) {
107:     SelectMapedBufferResult selectMapedBufferResult = this.findTransactionBuffer(tsOffset);
108:     if (selectMapedBufferResult != null) {
109:         try {
110: 
111:             // ....省略代码：校验是否能够更新
112: 
113:             // 更新事务状态
114:             selectMapedBufferResult.getByteBuffer().putInt(TS_STATE_POS, state);
115:         }
116:         catch (Exception e) {
117:             log.error("updateTransactionState exception", e);
118:         }
119:         finally {
120:             selectMapedBufferResult.release();
121:         }
122:     }
123: 
124:     return false;
125: }

3.1.1.3 【事务消息】回查

TranStateTable 每个 MappedFile 都对应一个 Timer 。 Timer 固定周期（默认：60s）遍历 MappedFile ，查找【half消息】，向 Producer 发起【事务消息】回查请求。【事务消息】回查结果的逻辑不在此处进行，在时执行。

实现代码如下：

1: // :arrow_down::arrow_down::arrow_down:【TransactionStateService.java】
  2: /**
  3:  * 初始化定时任务
  4:  */
  5: private void initTimerTask() {
  6:     //
  7:     final List<MapedFile> mapedFiles = this.tranStateTable.getMapedFiles();
  8:     for (MapedFile mf : mapedFiles) {
  9:         this.addTimerTask(mf);
 10:     }
 11: }
 12: 
 13: /**
 14:  * 每个文件初始化定时任务
 15:  * @param mf 文件
 16:  */
 17: private void addTimerTask(final MapedFile mf) {
 18:     this.timer.scheduleAtFixedRate(new TimerTask() {
 19:         private final MapedFile mapedFile = mf;
 20:         private final TransactionCheckExecuter transactionCheckExecuter = TransactionStateService.this.defaultMessageStore.getTransactionCheckExecuter();
 21:         private final long checkTransactionMessageAtleastInterval = TransactionStateService.this.defaultMessageStore.getMessageStoreConfig()
 22:                     .getCheckTransactionMessageAtleastInterval();
 23:         private final boolean slave = TransactionStateService.this.defaultMessageStore.getMessageStoreConfig().getBrokerRole() == BrokerRole.SLAVE;
 24: 
 25:         @Override
 26:         public void run() {
 27:             // Slave不需要回查事务状态
 28:             if (slave) {
 29:                 return;
 30:             }
 31:             // Check功能是否开启
 32:             if (!TransactionStateService.this.defaultMessageStore.getMessageStoreConfig()
 33:                 .isCheckTransactionMessageEnable()) {
 34:                 return;
 35:             }
 36: 
 37:             try {
 38:                 SelectMapedBufferResult selectMapedBufferResult = mapedFile.selectMapedBuffer(0);
 39:                 if (selectMapedBufferResult != null) {
 40:                     long preparedMessageCountInThisMapedFile = 0; // 回查的【half消息】数量
 41:                     int i = 0;
 42:                     try {
 43:                         // 循环每条【事务消息】状态，对【half消息】进行回查
 44:                         for (; i < selectMapedBufferResult.getSize(); i += TSStoreUnitSize) {
 45:                             selectMapedBufferResult.getByteBuffer().position(i);
 46: 
 47:                             // Commit Log Offset
 48:                             long clOffset = selectMapedBufferResult.getByteBuffer().getLong();
 49:                             // Message Size
 50:                             int msgSize = selectMapedBufferResult.getByteBuffer().getInt();
 51:                             // Timestamp
 52:                             int timestamp = selectMapedBufferResult.getByteBuffer().getInt();
 53:                             // Producer Group Hashcode
 54:                             int groupHashCode = selectMapedBufferResult.getByteBuffer().getInt();
 55:                             // Transaction State
 56:                             int tranType = selectMapedBufferResult.getByteBuffer().getInt();
 57: 
 58:                             // 已经提交或者回滚的消息跳过
 59:                             if (tranType != MessageSysFlag.TransactionPreparedType) {
 60:                                 continue;
 61:                             }
 62: 
 63:                             // 遇到时间不符合最小轮询间隔，终止
 64:                             long timestampLong = timestamp * 1000;
 65:                             long diff = System.currentTimeMillis() - timestampLong;
 66:                             if (diff < checkTransactionMessageAtleastInterval) {
 67:                                 break;
 68:                             }
 69: 
 70:                             preparedMessageCountInThisMapedFile++;
 71: 
 72:                             // 回查Producer
 73:                             try {
 74:                                 this.transactionCheckExecuter.gotoCheck(groupHashCode, getTranStateOffset(i), clOffset, msgSize);
 75:                             } catch (Exception e) {
 76:                                 tranlog.warn("gotoCheck Exception", e);
 77:                             }
 78:                         }
 79: 
 80:                         // 无回查的【half消息】数量，且遍历完，则终止定时任务
 81:                         if (0 == preparedMessageCountInThisMapedFile //
 82:                                 && i == mapedFile.getFileSize()) {
 83:                             tranlog.info("remove the transaction timer task, because no prepared message in this mapedfile[{}]", mapedFile.getFileName());
 84:                             this.cancel();
 85:                         }
 86:                     } finally {
 87:                         selectMapedBufferResult.release();
 88:                     }
 89: 
 90:                     tranlog.info("the transaction timer task execute over in this period, {} Prepared Message: {} Check Progress: {}/{}", mapedFile.getFileName(),//
 91:                             preparedMessageCountInThisMapedFile, i / TSStoreUnitSize, mapedFile.getFileSize() / TSStoreUnitSize);
 92:                 } else if (mapedFile.isFull()) {
 93:                     tranlog.info("the mapedfile[{}] maybe deleted, cancel check transaction timer task", mapedFile.getFileName());
 94:                     this.cancel();
 95:                     return;
 96:                 }
 97:             } catch (Exception e) {
 98:                 log.error("check transaction timer task Exception", e);
 99:             }
100:         }
101: 
102: 
103:         private long getTranStateOffset(final long currentIndex) {
104:             long offset = (this.mapedFile.getFileFromOffset() + currentIndex) / TransactionStateService.TSStoreUnitSize;
105:             return offset;
106:         }
107:     }, 1000 * 60, this.defaultMessageStore.getMessageStoreConfig().getCheckTransactionMessageTimerInterval());
108: }
109: 
110: // 【DefaultTransactionCheckExecuter.java】
111: @Override
112: public void gotoCheck(int producerGroupHashCode, long tranStateTableOffset, long commitLogOffset,
113:         int msgSize) {
114:     // 第一步、查询Producer
115:     final ClientChannelInfo clientChannelInfo = this.brokerController.getProducerManager().pickProducerChannelRandomly(producerGroupHashCode);
116:     if (null == clientChannelInfo) {
117:         log.warn("check a producer transaction state, but not find any channel of this group[{}]", producerGroupHashCode);
118:         return;
119:     }
120: 
121:     // 第二步、查询消息
122:     SelectMapedBufferResult selectMapedBufferResult = this.brokerController.getMessageStore().selectOneMessageByOffset(commitLogOffset, msgSize);
123:     if (null == selectMapedBufferResult) {
124:         log.warn("check a producer transaction state, but not find message by commitLogOffset: {}, msgSize: ", commitLogOffset, msgSize);
125:         return;
126:     }
127: 
128:     // 第三步、向Producer发起请求
129:     final CheckTransactionStateRequestHeader requestHeader = new CheckTransactionStateRequestHeader();
130:     requestHeader.setCommitLogOffset(commitLogOffset);
131:     requestHeader.setTranStateTableOffset(tranStateTableOffset);
132:     this.brokerController.getBroker2Client().checkProducerTransactionState(clientChannelInfo.getChannel(), requestHeader, selectMapedBufferResult);
133: }

3.1.1.4 初始化【事务消息】状态存储（TranStateTable）

根据最后 Broker 关闭是否正常，会有不同的初始化方式。

核心代码如下：

1: // :arrow_down::arrow_down::arrow_down:【TransactionStateService.java】
  2: /**
  3:  * 初始化 TranRedoLog
  4:  * @param lastExitOK 是否正常退出
  5:  */
  6: public void recoverStateTable(final boolean lastExitOK) {
  7:     if (lastExitOK) {
  8:         this.recoverStateTableNormal();
  9:     } else {
 10:         // 第一步，删除State Table
 11:         this.tranStateTable.destroy();
 12:         // 第二步，通过RedoLog全量恢复StateTable
 13:         this.recreateStateTable();
 14:     }
 15: }
 16: 
 17: /**
 18:  * 扫描 TranRedoLog 重建 StateTable
 19:  */
 20: private void recreateStateTable() {
 21:     this.tranStateTable = new MapedFileQueue(StorePathConfigHelper.getTranStateTableStorePath(defaultMessageStore
 22:                 .getMessageStoreConfig().getStorePathRootDir()), defaultMessageStore
 23:                 .getMessageStoreConfig().getTranStateTableMapedFileSize(), null);
 24: 
 25:     final TreeSet<Long> preparedItemSet = new TreeSet<Long>();
 26: 
 27:     // 第一步，从头扫描RedoLog
 28:     final long minOffset = this.tranRedoLog.getMinOffsetInQuque();
 29:     long processOffset = minOffset;
 30:     while (true) {
 31:         SelectMapedBufferResult bufferConsumeQueue = this.tranRedoLog.getIndexBuffer(processOffset);
 32:         if (bufferConsumeQueue != null) {
 33:             try {
 34:                 long i = 0;
 35:                 for (; i < bufferConsumeQueue.getSize(); i += ConsumeQueue.CQStoreUnitSize) {
 36:                     long offsetMsg = bufferConsumeQueue.getByteBuffer().getLong();
 37:                     int sizeMsg = bufferConsumeQueue.getByteBuffer().getInt();
 38:                     long tagsCode = bufferConsumeQueue.getByteBuffer().getLong();
 39: 
 40:                     if (TransactionStateService.PreparedMessageTagsCode == tagsCode) { // Prepared
 41:                         preparedItemSet.add(offsetMsg);
 42:                     } else { // Commit/Rollback
 43:                         preparedItemSet.remove(tagsCode);
 44:                     }
 45:                 }
 46: 
 47:                 processOffset += i;
 48:             } finally { // 必须释放资源
 49:                 bufferConsumeQueue.release();
 50:             }
 51:         } else {
 52:             break;
 53:         }
 54:     }
 55:     log.info("scan transaction redolog over, End offset: {},  Prepared Transaction Count: {}", processOffset, preparedItemSet.size());
 56: 
 57:     // 第二步，重建StateTable
 58:     Iterator<Long> it = preparedItemSet.iterator();
 59:     while (it.hasNext()) {
 60:         Long offset = it.next();
 61:         MessageExt msgExt = this.defaultMessageStore.lookMessageByOffset(offset);
 62:         if (msgExt != null) {
 63:             this.appendPreparedTransaction(msgExt.getCommitLogOffset(), msgExt.getStoreSize(),
 64:                 (int) (msgExt.getStoreTimestamp() / 1000),
 65:                 msgExt.getProperty(MessageConst.PROPERTY_PRODUCER_GROUP).hashCode());
 66:             this.tranStateTableOffset.incrementAndGet();
 67:         }
 68:     }
 69: }
 70: 
 71: /**
 72:  * 加载（解析）TranStateTable 的 MappedFile
 73:  * 1. 清理多余 MappedFile，设置最后一个 MappedFile的写入位置(position
 74:  * 2. 设置 TanStateTable 最大物理位置（可写入位置）
 75:  */
 76: private void recoverStateTableNormal() {
 77:     final List<MapedFile> mapedFiles = this.tranStateTable.getMapedFiles();
 78:     if (!mapedFiles.isEmpty()) {
 79:         // 从倒数第三个文件开始恢复
 80:         int index = mapedFiles.size() - 3;
 81:         if (index < 0) {
 82:             index = 0;
 83:         }
 84: 
 85:         int mapedFileSizeLogics = this.tranStateTable.getMapedFileSize();
 86:         MapedFile mapedFile = mapedFiles.get(index);
 87:         ByteBuffer byteBuffer = mapedFile.sliceByteBuffer();
 88:         long processOffset = mapedFile.getFileFromOffset();
 89:         long mapedFileOffset = 0;
 90:         while (true) {
 91:             for (int i = 0; i < mapedFileSizeLogics; i += TSStoreUnitSize) {
 92: 
 93:                 final long clOffset_read = byteBuffer.getLong();
 94:                 final int size_read = byteBuffer.getInt();
 95:                 final int timestamp_read = byteBuffer.getInt();
 96:                 final int groupHashCode_read = byteBuffer.getInt();
 97:                 final int state_read = byteBuffer.getInt();
 98: 
 99:                 boolean stateOK = false;
100:                 switch (state_read) {
101:                 case MessageSysFlag.TransactionPreparedType:
102:                 case MessageSysFlag.TransactionCommitType:
103:                 case MessageSysFlag.TransactionRollbackType:
104:                     stateOK = true;
105:                     break;
106:                 default:
107:                     break;
108:                 }
109: 
110:                 // 说明当前存储单元有效
111:                 if (clOffset_read >= 0 && size_read > 0 && stateOK) {
112:                     mapedFileOffset = i + TSStoreUnitSize;
113:                 } else {
114:                     log.info("recover current transaction state table file over,  " + mapedFile.getFileName() + " "
115:                             + clOffset_read + " " + size_read + " " + timestamp_read);
116:                     break;
117:                 }
118:             }
119: 
120:             // 走到文件末尾，切换至下一个文件
121:             if (mapedFileOffset == mapedFileSizeLogics) {
122:                 index++;
123:                 if (index >= mapedFiles.size()) { // 循环while结束
124:                     log.info("recover last transaction state table file over, last maped file " + mapedFile.getFileName());
125:                     break;
126:                 } else { // 切换下一个文件
127:                     mapedFile = mapedFiles.get(index);
128:                     byteBuffer = mapedFile.sliceByteBuffer();
129:                     processOffset = mapedFile.getFileFromOffset();
130:                     mapedFileOffset = 0;
131:                     log.info("recover next transaction state table file, " + mapedFile.getFileName());
132:                 }
133:             } else {
134:                 log.info("recover current transaction state table queue over " + mapedFile.getFileName() + " " + (processOffset + mapedFileOffset));
135:                 break;
136:             }
137:         }
138: 
139:         // 清理多余 MappedFile，设置最后一个 MappedFile的写入位置(position
140:         processOffset += mapedFileOffset;
141:         this.tranStateTable.truncateDirtyFiles(processOffset);
142: 
143:         // 设置 TanStateTable 最大物理位置（可写入位置）
144:         this.tranStateTableOffset.set(this.tranStateTable.getMaxOffset() / TSStoreUnitSize);
145:         log.info("recover normal over, transaction state table max offset: {}", this.tranStateTableOffset.get());
146:     }
147: }

3.1.1.5 补充

为什么 V3.1.5 开始，使用数据库实现【事务状态】的存储？如下是来自官方文档的说明，可能是一部分原因：

RocketMQ 这种实现事务方式，没有通过 KV 存储做，而是通过 Offset 方式，存在一个显著缺陷，即通过 Offset 更改数据，会令系统的脏页过多，需要特别关注。

3.1.2 官方V4.0.0：基于数据库

仓库地址： https://github.com/apache/incubator-rocketmq

官方V4.0.0 暂时未完全开源【事务消息回查】功能， So 我们需要进行一些猜想，可能不一定正确:smiling_imp: 。

:laughing:我们来对比【官方V3.1.4：基于文件】的实现。

TransactionRecord ：记录每条【事务消息】。类似 TranStateTable 。

TranStateTable	TransactionRecord
offset	offset
producerGroupHash	producerGroup
size	无	非必须字段：【事务消息】回查时，使用 offset 读取 CommitLog 获得。
timestamp	无	非必须字段：【事务消息】回查时，使用 offset 读取 CommitLog 获得。
state	无	非必须字段：事务开始，增加记录；事务结束，删除记录。

另外，数据库本身保证了数据存储的可靠性，无需 TranRedoLog 。

简单手绘逻辑图如下:smiling_imp:：

3.2 Producer 接收【事务消息回查】

顺序图如下：

核心代码如下：

1: // :arrow_down::arrow_down::arrow_down:【DefaultMQProducerImpl.java】
  2: /**
  3:  * 检查【事务状态】状态
  4:  *
  5:  * @param addr broker地址
  6:  * @param msg 消息
  7:  * @param header 请求
  8:  */
  9: @Override
 10: public void checkTransactionState(final String addr, final MessageExt msg, final CheckTransactionStateRequestHeader header) {
 11:     Runnable request = new Runnable() {
 12:         private final String brokerAddr = addr;
 13:         private final MessageExt message = msg;
 14:         private final CheckTransactionStateRequestHeader checkRequestHeader = header;
 15:         private final String group = DefaultMQProducerImpl.this.defaultMQProducer.getProducerGroup();
 16: 
 17:         @Override
 18:         public void run() {
 19:             TransactionCheckListener transactionCheckListener = DefaultMQProducerImpl.this.checkListener();
 20:             if (transactionCheckListener != null) {
 21:                 // 获取事务执行状态
 22:                 LocalTransactionState localTransactionState = LocalTransactionState.UNKNOW;
 23:                 Throwable exception = null;
 24:                 try {
 25:                     localTransactionState = transactionCheckListener.checkLocalTransactionState(message);
 26:                 } catch (Throwable e) {
 27:                     log.error("Broker call checkTransactionState, but checkLocalTransactionState exception", e);
 28:                     exception = e;
 29:                 }
 30: 
 31:                 // 处理事务结果，提交消息 COMMIT / ROLLBACK
 32:                 this.processTransactionState(//
 33:                     localTransactionState, //
 34:                     group, //
 35:                     exception);
 36:             } else {
 37:                 log.warn("checkTransactionState, pick transactionCheckListener by group[{}] failed", group);
 38:             }
 39:         }
 40: 
 41:         /**
 42:          * 处理事务结果，提交消息 COMMIT / ROLLBACK
 43:          *
 44:          * @param localTransactionState 【本地事务】状态
 45:          * @param producerGroup producerGroup
 46:          * @param exception 检查【本地事务】状态发生的异常
 47:          */
 48:         private void processTransactionState(//
 49:             final LocalTransactionState localTransactionState, //
 50:             final String producerGroup, //
 51:             final Throwable exception) {
 52:             final EndTransactionRequestHeader thisHeader = new EndTransactionRequestHeader();
 53:             thisHeader.setCommitLogOffset(checkRequestHeader.getCommitLogOffset());
 54:             thisHeader.setProducerGroup(producerGroup);
 55:             thisHeader.setTranStateTableOffset(checkRequestHeader.getTranStateTableOffset());
 56:             thisHeader.setFromTransactionCheck(true);
 57: 
 58:             // 设置消息编号
 59:             String uniqueKey = message.getProperties().get(MessageConst.PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX);
 60:             if (uniqueKey == null) {
 61:                 uniqueKey = message.getMsgId();
 62:             }
 63:             thisHeader.setMsgId(uniqueKey);
 64: 
 65:             thisHeader.setTransactionId(checkRequestHeader.getTransactionId());
 66:             switch (localTransactionState) {
 67:                 case COMMIT_MESSAGE:
 68:                     thisHeader.setCommitOrRollback(MessageSysFlag.TRANSACTION_COMMIT_TYPE);
 69:                     break;
 70:                 case ROLLBACK_MESSAGE:
 71:                     thisHeader.setCommitOrRollback(MessageSysFlag.TRANSACTION_ROLLBACK_TYPE);
 72:                     log.warn("when broker check, client rollback this transaction, {}", thisHeader);
 73:                     break;
 74:                 case UNKNOW:
 75:                     thisHeader.setCommitOrRollback(MessageSysFlag.TRANSACTION_NOT_TYPE);
 76:                     log.warn("when broker check, client does not know this transaction state, {}", thisHeader);
 77:                     break;
 78:                 default:
 79:                     break;
 80:             }
 81: 
 82:             String remark = null;
 83:             if (exception != null) {
 84:                 remark = "checkLocalTransactionState Exception: " + RemotingHelper.exceptionSimpleDesc(exception);
 85:             }
 86: 
 87:             try {
 88:                 // 提交消息 COMMIT / ROLLBACK
 89:                 DefaultMQProducerImpl.this.mQClientFactory.getMQClientAPIImpl().endTransactionOneway(brokerAddr, thisHeader, remark,
 90:                     3000);
 91:             } catch (Exception e) {
 92:                 log.error("endTransactionOneway exception", e);
 93:             }
 94:         }
 95:     };
 96: 
 97:     // 提交执行
 98:     this.checkExecutor.submit(request);
 99: }
100: 
101: // :arrow_down::arrow_down::arrow_down:【DefaultMQProducerImpl.java】
102: /**
103:  * 【事务消息回查】检查监听器
104:  */
105: public interface TransactionCheckListener {
106: 
107:     /**
108:      * 获取（检查）【本地事务】状态
109:      *
110:      * @param msg 消息
111:      * @return 事务状态
112:      */
113:     LocalTransactionState checkLocalTransactionState(final MessageExt msg);
114: 
115: }

以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持码农网

查看所有标签

猜你喜欢:

本站部分资源来源于网络，本站转载出于传递更多信息之目的，版权归原作者或者来源机构所有，如转载稿涉及版权问题，请联系我们。

码农书籍

AI Algorithms, Data Structures, and Idioms in Prolog, Lisp, and

George F. Luger、William A Stubblefield / Addison Wesley / 2008-09-04 / USD 22.20

This book is designed for three primary purposes. The first is as a programming language component of a general class in Artificial Intelligence. From this viewpoint, the authors see as essential that......一起来看看《AI Algorithms, Data Structures, and Idioms in Prolog, Lisp, and 》这本书的介绍吧!

码农工具

在线进制转换器

各进制数互转换器

MD5 加密

MD5 加密工具

RGB HSV 转换

RGB HSV 互转工具