RocketMQ事务消息保证消息的可靠性和一致性
作者：Acqierement 发布时间：2022-09-05 00:25:26
这篇讲解一下rocketMq的事务消息的原理
在发送事务消息的时候，会加一个标识，表示这个消息是事务消息。broker接收到消息后，在我们之前看的代码里org.apache.rocketmq.broker.processor.SendMessageProcessor#sendMessage会判断是否是事务消息。
if (sendTransactionPrepareMessage) {
asyncPutMessageFuture = this.brokerController.getTransactionalMessageService().asyncPrepareMessage(msgInner);
} else {
asyncPutMessageFuture = this.brokerController.getMessageStore().asyncPutMessage(msgInner);
}
sendTransactionPrepareMessage=true表示是事务消息，所以走了一个单独的逻辑。
public CompletableFuture<PutMessageResult> asyncPutHalfMessage(MessageExtBrokerInner messageInner) {
return store.asyncPutMessage(parseHalfMessageInner(messageInner));
}
这里parseHalfMessageInner这个方法里面开始了偷梁换柱，把topic和queueId都改了，把原本的信息先存在变量里面。所以实际上这个消息发到了半消息专有的topic里面，topic名字叫做RMQ_SYS_TRANS_HALF_TOPIC
private MessageExtBrokerInner parseHalfMessageInner(MessageExtBrokerInner msgInner) {
MessageAccessor.putProperty(msgInner, MessageConst.PROPERTY_REAL_TOPIC, msgInner.getTopic());
MessageAccessor.putProperty(msgInner, MessageConst.PROPERTY_REAL_QUEUE_ID,
String.valueOf(msgInner.getQueueId()));
msgInner.setSysFlag(
MessageSysFlag.resetTransactionValue(msgInner.getSysFlag(), MessageSysFlag.TRANSACTION_NOT_TYPE));
msgInner.setTopic(TransactionalMessageUtil.buildHalfTopic());
msgInner.setQueueId(0);
msgInner.setPropertiesString(MessageDecoder.messageProperties2String(msgInner.getProperties()));
return msgInner;
}
然后其他代码还是和普通的消息一样，就是把事务消息做了转发，存在了RMQ_SYS_TRANS_HALF_TOPIC里面。
到这里发送半消息就成功了，然后最后客户端发送了半消息之后，会查一下本地事务的情况是否完成。这里有3种情况：commit、rollback、未知。完成和回滚都是确认的状态，这个比较好处理，比较难的是未知。我们先看能得到确认结果的情况。
如果完成和回滚，会给客户端发送结束事务的消息，这个消息叫END_TRANSACTION，包括消息里面包括了之前发送的半消息的id和offset。
broker处理的代码在org.apache.rocketmq.broker.processor.EndTransactionProcessor#processRequest中。就是根据offset拿到半消息，然后如果是commit，就是把原本的topic和queueId还原，发到原本的队列里面，这样就可以正常消费了。然后把这个半消息“删除”。如果是rollBack，也是拿到这个半消息，然后直接“删除”就可以了。接下来看一下怎么“删除”。
为什么我删除会打引号呢？因为半消息其实就是跟正常的消息一样，存在commitLog文件里面，mq的设计，就没有删除这个功能。所以所谓的删除其实就是把这个消息消费掉，不做任何处理，就是删除了。
想象一下，这个半消息有commit/rollBack/未知，3种状态，未知的肯定不能删除，那他怎么知道哪些消息是可以删除的呢？总不能所有的都再去客户端查一下事务的结果吧？mq怎么做的呢？前面提到的删除其实就是把这些commit和rollBack处理过后的半消息，再保存起来，后面消费半消息的数据的时候，只要从里面查一下是否需要删除就可以了。
这里又有一个问题，怎么把需要删除的半消息存起来呢？mq存储数据就是commitLog，所以其实这些需要删除的数据，就是又发到了一个特定的topic里面。这个topic名字是RMQ_SYS_TRANS_OP_HALF_TOPIC。主意区分，原本半消息的topic名字是half_topic，这个topic名字是op_half_topic，存储的是处理过后，可以删除的半消息。
所以说前面提到的带引号的“删除”，就是把消息发到op_half_topic就表示是删除了，这个op_half_topic消息的内容就是half_topic的offset。那么现在需要有个地方，来消费half_topic，然后判断是否存在于op_half_topic，如果是表示可以删除了，如果不是，就接着保存起来。
处理逻辑就在TransactionalMessageCheckService这个定时任务中。具体是在TransactionalMessageServiceImpl#check方法里面
@Override
public void check(long transactionTimeout, int transactionCheckMax,
AbstractTransactionalMessageCheckListener listener) {
try {
String topic = TopicValidator.RMQ_SYS_TRANS_HALF_TOPIC;
// 先拿到半消息
Set<MessageQueue> msgQueues = transactionalMessageBridge.fetchMessageQueues(topic);
if (msgQueues == null || msgQueues.size() == 0) {
log.warn("The queue of topic is empty :" + topic);
return;
}
log.debug("Check topic={}, queues={}", topic, msgQueues);
for (MessageQueue messageQueue : msgQueues) {
long startTime = System.currentTimeMillis();
MessageQueue opQueue = getOpQueue(messageQueue);
// 拿到半消息的最小偏移量
long halfOffset = transactionalMessageBridge.fetchConsumeOffset(messageQueue);
// 拿到op_half的最小偏移量
long opOffset = transactionalMessageBridge.fetchConsumeOffset(opQueue);
log.info("Before check, the queue={} msgOffset={} opOffset={}", messageQueue, halfOffset, opOffset);
if (halfOffset < 0 || opOffset < 0) {
log.error("MessageQueue: {} illegal offset read: {}, op offset: {},skip this queue", messageQueue,
halfOffset, opOffset);
continue;
}
List<Long> doneOpOffset = new ArrayList<>();
HashMap<Long, Long> removeMap = new HashMap<>();
// 拉取op的消息（32条），op消息内容是half的offset，跟half_topic的最小offset比较，如果op的小于最小的，就说明已经处理过了，放在doneOpOffset，反之，则说明还没处理过，就先放在removeMap里面
PullResult pullResult = fillOpRemoveMap(removeMap, opQueue, opOffset, halfOffset, doneOpOffset);
if (null == pullResult) {
log.error("The queue={} check msgOffset={} with opOffset={} failed, pullResult is null",
messageQueue, halfOffset, opOffset);
continue;
}
// single thread
int getMessageNullCount = 1;
long newOffset = halfOffset;
long i = halfOffset;
// 然后对half_topic进行处理
while (true) {
if (System.currentTimeMillis() - startTime > MAX_PROCESS_TIME_LIMIT) {
log.info("Queue={} process time reach max={}", messageQueue, MAX_PROCESS_TIME_LIMIT);
break;
}
// 如果这个offset已经处理过了，就接着处理下一个
if (removeMap.containsKey(i)) {
log.debug("Half offset {} has been committed/rolled back", i);
Long removedOpOffset = removeMap.remove(i);
doneOpOffset.add(removedOpOffset);
} else {
// 如果没有处理过，就要把数据捞出来重新投递
GetResult getResult = getHalfMsg(messageQueue, i);
MessageExt msgExt = getResult.getMsg();
if (msgExt == null) {
if (getMessageNullCount++ > MAX_RETRY_COUNT_WHEN_HALF_NULL) {
break;
}
if (getResult.getPullResult().getPullStatus() == PullStatus.NO_NEW_MSG) {
log.debug("No new msg, the miss offset={} in={}, continue check={}, pull result={}", i,
messageQueue, getMessageNullCount, getResult.getPullResult());
break;
} else {
log.info("Illegal offset, the miss offset={} in={}, continue check={}, pull result={}",
i, messageQueue, getMessageNullCount, getResult.getPullResult());
i = getResult.getPullResult().getNextBeginOffset();
newOffset = i;
continue;
}
}
if (needDiscard(msgExt, transactionCheckMax) || needSkip(msgExt)) {
listener.resolveDiscardMsg(msgExt);
newOffset = i + 1;
i++;
continue;
}
if (msgExt.getStoreTimestamp() >= startTime) {
log.debug("Fresh stored. the miss offset={}, check it later, store={}", i,
new Date(msgExt.getStoreTimestamp()));
break;
}
long valueOfCurrentMinusBorn = System.currentTimeMillis() - msgExt.getBornTimestamp();
long checkImmunityTime = transactionTimeout;
String checkImmunityTimeStr = msgExt.getUserProperty(MessageConst.PROPERTY_CHECK_IMMUNITY_TIME_IN_SECONDS);
if (null != checkImmunityTimeStr) {
checkImmunityTime = getImmunityTime(checkImmunityTimeStr, transactionTimeout);
if (valueOfCurrentMinusBorn < checkImmunityTime) {
if (checkPrepareQueueOffset(removeMap, doneOpOffset, msgExt)) {
newOffset = i + 1;
i++;
continue;
}
}
} else {
if (0 <= valueOfCurrentMinusBorn && valueOfCurrentMinusBorn < checkImmunityTime) {
log.debug("New arrived, the miss offset={}, check it later checkImmunity={}, born={}", i,
checkImmunityTime, new Date(msgExt.getBornTimestamp()));
break;
}
}
List<MessageExt> opMsg = pullResult.getMsgFoundList();
boolean isNeedCheck = opMsg == null && valueOfCurrentMinusBorn > checkImmunityTime
|| opMsg != null && opMsg.get(opMsg.size() - 1).getBornTimestamp() - startTime > transactionTimeout
|| valueOfCurrentMinusBorn <= -1;
if (isNeedCheck) {
// 重新投递
if (!putBackHalfMsgQueue(msgExt, i)) {
continue;
}
// 再重新确认事务
listener.resolveHalfMsg(msgExt);
} else {
pullResult = fillOpRemoveMap(removeMap, opQueue, pullResult.getNextBeginOffset(), halfOffset, doneOpOffset);
log.debug("The miss offset:{} in messageQueue:{} need to get more opMsg, result is:{}", i,
messageQueue, pullResult);
continue;
}
}
newOffset = i + 1;
i++;
}
// 更新offset
if (newOffset != halfOffset) {
transactionalMessageBridge.updateConsumeOffset(messageQueue, newOffset);
}
long newOpOffset = calculateOpOffset(doneOpOffset, opOffset);
if (newOpOffset != opOffset) {
transactionalMessageBridge.updateConsumeOffset(opQueue, newOpOffset);
}
}
} catch (Throwable e) {
log.error("Check error", e);
}
}
我讲解一下这个代码做了啥。我们先明确这个代码是要实现什么功能。就是消费half_topic，然后去根据op_half_topic的数据来判断half_topc的消息是否被处理过，处理过了就直接忽略、丢弃，如果没有处理过，就“保留”这个消息，等待后面事务确认了再处理。
这里“保留”我也是加了引号，因为mq消费是一条一条按顺序消费，如果中间有一个数据卡住了，后面数据就没法消费了。所以这里“保留”，其实也是消费了，只是他消费到了不确定结果的消息，他是重新投递到了half_topic，来实现“保留”的目的。
来源：https://blog.csdn.net/weixin_43094917/article/details/130242296
0
投稿猜你喜欢
池塘里养：Object；一、设计与原理1、基础案例首先看一个基于common-pool2对象池组件的应用案例，主要有工厂类、对象池、对象三个- 前言开发做得久了，总免不了会遇到各种坑。而在Android开发的路上，『软键盘挡住了输入框』这个坑，可谓是一个旷日持久的巨坑——来来来，我们
- 当时用逆向生成后，实体类中的下划线都被去掉，这时只需要在sqlmap.xml中加以下代码即可。打开mybatis驼峰法则。 <sett
- C#中常涉及到对用户密码的加密于解密的算法，其中使用MD5加密是最常见的的实现方式。本文总结了通用的算法并结合了自己的一点小经验，分享给大家
- 这篇文章主要介绍了springboot日期转换器实现实例解析,文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值,需
- 问题情况:在使用 @TableId(type = IdType.AUTO)之后添加的id数字特别大原因:因为在第一次使用的时候没有加注解 所
- 个人认为单例模式是设计模式中最简单也是最常用的一种，是对有限资源合理利用的一种方式。这个模式看似简单，但是其中蕴含了关于并发、类加载、序列化
- Android开发中会有很多很新奇的交互，比如天猫商城的首页头部的分类，使用的是GridLayoutManager+横向指示器实现的，效果如
- 环境:maven+idea。1. 需要的jar包基本的spring和mybatis依赖包就不说了，在pom文件的build->plug
- 前言最近参加了21天打卡活动，希望可以让自己养成写博客的习惯…ArrayList和LinkedListArrayLis
- JSON字符串和java对象的互转【json-lib】在开发过程中，经常需要和别的系统交换数据，数据交换的格式有XML、JSON等，JSON
- 前言老规矩，还是从最简单粗暴的开始。那么多简单算简单？多粗暴算粗暴？我告诉你可以不写一句代码，你信吗？直接把一个文件往IIS服务器上一扔，就
- 本地异步处理，采用事件机制 可以使 代码解耦，更易读。事件机制实现模式是 观察者模式（或发布订阅模式），主要分为三部分：发布者、监听者、事件
- 现在越来越多的软件都开始使用沉浸式状态栏了，下面总结一下沉浸式状态栏的两种使用方法注意！沉浸式状态栏只支持安卓4.4及以上的版本状态栏：4.
- 目录前言1. 如何让两个线程依次执行？2. 如何让两个线程按照指定的方式有序相交？3. 线程 D 在A、B、C都同步执行完毕后执行4. 三个
- 一、this可以代表引用类的当前实例，包括继承而来的方法，通常可以省略。public class Person{ &n
- 本文实例为大家分享了C语言实现代码雨效果的具体代码，供大家参考，具体内容如下一、项目描述和最终的效果展示项目: 让字符从上到下
- 前言我们之前学的单链表，默认只能从链表的头部遍历到链表的尾部，在实际中应用太少见，太局限；而双向链表，对于该链表中的任意节点，既可以通过该节
- 可重入锁 广义上的可重入锁指的是可重复可递归调用的锁，在外层使用锁之后，在内层仍然可以使用，并且不发生死锁（前提得是同一个对象或者
- 倒序拼接字符串@ApiOperation("分页查询") @GetMapping(value
