怅然|消息队列之事务消息，RocketMQ 和 Kafka是如何做的？( 三 )
流程也就是我们上面分析的 ， 将消息塞入一些属性 ， 标明此时这个消息还是半消息 ， 然后发送至 Broker， 然后执行本地事务 ， 然后将本地事务的执行状态发送给 Broker， 我们现在再来看下 Broker 到底是怎么处理这个消息的 。
在 Broker 的 SendMessageProcessor#sendMessage 中会处理这个半消息请求 ， 因为今天主要分析的是事务消息 ， 所以其他流程不做分析 ， 我大致的说一下原理 。
简单的说就是sendMessage中查到接受来的消息的属性里面
MessageConst.PROPERTY_TRANSACTION_PREPARED是true ， 那么可以得知这个消息是事务消息 ， 然后再判断一下这条消息是否超过最大消费次数 ， 是否要延迟 ， Broker 是否接受事务消息等操作后 ， 将这条消息真正的topic和队列存入属性中 ， 然后重置消息的 topic 为RMQ_SYS_TRANS_HALF_TOPIC ， 并且队列是 0 的队列中 ， 使得消费者无法读取这个消息 。
以上就是整体处理半消息的流程 ， 我们来看一下源码 。
就是来了波狸猫换太子 ， 其实延时消息也是这么实现的 ， 最终将换了皮的消息入盘 。
Broker处理提交或者回滚消息的处理方法是
EndTransactionProcessor#processRequest， 我们来看一看它做了什么操作 。
可以看到 ， 如果是提交事务就是把皮再换回来写入真正的topic所属的队列中 ， 供消费者消费 ， 如果是回滚则是将半消息记录到一个 half_op 主题下 ， 到时候后台服务扫描半消息的时候就依据其来判断这个消息已经处理过了 。
那个后台服务就是TransactionalMessageCheckService服务 ， 它会定时的扫描半消息队列 ， 去请求反查接口看看事务成功了没 ， 具体执行的就是
TransactionalMessageServiceImpl#check方法 。
我大致说一下流程 ， 这一步骤其实涉及到的代码很多 ， 我就不贴代码了 ， 有兴趣的同学自行了解 。 不过我相信用语言也是能说清楚的 。
首先取半消息 topic 即RMQ_SYS_TRANS_HALF_TOPIC下的所有队列 ， 如果还记得上面内容的话 ， 就知道半消息写入的队列是 id 是 0 的这个队列 ， 然后取出这个队列对应的half_op主题下的队列 ， 即RMQ_SYS_TRANS_OP_HALF_TOPIC主题下的队列 。
这个 half_op 主要是为了记录这个事务消息已经被处理过 ， 也就是说已经得知此事务消息是提交的还是回滚的消息会被记录在 half_op 中 。
然后调用 fillOpRemoveMap 方法 ， 从half_op取一批已经处理过的消息来去重 ， 将那些没有记录在half_op里面的半消息调用putBackHalfMsgQueue又写入了commitlog中 ， 然后发送事务反查请求 ， 这个反查请求也是oneWay ， 即不会等待响应 。 当然此时的半消息队列的消费 offset 也会推进 。
然后producer中的 ClientRemotingProcessor#processRequest 会处理这个请求 ， 会把任务扔到 TransactionMQProducer 的线程池中进行 ， 最终会调用上面我们发消息时候定义的checkLocalTransactionState方法 ， 然后将事务状态发送给Broker ， 也是用oneWay的方式 。
看到这里相信大家会有一些疑问 ， 比如为什么要有个half_op ， 为什么半消息处理了还要再写入commitlog中别急听我一一道来 。
推荐阅读
- 湖人队|狂轰62+21+13！浓眉太无解，詹姆斯后仰杀死比赛，湖人却获坏消息！
- 智通财经|| 远洋服务向港交所递表 在管面积约4230万平方米，新股消息
- 第一财经|免税概念利好消息频出，机构看好板块未来强劲增长丨牛熊眼
- 第一财经|新力控股联席董事长陈凯传出离职消息
- 带你逛服贸 | 文化服务专题展 规模大、颜值高、创意多
- 山东鲁能|山东鲁能迎来好消息，“大鱼”正积极运作，李霄鹏乐坏了
- 财经|新力控股联席董事长陈凯传出离职消息
- 半年|新力控股联席董事长陈凯传出离职消息
- 中国军备|外媒大胆预测，今年下半年中国将迎来三个好消息，每个都举世瞩目
- 华光环能|好消息！华光环能：子公司中标污水处理总承包项目