全网最深入的RocketMQ Consumer 学习笔记
本文来源:http://r6d.cn/Zz4w
学习一下RocketMQ - 消费者的原理和使用🚀
消费模式
消息消费有两种模式:
1、并发消费
并发消费是默认的处理方法,一个消费者使用线程池技术,可以并发消费多条消息,提升机器的资源利用率。默认配置是 20 个线程,所以一台机器默认情况下,同一瞬间可以消费 20 个消息。关注公众后码猿技术专栏获取更多面试资源。
其中 ConsumeMessageConcurrentlyService
的构造函数如下:
public ConsumeMessageConcurrentlyService(DefaultMQPushConsumerImpl defaultMQPushConsumerImpl,
MessageListenerConcurrently messageListener) {
this.defaultMQPushConsumerImpl = defaultMQPushConsumerImpl;
this.messageListener = messageListener;
this.defaultMQPushConsumer = this.defaultMQPushConsumerImpl.getDefaultMQPushConsumer();
this.consumerGroup = this.defaultMQPushConsumer.getConsumerGroup();
this.consumeRequestQueue = new LinkedBlockingQueue();
this.consumeExecutor = new ThreadPoolExecutor(
this.defaultMQPushConsumer.getConsumeThreadMin(),
this.defaultMQPushConsumer.getConsumeThreadMax(),
1000 * 60,
TimeUnit.MILLISECONDS,
this.consumeRequestQueue,
new ThreadFactoryImpl("ConsumeMessageThread_"));
this.scheduledExecutorService = Executors.newSingleThreadScheduledExecutor(new ThreadFactoryImpl("ConsumeMessageScheduledThread_"));
this.cleanExpireMsgExecutors = Executors.newSingleThreadScheduledExecutor(new ThreadFactoryImpl("CleanExpireMsgScheduledThread_"));
}
2、顺序消费
有些业务场景,消息的消费需要顺序性,例如购物时,下订单、库存校验、支付、发送物流,虽然都属于「购物」这个场景的子任务,但他们之间是有顺序性的。如果它们业务处理通过消息解耦,那消息消费也得要有顺序性。
RocketMQ
的做法就是分区有序性,首先需要发送者,将有顺序的消息发往 Topic
下同一个 MessageQueue
,然后消费者,顺序地一个一个进行消费,消费失败将会一直重试,前面消息消费完成才能进行下一个,所以需要在业务上确保消息失败机制,避免消息阻塞。
幂等消费
在 RocketMQ
的设计中,是不保证消息的幂等性,这时候需要业务方自行保证,重复消费消费不会对数据造成影响,从数学意义上来说,f(x) = f(f(x))
,多次计算的结果都是一致的。
RocketMQ
保证存储在 Broker
的消息最少投递一次,该特性保证消息一定会被消费,但由于网络抖动或者其它场景,导致一条消息可能被消费多次。
在相同业务类型的消息中,这里需要考虑两个场景
并发消费 消息消费超时后重复投递
第一个场景很好理解,一条相同类型的消息被不同的消费者同时拉取,可能是不同发送者同时发送的,例如喜闻乐见的 A B
转账问题。关注公众后码猿技术专栏获取更多面试资源。
第二个场景比较难遇到,默认情况,消息处理超过 15 分钟后,将会重新投递消费,如果原来服务器 A
还在处理中,重新投递的消息被服务器 B
拉取了;另一种就是手动重发消息,通过控制台可以重新发送一模一样的消息,MessageID
和消息体跟之前一样,这两种情况下也会造成消息重复消费。
于是设计上,考虑了使用 Redis
做分布式锁,通过竞争锁来避免同时消息,以及用 Redis
暂存消费状态,设计如下:
注意点:
1、锁资源 key 的组装规则(【消费组】+【:】+【主题 topic】+【:】+【messageId 或者 messageKey】
2、锁对应的状态流转(Processing or Successed)
3、避免处理耗时超过锁 expire 时间,导致其它服务器订阅消息并成功消费。加入一个定时线程池,抢到锁资源后,组装定时任务,进行【续时】
4、任务成功后,修改状态为【Successed】,失效时间订为 1h;失败情况,清理掉所有锁资源和定时任务,返回失败重试策略
5、根据 Redis 中保存的状态,过滤重复的消息
在消息 SDK
代码实现上,通过装饰器模式,将 MessageConsumer
包装起来,在业务逻辑不需改动太大情况下,动态增加了幂等消费的功能。
负载均衡
上图展示了,在一个 pullRequestQueue
,可能获取到多个消息 MessageExt
,然后每个消息将会进入消费线程池中消费。
Consumer 端使用 RebalanceImpl
来实现负载均衡,所以想要理解拉取消息的流程,需要重点查看它实现。
Consumer 实例启动时,在工厂 MQClientInstance
中能够看到 new RebalanceService(this);
,启动了一个后台线程,每隔 20s 进行重平衡操作 mqClientFactory.doRebalance()
同样按照消费者的消费模式,重平衡逻辑处理分成两个 switch 分支,接下来讨论的是『并发消费』逻辑
一、获取 MessageQueue 列表
在 RebalanceImpl
维护了一份 map
结构的本地缓存 topicSubscribeInfoTable
,以 topic
维度保存了对应的 MesssageQueue
列表
Set mqSet = this.topicSubscribeInfoTable.get(topic);
二、获取在线的消费者终端列表
List cidAll = this.mQClientFactory.findConsumerIdList(topic, consumerGroup);
findConsumerIdList 方法接受两个参数:Topic 主题和 ConsumerGroup 消费组
底层通过发送 RequestCode.GET_CONSUMER_LIST_BY_GROUP
请求码的 RemotingCommand
到 Broker
查询在线消费者列表,拿到结果后反序列化
三、分配 MessageQueue
根据分配策略,确定当前消费者实例要从哪些 MessageQueue 获取消息
List allocateResult =
strategy.allocate(this.consumerGroup,
this.mQClientFactory.getClientId(),
mqAll,
cidAll);
默认分配策略是平均分配,取当前下标 index
,队列数取余机器数 mod
,然后按照区间给当前应用分配。
例如有 8 个队列,2 台在线服务器,那平均消费 4 个队列,4 4 分配;3 台服务器的,按照 3 3 2 分配。
注意:
目前遇到很多业务团队,在开发过程中,使用了相同的分组名,但是订阅信息不一致,例如之前已经部署了两台应用,本期开发时,新增了 Topic 后,反馈有些消息无法消费,查看 Topic 消费情况表现如下:
根本原因就是前面说的 MessageQueue
平均分配后,之前的应用没有订阅新 Topic
,于是这些消息状态一直处于 Not Consumed Yet
, 解决方法就是统一订阅信息或者更换 ConsumerGroup
进行测试。
四、刷新本地缓存 & 构建请求列表
接下来,会根据前面分配的消息队列,来构建获取消息的请求 pullRequest
队列
“org.apache.rocketmq.client.impl.consumer.RebalanceImpl#updateProcessQueueTableInRebalance
private boolean updateProcessQueueTableInRebalance(final String topic, final Set mqSet, final boolean isOrder) {
boolean changed = false;
Iterator> it = this.processQueueTable.entrySet().iterator();
while (it.hasNext()) {
Entry next = it.next();
MessageQueue mq = next.getKey();
ProcessQueue pq = next.getValue();
....
}
List pullRequestList = new ArrayList();
for (MessageQueue mq : mqSet) {
if (!this.processQueueTable.containsKey(mq)) {
this.removeDirtyOffset(mq);
ProcessQueue pq = new ProcessQueue();
long nextOffset = this.computePullFromWhere(mq);
if (nextOffset >= 0) {
ProcessQueue pre = this.processQueueTable.putIfAbsent(mq, pq);
PullRequest pullRequest = new PullRequest();
pullRequest.setConsumerGroup(consumerGroup);
pullRequest.setNextOffset(nextOffset);
pullRequest.setMessageQueue(mq);
pullRequest.setProcessQueue(pq);
pullRequestList.add(pullRequest);
changed = true;
}
}
}
}
}
this.dispatchPullRequest(pullRequestList);
return changed;
}
@Override
public void dispatchPullRequest(List pullRequestList) {
for (PullRequest pullRequest : pullRequestList) {
this.defaultMQPushConsumerImpl.executePullRequestImmediately(pullRequest);
}
}
public void executePullRequestImmediately(final PullRequest pullRequest) {
this.pullRequestQueue.put(pullRequest);
}
updateProcessQueueTableInRebalance
的作用:更新订阅关系
①、消费节点上下线
②、Topic
的队列分区参数调整
以上两种行为,将会影响到消息订阅的分配,所以需要客户端在消费消息前,先确定自己被分配到哪几个 MessageQueue
,在构建 PullRequest
时,参数中带上监听的 queueId
。
最后,为过滤后的消息队列集合(mqSet)中的每个 MessageQueue
创建一个 ProcessQueue
对象,并存入 RebalanceImpl
的 processQueueTable
队列中。
接着构建 PullRequest
,并调用 dispatchPullRequest
方法,将拉取消息的请求放入到 pullRequestQueue
队列中,等待后面的 PullMessageService
取出来调用。
五、后台线程不停从 Broker
拉取消息
后台线程是:PullMessageService
“org.apache.rocketmq.client.impl.consumer.PullMessageService#run
@Override
public void run() {
while (!this.isStopped()) {
try {
PullRequest pullRequest = this.pullRequestQueue.take();
this.pullMessage(pullRequest);
} catch (InterruptedException ignored) {
} catch (Exception e) {
log.error("Pull Message Service Run Method exception", e);
}
}
}
pullRequestQueue
请求队列,就是前面重平衡服务,构建好放入该队列中的,然后在 PullMessageService
中的 run
方法,使用 while
死循环,不停的去 Broker
请求新消息
六、消息消费
在获取消息时,会注册一个回调接口,具体入口在 MQConsumerInner
,然后在 PullCallback
里调用 messageListener
进行消费,也就是我们写的业务处理逻辑。
在正常消费完成后,将 pullRequest
重新放回拉取消息的任务队列中,等待 PullMessageService
的下一次获取,拉取新消息。
正常消费,业务处理没有异常的话,将会返回 ConsumeReturnType.SUCCESS
表示成功确认,消费位点也能继续前进。
消费失败将会触发补偿机制
ConsumeMessageConcurrentlyService
并发模式下,它会将消息投递到 %retry
队列,更新当前位点,让后面的消息继续消费,如果该消息一直失败,默认最多重试 16 次就会丢到死信队列中。
ConsumeMessageOrderlyService
顺序模式需要注意下,出现失败它不会投递到重试队列,而是将一直在本地重试,直到消费成功为止,所以有可能出现某个 MessageQueue
消费卡住,并且后面消息都不能消费的场景,注意捕获业务处理异常。
“org.apache.rocketmq.client.impl.consumer.DefaultMQPushConsumerImpl#pullMessage
public void pullMessage(final PullRequest pullRequest) {
final ProcessQueue processQueue = pullRequest.getProcessQueue();
...
final SubscriptionData subscriptionData = this.rebalanceImpl.getSubscriptionInner().get(pullRequest.getMessageQueue().getTopic());
// 构建回调接口
PullCallback pullCallback = new PullCallback() {
@Override
public void onSuccess(PullResult pullResult) {
if (pullResult != null) {
pullResult = DefaultMQPushConsumerImpl.this.pullAPIWrapper.processPullResult(pullRequest.getMessageQueue(), pullResult, subscriptionData);
switch (pullResult.getPullStatus()) {
case FOUND:
...
// 这一步消息消费,进入设定的消费逻辑 messageListener
if (pullResult.getMsgFoundList() == null || pullResult.getMsgFoundList().isEmpty()) {
DefaultMQPushConsumerImpl.this.executePullRequestImmediately(pullRequest);
} else {
firstMsgOffset = pullResult.getMsgFoundList().get(0).getQueueOffset();
DefaultMQPushConsumerImpl.this.getConsumerStatsManager().incPullTPS(pullRequest.getConsumerGroup(),
pullRequest.getMessageQueue().getTopic(), pullResult.getMsgFoundList().size());
boolean dispatchToConsume = processQueue.putMessage(pullResult.getMsgFoundList());
DefaultMQPushConsumerImpl.this.consumeMessageService.submitConsumeRequest(pullResult.getMsgFoundList(), processQueue, pullRequest.getMessageQueue(), dispatchToConsume);
// 是否稍后才消费,重新扔回到请求队列中
if (DefaultMQPushConsumerImpl.this.defaultMQPushConsumer.getPullInterval() > 0) {
DefaultMQPushConsumerImpl.this.executePullRequestLater(pullRequest, DefaultMQPushConsumerImpl.this.defaultMQPushConsumer.getPullInterval());
} else {
DefaultMQPushConsumerImpl.this.executePullRequestImmediately(pullRequest);
}
}
break;
...
default:
break;
}
}
}
@Override
public void onException(Throwable e) {
DefaultMQPushConsumerImpl.this.executePullRequestLater(pullRequest, pullTimeDelayMillsWhenException);
}
};
...
try {
// 从 Broker 端获取消息
this.pullAPIWrapper.pullKernelImpl(
pullRequest.getMessageQueue(),
subExpression,
subscriptionData.getExpressionType(),
subscriptionData.getSubVersion(),
pullRequest.getNextOffset(),
this.defaultMQPushConsumer.getPullBatchSize(),
sysFlag,
commitOffsetValue,
BROKER_SUSPEND_MAX_TIME_MILLIS,
CONSUMER_TIMEOUT_MILLIS_WHEN_SUSPEND,
CommunicationMode.ASYNC,
pullCallback
);
} catch (Exception e) {
this.executePullRequestLater(pullRequest, pullTimeDelayMillsWhenException);
}
}
获取消息的这个方法中,有两个核心部分
①、构建消费回调函数
②、从 Broker
端获取新消息
回调接口中,设定了对新消息的处理逻辑,包括顺序消息的特殊处理,还有是否需要等待一段时间才消费,真正执行业务方设定的消费逻辑在 DefaultMQPushConsumerImpl.this.executePullRequestImmediately(pullRequest)
或 DefaultMQPushConsumerImpl.this.consumeMessageService.submitConsumeRequest
中。
然后将回调函数作为参数,放入 this.pullAPIWrapper.pullKernelImpl
方法中,接收消息后,执行回调函数来处理消息。
到这一步为止,从消息获取到消息消费,执行本地业务逻辑的基本流程就基本了解清楚,后面的状态确认以及位点 offset
更新,感兴趣的可以再去跟踪一下。
小结
消费者的深入学习分成以下几部分
消费模式 幂等消费概念 负载均衡
记录了并发模式和广播模式的区别,使用上需要注意的细节。
跟大家分享了一下在原生 RMQ 不支持幂等消费,同时不需要业务方做过多改造的情况下,通过封装 SDK,在里面实现幂等消费的方案。
最后梳理了一下消费者如何重平衡、构建拉取消息的请求最后消费消息的代码过程。
其中还有很多细节点还需要去了解,例如重平衡 doReblance
阶段,出现服务器上下线,还处于消费的 MessageQueue
如何处理(看了一下有加锁 lock 操作,避免两台服务器同时操作同一个队列)的代码如何实现等等
最后,MQ 的学习之旅,从点到面,还需要继续学习,之后再见👋~