RocketMQ消费者消费消息核心原理（含长轮询机制）-阿里云开发者社区

RocketMQ消费者消费消息核心原理（含长轮询机制）

2024-08-16 796

版权

本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。

简介： 这篇文章深入探讨了Apache RocketMQ消息队列中消费者消费消息的核心原理，特别是长轮询机制。文章从消费者和Broker的交互流程出发，详细分析了Push和Pull两种消费模式的内部实现，以及它们是如何通过长轮询机制来优化消息消费的效率。文章还对RocketMQ的消费者启动流程、消息拉取请求的发起、Broker端处理消息拉取请求的流程进行了深入的源码分析，并总结了RocketMQ在设计上的优点，如单一职责化和线程池的使用等。

前言

在消息系统中，消费者消费消息有拉和推消息两种实现方式，拉消息是消费者主动向消息服务器发送拉消息请求，消息服务器将消息返回给消费者，而推消息是消息服务器主动向消费者推送消息的形式，这两种消息消费实现各有各的优势和劣势。

在RocketMQ中，有两种消费者客户端,一种是Push模式消费者, 一种是Pull模式消费者，这两个其实都是表现，在RocketMQ底层实现中采用长轮询的机制来实现消息拉取消费功能。

长轮询模式兼顾了拉和推消息的优势。本文将从源码层面分析RocketMQ的消费者角色的工作原理。从整体看下org.apache.rocketmq.client.consumer.DefaultMQPushConsumer消费者实现类在消费客户端的启动流程，RocketMQ的消费者客户端即支持拉又支持推，但是底层都是基于长轮询的实现。

既然内核底层是长轮训模式，为什么RocketMQ会有Push,Pull客户端呢？

Push只是在客户端，RocketMQ客户端组件将消息拉取到本地后，自动回调我们业务方的业务逻辑执行消费。

Pull模式，其实也是Rocketmq会将消息拉取到本地，我们的业务代码主动从本地读消息进行消费。

这两种客户端对我们开发人员更加友好，可以自己选择最合适的消费方式。

Push消费者启动流程

看源码前，我们可以思考下，消费者能够从broker拉取到消息主要需要做哪些事情？

1、首先肯定需要和Broker创建连接，RocketMQ底层使用Netty组件通信，肯定会使用Netty和Broker建立连接。

2、和Broker建立连接之后，需要告诉Broker拉取Topic下哪个队列的消息，这里就需要用到消费者负载均衡机制了。

3、向Broker拉取消息，还有一个关键的属性，那就是消息拉取偏移量offset, 消费者需要知道下次从哪个偏移量开始拉取消息。在RocketMQ中广播消息模式是将偏移量存储在消费者本地, 集群消息模式是将消息偏移量存储在Broker。因此消费者启动时还需要加载好各个topic的偏移量

分析了上面主线逻辑之后，我们可以看看源码是如何做的。

消费者启动流程在源码 org.apache.rocketmq.client.impl.consumer.DefaultMQPushConsumerImpl#start

加载偏移量存储服务。

偏移量根据消费模式有两个策略，如果是广播��息偏移量存储在本地，如果是集群消息则存储在broker。

消费者调用业务逻辑执行消费逻辑

消费者实例启动，里面的主逻辑非常清晰。

org.apache.rocketmq.client.impl.factory.MQClientInstance#start

public void start() throws MQClientException {
   
   

                    this.serviceState = ServiceState.START_FAILED;
                    // If not specified,looking address from name server
                    if (null == this.clientConfig.getNamesrvAddr()) {
   
   
                        this.mQClientAPIImpl.fetchNameServerAddr();
                    }
                    // 启动netty客户端，和Broker建立通信连接
                    this.mQClientAPIImpl.start();
                    // 定时一些任务
                    this.startScheduledTask();
                    // 启动拉消息的服务，通过线程异步拉取
                    this.pullMessageService.start();
                    // 启动负载均衡服务
                    this.rebalanceService.start();
                    // 启动内置消息发送服务 this.defaultMQProducer.getDefaultMQProducerImpl().start(false);
                    log.info("the client factory [{}] start OK", this.clientId);
                    //更新消费者启动状态
                    this.serviceState = ServiceState.RUNNING;

}

通过上面的步骤消费者实例就启动完成了。

我们发现，消费者客户端主要包含了几个核心组件服务

1、偏移量加载服务 => 记录消息消费进度

2、消息拉取服务 => 建立连接，拉取消息

3、负载均衡服务 => 分配消费队列

4、消息消费服务 => 回调消费消息业务逻辑

和我们一开始的想的没有很大差别。这里体现了单一职责设计原则的设计思想,不同的类专门负责各自不同的职责。

看到这里，我们就把消费者客户端启动过程的主线捋清楚了。

客户端拉取消息请求发起

我们接着看最主要的流程，拉取消息服务PullMessageService，看看他做了哪些事情。

点进去看发现他是一个线程任务。自然我们就会想到去看他的run方法实现，因此我们可以猜测，他是通过异步拉取消息的。

@Override
    public void run() {
   
   
        log.info(this.getServiceName() + " service started");

        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);
            }
        }

        log.info(this.getServiceName() + " service end");
    }

从阻塞队列拿拉数据请求

private final LinkedBlockingQueue<PullRequest> pullRequestQueue = new LinkedBlockingQueue<PullRequest>();

每个拉取请求里包含了对应的消费者组，消息队列，拉取偏移量，从这些信息就可以知道要拉取哪个topic下面哪个队列，哪个偏移量开始的消息了。

public class PullRequest {
   
   
    private String consumerGroup;
    private MessageQueue messageQueue;
    private ProcessQueue processQueue;
    private long nextOffset;
    private boolean lockedFirst = false;