全网最深入的RocketMQ Consumer 学习笔记

码农沉思录

共 10976字,需浏览 22分钟

 ·

2021-02-06 12:12


本文来源: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 请求码的 RemotingCommandBroker 查询在线消费者列表,拿到结果后反序列化

三、分配 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 对象,并存入 RebalanceImplprocessQueueTable 队列中。

接着构建 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 的学习之旅,从点到面,还需要继续学习,之后再见👋~


嘿,你在看吗
浏览 46
点赞
评论
收藏
分享

手机扫一扫分享

分享
举报
评论
图片
表情
推荐
点赞
评论
收藏
分享

手机扫一扫分享

分享
举报