RocketMQ之offset确认机制
Posted
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了RocketMQ之offset确认机制相关的知识，希望对你有一定的参考价值。
参考技术A本文要探讨的offset指的是上图中的Queue Offset。
为了保存消费的消费进度，避免重复消费，我们需要将offset保存下来。
针对集群消费，offset保存在broker，在客户端使用RemoteBrokerOffsetStore。
针对广播消费，offset保存在本地，在客户端使用LocalFileOffsetStore。
最后，比较重要的一点是，保存的offset指的是下一条消息的offset，而不是消费完最后一条消息的offset。
比如，你消费了上图中第一个Queue的offset为0的消息，其实保存的offset为1，表示下次我从offset=1的位置进行消费。
在broker端，通过ConsumerOffsetManager中的offsetTable来保存Topic下各个ConsumerGroup的消费进度。
从offsetTable的双层Map结构也是能够看出，我上面说的消费进度，细指为ConsumerGroup在Topic下每个queue的消费进度。
offsetTable毕竟只是内存结构，因此ConsumerOffsetManager继承了ConfigManager实现了持久化功能。
实现了encode，decode，configFilePath三个模板方法。用于指定序列化，反序列化的逻辑以及保存位置
其中序列化，反序列化的逻辑很简单，就是使用到了我们的FastJson。
保存文件名为consumerOffset.json。
broker启动时从本地文件加载
org.apache.rocketmq.broker.BrokerController#initialize
定时触发，持久化到磁盘
org.apache.rocketmq.broker.BrokerController#initialize
BrokerController#shutdown
用于consumer定时同步offset
拉取消息时会顺带确认offset
事务回查触发，暂不深入研究
本文只讨论PUSH模式的集群消费，本地的offset缓存到RemoteBrokerOffsetStore的offsetTable中，定期同步到broker。
因为consumer每次重启都会重新拉取offset，只是一个临时存储，因此RemoteBrokerOffsetStore的offsetTable的设计没有像ConsumerOffsetManager那么复杂。
consumer启动后会进行第一次rebalance，并且之后都会定期rebalance。
在rebalance分配好messagequeue之后，会根据messagequeue生成processqueue进行消息拉取。
而在进行消息拉取前，有一个关键的操作， 拉取对应messagequeue的offset 。
RebalanceImpl#updateProcessQueueTableInRebalance
其中获取消息拉取初始位置有三种策略
CONSUME_FROM_LAST_OFFSET 最新的offset
CONSUME_FROM_FIRST_OFFSET 第一个offset
CONSUME_FROM_TIMESTAMP 根据时间戳获取offset
但是从源码中可以看出来，实际上的逻辑和我们想象的有点不同，上面三个的逻辑的触发前提是，从broker拉取不到offset进度。
这应该是为了防止重复消费以及少消费，毕竟rocketmq是业务相关的mq。
在consumer端，针对offsetTable的更新，当然通过消费消息触发。
ConsumeMessageConcurrentlyService#processConsumeResult
针对并发消费的offset，更新值来源于ProcessQueue#removeMessage方法
removeMessage的逻辑，用到了滑动窗口的算法。
比如10条消息，offset为 0 - 9。
在多线程并发消费的场景下
比如我第一个线程消费了offset为0的消息，那么offsetTable中的offset更新为1
然后我第二个线程消费了offset为5的消息，removeMessage返回的offset还是为1
只有前面的消息全被消费了，窗口才会滑动
ConsumeMessageOrderlyService#processConsumeResult
顺序消费，暂不研究。
最终的offset以broker为准，因此本地的offset要定期持久化到offset。
主要持久化逻辑在persistAll和persist方法。
org.apache.rocketmq.client.consumer.store.RemoteBrokerOffsetStore#persist
persistAll和persist逻辑大致相同，核心逻辑都是通过updateConsumeOffsetToBroker持久化到broker。
触发持久化逻辑的时机有以下4个
MQClientInstance#startScheduledTask
DefaultMQPullConsumerImpl#shutdown
DefaultMQPushConsumerImpl#shutdown
当一个queue不再属于当前consumer的时候，需要同步进步给broker，以便于新拿到queue的consumer从最新未消费的消息开始拉取
RebalancePullImpl#removeUnnecessaryMessageQueue
拉取消息的时候会顺带commit offset
DefaultMQPushConsumerImpl#pullMessage
PullMessageProcessor#processRequest
正常情况下，消息消费失败不会影响窗口滑动，因为针对消费失败的消息，client会进行sendback。
sendback之后，消息经过延迟之后会发往Topic=%RETRY%CONSUMERGROUP的Retry队列
ConsumeMessageConcurrentlyService#processConsumeResult
而sendMessageBack失败的消息，会重新封装成另一个ConsumeRequest在本地再次消费。
这些失败的消息会从之前的consumeRequest移除，因此也就影响到了ProcessQueue#removeMessage的返回值。
但是这是一个优化，重试之后窗口大概率上还是会正常滑动。
如何保证并发消费提交偏移量正确？
基于TreeMap的滑动窗口
如何保证消息消费不丢失？
滑动窗口+broker远端保存+sendback+本地重试兜底
如果broker保存了offset
那么从对应offset重新拉取消息
如果broker没有保存offset，或者其他情况丢失
那么根据配置的策略，从对应的offset开始拉取
RocketMQ广播消费本地Offset文件丢失问题探秘
文章目录
今天本来在用RocketMQ做一个大的功能改造，中间有个小问题随意跟了下源码，突然还发现一个小BUG。一通跟踪调试，虽然最后还是没有解决问题，但是受益匪浅。记录一下。
一、问题出发点
我们知道RocketMQ的消费者有两种消费模式MessageModel，一种是集群消费，一种是广播消费。这两种消费机制最本质的区别在于，集群消费是在Broker端保存各个ConsumeGroup的消费进度Offset，而广播消费则是在Consumer本地记录消费进度Offset。
集群消费用得比较多，之前也认真跟过源码。但是广播消费因为用得比较少，所以基本也跟你一样，背过一点八股文，也就没有太过关注。但是，今天一个偶然的业务场景用到了广播消费。当我想要去本地看一下Offset的记录时，却发现怎么也找不到。这到底是我哪个业务配置配错了？还是万能的八股文欺骗了我？这背后藏着什么样的秘密？
问题1:Offset的本地存储的目录如何进行定制？
在RocketMQ的原生API中，提供了DefaultLitePullConsumer和DefaultMQPushConsumer(还有一个DefaultMQPullConsumer已经过时了)。在这两个消费实例中，都可以设定一个MessageModel属性，这个属性有两个枚举值，MessageModel.BROADCASTING(广播消费)和MessageModel.CLUSTERING(集群消费)。对于广播消费的消费者，默认会在消费者程序所在的机器本地的$user.home/rocketmq_offset/$clientIp@DEFAULT/$group/offset.json文件中保存消费进度。
其中$部分问变量。
$user.home为系统所属用户目录。在windows下默认是C:\\Users\\$用户名。
$clientIp是消费者端的IP地址。
$group是消费者指定的所属消费者组。
这里就有了第一个问题：这个本地存储的目录如何进行定制？
问题2:rocketmq-spring-boot-starter插件如何配置这个本地存储目录？
在使用RocketMQ进行业务开发时，经常会用到SpringBoott框架来整合RocketMQ的客户端。于是，会使用到rocketmq提供的下面的Maven依赖包。
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-spring-boot-starter</artifactId>
<version>2.2.1</version>
</dependency>
然后，就可以使用配置或注解的方式来声明客户端，而不用再去接触RocketMQ的原生消费者API了。例如
@Component
@RocketMQMessageListener(consumerGroup = "MyConsumerGroup", topic = "TestTopic",messageModel= MessageModel.BROADCASTING)
public class SpringConsumer implements RocketMQListener<String>
@Override
public void onMessage(String message)
System.out.println("Received message : "+ message);
工作起来没有问题，但是，当你沿着上面的思路，想要去本地找一找这个消费者的本地缓存时，你会发现一个问题，找不到本地Offset文件(不要怀疑，如果你是用windows开发，一定找不到)。
于是，我开始冒出另外的两个问题：**问题2：rocketmq-spring-boot-starter中有几种声明消费者的方式？**是不是我的使用方式不对？**问题3：rocketmq-spring-boot-starter中是如何指定本地目录地址的？**要怎么找到本地缓存？
什么？你还不会用RocketMQ，不知道我在说什么？试试看看我的RocketMQ教程。顺便支持鼓励一下。
二、问题原因分析
查问题的时候是忙忙碌碌瞎查的，现在回顾问题，就把思路整理一下，也便于你理解。
1、RocketMQ中是如何记录广播消费的本地Offset的？
原生的RocketMQ消费者API是基础，所以这个问题必须先梳理清楚。 以下分析RocketMQ源码用的是4.9.1版本
原生的RocketMQ提供了两个消费者对象，DefaultLitePullConsumer和DefaultMQPushConsumer(还有一个DefaultMQPullConsumer已经过时了)。在这两个消费实例中，都可以设定一个MessageModel属性，这个属性有两个枚举值，MessageModel.BROADCASTING和MessageModel.CLUSTERING。这个属性是怎么跟Offset文件存储对应上的呢？其实从服务的启动过程中就能很清晰的跟踪到。
1.1、看看DefaultMQPushConsumer。
他的start()启动方法中，会启动一个defaultMQPushConsumerImpl实例，而这个defaultMQPushConsumerImpl示例会调用一个同步的start()方法。这个start()是同步的，针对Offset文件存盘问题，是为了保证存盘的Offset文件，在内存中只有一份统一的副本。带着今天的问题，可以在这个start()方法中找到一段很刺眼的代码：
if (this.defaultMQPushConsumer.getOffsetStore() != null)
this.offsetStore = this.defaultMQPushConsumer.getOffsetStore();
else
//从这里可以看出，广播模式与集群模式的最本质区别就是offset存储的地方不一样。
switch (this.defaultMQPushConsumer.getMessageModel())
//广播模式是在消费者本地存储offset
case BROADCASTING:
this.offsetStore = new LocalFileOffsetStore(this.mQClientFactory, this.defaultMQPushConsumer.getConsumerGroup());
break;
//集群模式是在Broker远端存储offset
case CLUSTERING:
this.offsetStore = new RemoteBrokerOffsetStore(this.mQClientFactory, this.defaultMQPushConsumer.getConsumerGroup());
break;
default:
break;
this.defaultMQPushConsumer.setOffsetStore(this.offsetStore);
这里就能看到广播模式和集群模式在Offset存储上的区别。
1.2：再来看看DefaultLitePullConsumer。
在他的启动过程中，类似的会创建一个defaultLitePullConsumerImpl对象。而在defaultLitePullConsumerImpl对象的启动过程中，会调用到一个initOffsetStore()方法。这个方法里的实现也是一样的刺眼：
private void initOffsetStore() throws MQClientException
if (this.defaultLitePullConsumer.getOffsetStore() != null)
this.offsetStore = this.defaultLitePullConsumer.getOffsetStore();
else
switch (this.defaultLitePullConsumer.getMessageModel())
case BROADCASTING:
this.offsetStore = new LocalFileOffsetStore(this.mQClientFactory, this.defaultLitePullConsumer.getConsumerGroup());
break;
case CLUSTERING:
this.offsetStore = new RemoteBrokerOffsetStore(this.mQClientFactory, this.defaultLitePullConsumer.getConsumerGroup());
break;
default:
break;
this.defaultLitePullConsumer.setOffsetStore(this.offsetStore);
this.offsetStore.load();
所以，问题自然就集中到了这个LocalFileOffsetStore中了。
1.3：看LocalFileOffsetStore是如何指定存储路径的
RocketMQ的源码中采用了充血模型的实现方式，所有关于LocalFileOffset的业务动作，都集中到了LocalFileOffsetStore.java当中。在他的构造方法当中就直接维护了一个storePath属性来维护本地存储地址。
public class LocalFileOffsetStore implements OffsetStore
//本地存储目录
public final static String LOCAL_OFFSET_STORE_DIR = System.getProperty(
"rocketmq.client.localOffsetStoreDir",
System.getProperty("user.home") + File.separator + ".rocketmq_offsets");
.....
public LocalFileOffsetStore(MQClientInstance mQClientFactory, String groupName)
.....
//本地存储文件
this.storePath = LOCAL_OFFSET_STORE_DIR + File.separator +
this.mQClientFactory.getClientId() + File.separator +
this.groupName + File.separator +
"offsets.json";
.....
｝
先看LOCAL_OFFSET_STORE_DIR，这是本地存储的根目录。跟rocketmq.client.localOffsetStoreDir这个系统属性有关，**所以要定制本地存储的目录，只需要设定rocketmq.client.localOffsetStoreDir系统属性即可。**而这个系统属性还没有支持前端配置，所以，修改的方式，只能是在应用启动时手动进行指定。 例如 System.setProperty(“rocketmq.client.localOffsetStoreDir”,“D:/.rockemtq_offset”)。
然后，再来看offsets.json的具体存储路径storePath。看这个结构，就跟最开始直接指出的结论对应上了。但是这样其中还有一个变量，就是clinetId。这个属性是如何指定的呢？是不是可以定制呢？
这里面的mQClientFactory是一个MQClientInstance的实例对象。而MQClientInstance则是RocketMQ中对所有客户端的抽象。生产者和消费者最终都会交由MQClientInstance进行统一的服务启动。在MQClientInstance的start()方法中给所有的客户端定义了一个统一的启动标准，不同的客户端只要按照标准去注册不同的信息即可。而ClientId就是MQClientInstance在初始化的过程中指定的一个属性(初始化过程见MQClientManager.java#getOrCreateMQClientInstance方法)。ClientId的具体生成逻辑则是在ClientConfig对象中的buildMQClientId方法中。
//指定instanceName
private String instanceName = System.getProperty("rocketmq.client.name", "DEFAULT");
//构建clientId
public String buildMQClientId()
StringBuilder sb = new StringBuilder();
sb.append(this.getClientIP());
sb.append("@");
sb.append(this.getInstanceName());
if (!UtilAll.isBlank(this.unitName))
sb.append("@");
sb.append(this.unitName);
return sb.toString();
这里面又引出了一个变量instanceName，这个变量对于广播消息的本地Offset存储路径起了很重要的作用。
-
这个instanceName，默认值是DEFAULT。
-
可以由系统属性rocketmq.client.name来替代这个默认值，这个系统属性依然是没有配置属性定制的，只能手动修改。
-
这个instanceName在ClientConfig中还有对应的setter方法。可以由消费者客户端自行指定。而常用的RocketMQ客户端都是ClientConfig的子类，所以，他们都可以通过setter方法定制instanceName。
这个unitMode和unitName不太清楚具体干嘛的，unitMode默认是false，然后在源码中搜索了一下，好像也就在ClientConfig配置本地存储目录时用到了，在核心业务中并没有太多的作用。
github仓库中只有一条关于unitMode的issue，回复也是目前并没有做什么实际的工作。参见https://github.com/apache/rocketmq/issues/639
1.4：Offsets.json文件何时写入？
基于充血模型设计的好处在于，系统内的所有核心业务能力都在充血实体中统一整理，不需要去其他地方到处乱搜。对于Offsets.json文件的核心写入能力，也一起体现在了LocalFileOffsetStore类中。在LocalFileOffsetStore类中，有一个persistAll方法，实现了文件的写入。
public void persistAll(Set<MessageQueue> mqs)
if (null == mqs || mqs.isEmpty())
return;
OffsetSerializeWrapper offsetSerializeWrapper = new OffsetSerializeWrapper();
for (Map.Entry<MessageQueue, AtomicLong> entry : this.offsetTable.entrySet())
if (mqs.contains(entry.getKey()))
AtomicLong offset = entry.getValue();
offsetSerializeWrapper.getOffsetTable().put(entry.getKey(), offset);
String jsonString = offsetSerializeWrapper.toJson(true);
if (jsonString != null)
try
MixAll.string2File(jsonString, this.storePath);
catch (IOException e)
log.error("persistAll consumer offset Exception, " + this.storePath, e);
在这个方法中可以看到，如果offsets.json文件写入失败，RocketMQ只是记录一条log日志就没事了，甚至连异常都没有往外抛。这意味着如果广播消息本地的offsets.json进度没有更新，RocketMQ不会做任何的补救措施。
不过，RocketMQ的客户端会启动一个线程，不断的尝试将这些offset偏移信息写入到文件当中，这也算是一种处理的方式把。
this.scheduledExecutorService.scheduleAtFixedRate(new Runnable()
@Override
public void run()
try
MQClientInstance.this.persistAllConsumerOffset();
catch (Exception e)
log.error("ScheduledTask persistAllConsumerOffset exception", e);
, 1000 * 10, this.clientConfig.getPersistConsumerOffsetInterval(), TimeUnit.MILLISECONDS);
1.5: 针对问题的总结
经过一通整理，回到之前提到的广播消息的offsets文件内容定制的问题，可以总结出这样几个结论：
- 消费者端存储广播消费的本地offsets文件的默认缓存目录是 System.getProperty(“user.home”) + File.separator + “.rocketmq_offsets” ，可以通过定制 rocketmq.client.localOffsetStoreDir 系统属性进行修改。
- 本地offsets文件在缓存目录中的具体位置与消费者的clientIp 和 instanceName有关。其中instanceName默认是DEFAULT，可以通过定制系统属性 rocketmq.client.name 进行修改。另外，每个消费者对象也可以单独设定instanceName。
- RocketMQ会通过定时任务不断尝试本地Offsets文件的写入，但是，如果本地Offsets文件写入失败，RocketMQ不会进行任何的不就，也就是说不会对业务有很大的影响。
2、rocketmq-spring-boot-starter如何创建RocketMQ的消费者？
接下来开始梳理rocketmq-spring-boot-starter对消费者端的封装。梳理封装的过程，其实也是在整理如何高效的使用RocketMQ的原生API，这一对于加深对于RocketMQ原生API的理解是非常有帮助的。如果你也尝试跟着这篇文章一起梳理源码，记得带上之前剔除的小问题。问题的本身相对比较简单，但是带上一个具体的问题去看源码，绝对会让你看源码的感觉不一样。
rocketmq-spring-boot-starter中在RocketMQ消费者这一块，封装了封装了两种模式，Push模式和Pull模式。
- Push模式就是通过@RocketMQMessageListener注解声明一个RocketMQListener接口的实现类来声明一个消费者。Broker推过来的消息，会自行进入其中的onMessage方法进行处理。
- Pull模式是通过内置的RocketMQTemplate对象的receive方法以及一系列的sendAndReceive方法，由消费者端主动去拉取消息。消息拉取过来后，可以通过里面的convert对象转换成所需要的结果对象。
- 在Pull模式中，如果一个restTemplate实例不够用，还可以通过@ExtRocketMQTemplateConfiguration和@ExtRocketMQConsumerConfiguration两个注解来注册额外的rocketmqTemplate实例。一个用来初始化rocketmqTemplate中的消息发送者。例如发送事务消息时，一个restTemplate只能对应一个事务监听逻辑。如果你的项目中有多个事务消息逻辑，就需要注册多个restTemplate实例，来对应不同的事务监听器。另一个用来初始化rocketmqTemplate中的消费者。
老规矩，如果使用还不太熟练，可以看下我的RocketMQ教程。支持鼓励一下。
接下来，就按Push模式和Pull模式分开来梳理。其中，Push模式是分析的重点，因为这是平常用得做多的一种模式。Pull模式用得比较少，就当做是补充的彩蛋了。
2.1、Push模式
Push模式对于@RocketMQMessageListener注解的处理方式，入口在rocketmq-spring-boot-2.2.1.jar中的org.apache.rocketmq.spring.autoconfigure.ListenerContainerConfiguration类中。
怎么找到的？评经验猜以及碰运气。
这个ListenerContainerConfiguration类继承了Spring当中的SmartInitializingSingleton接口，当Spring容器当中所有非懒加载的实例加载完成后，就会触发他的afterSingletonsInstantiated方法进行初始化。在这个方法中会去扫描所有带有注解@RocketMQMessageListener注解的类，将他注册到内部一个Container容器当中。
public void afterSingletonsInstantiated()
Map<String, Object> beans = this.applicationContext.getBeansWithAnnotation(RocketMQMessageListener.class)
.entrySet().stream().filter(entry -> !ScopedProxyUtils.isScopedTarget(entry.getKey()))
.collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
beans.forEach(this::registerContainer);
这里这个Container可以认为是客户端实例的一个容器，通过这个容器来封装RocketMQ的原生API。
registerContainer的方法挺长的，我这里截取出跟今天的主题相关的几行重要的源码：
private void registerContainer(String beanName, Object bean)
.....
//获取Bean上面的注解
RocketMQMessageListener annotation = clazz.getAnnotation(RocketMQMessageListener.class);
...
//检查注解的配置情况
validate(annotation);
String containerBeanName = String.format("%s_%s", DefaultRocketMQListenerContainer.class.getName(),
counter.incrementAndGet());
GenericApplicationContext genericApplicationContext = (GenericApplicationContext) applicationContext;
//将扫描到的注解转化成为Container，并注册到上下文中。
genericApplicationContext.registerBean(containerBeanName, DefaultRocketMQListenerContainer.class,
() -> createRocketMQListenerContainer(containerBeanName, bean, annotation));
DefaultRocketMQListenerContainer container = genericApplicationContext.getBean(containerBeanName,
DefaultRocketMQListenerContainer.class);
//启动容器，这里就相当于是启动了消费者
if (!container.isRunning())
try
container.start();
catch (Exception e)
log.error("Started container failed. ", container, e);
throw new RuntimeException(e);
log.info("Register the listener to container, listenerBeanName:, containerBeanName:", beanName, containerBeanName);
这其中最关注的，当然是创建容器的createRocketMQListenerContainer方法中。而在这个方法中，你基本看不到RocketMQ的原生API，都是在创建并维护一个DefaultRocketMQListenerContainer对象。而这个DefaultRocketMQListenerContainer类，就是我们今天关注的重点。
DefaultRocketMQListenerContainer类实现了InitializingBean接口，自然要先关注他的afterPropertiesSet方法。这是Spring提供的对象初始化的扩展机制。
public void afterPropertiesSet() throws Exception
initRocketMQPushConsumer();
this.messageType = getMessageType();
this.methodParameter = getMethodParameter();
log.debug("RocketMQ messageType: ", messageType);
这个方法就是用来初始化RocketMQ消费者的。在这个方法里就会创建一个RocketMQ原生的DefaultMQPushConsumer消费者。同样，方法很长，抽取出比较关注的重点源码。
private void initRocketMQPushConsumer() throws MQClientException
.....
//检查并创建consumer对象。
if (Objects.nonNull(rpcHook))
consumer = new DefaultMQPushConsumer(consumerGroup, rpcHook, new AllocateMessageQueueAveragely(),
enableMsgTrace, this.applicationContext.getEnvironment().
resolveRequiredPlaceholders(this.rocketMQMessageListener.customizedTraceTopic()));
consumer.setVipChannelEnabled(false);
else
log.debug("Access-key or secret-key not configure in " + this + ".");
consumer = new DefaultMQPushConsumer(consumerGroup, enableMsgTrace,
this.applicationContext.getEnvironment().
resolveRequiredPlaceholders(this.rocketMQMessageListener.customizedTraceTopic()));
// 定制instanceName，有没有很熟悉！！！
consumer.setInstanceName(RocketMQUtil.getInstanceName(nameServer));
.....
//设定广播消费还是集群消费。
switch (messageModel)
case BROADCASTING:
consumer.setMessageModel(org.apache.rocketmq.common.protocol.heartbeat.MessageModel.BROADCASTING);
break;
case CLUSTERING:
consumer.setMessageModel(org.apache.rocketmq.common.protocol.heartbeat.MessageModel.CLUSTERING);
break;
default:
throw new IllegalArgumentException("Property 'messageModel' was wrong.");
//维护消费者的其他属性。
...
//指定Consumer的消费监听 --》在消费以上是关于RocketMQ之offset确认机制的主要内容，如果未能解决你的问题，请参考以下文章
RocketMQ使用之消息保证，重复读，积压，顺序，过滤，延时，事务，死信
Netty协议-Rocket MQ之NettyRemotingClient/Server
Netty协议-Rocket MQ之NettyRemotingClient/Server