RocketMQ之十：RocketMQ消息接收源码

1. 简介

1.1、接收消息

RebalanceService：均衡消息队列服务，负责通过MQClientInstance分配当前 Consumer 可消费的消息队列( MessageQueue )。当有新的 Consumer 的加入或移除，都会重新分配消息队列。
主要实现consumer的负载均衡，但是并不会直接发送获取消息的请求，而是构造PullRequest之后放到pullRequestQueue中，PullMessageService中，等待PullMessageService的线程取出执行。
PullMessageService：拉取消息服务，不断的从 Broker 拉取消息，包含一个需要获取消息的pullRequestQueue（是阻塞的），这个队列的由RebalanceService放PullRequest对象，并不断依次从队列中取出请求向broker send Request。并提交消费任务到 ConsumeMessageService。只有在PUSH模式下才会使用PullMessageService服务线程，该线程主要是对pullRequestQueue:LinkedBlockingQueue<PullRequest>队列进行监测，处理该队列中的PullRequest请求对象；当队列里有PullRequest对象时，从Broker中拉取消息，如果队列为空，则阻塞。同时该线程也提供了两种拉取方式，分别是立即拉取和延迟拉取两种；

ConsumeMessageService：消费消息服务，不断的消费消息，并处理消费结果。
RemoteBrokerOffsetStore:Consumer 消费进度管理，负责从 Broker 获取消费进度，同步消费进度到 Broker。
ProcessQueue ：消息处理队列。
MQClientInstance ：是一个单例模式，封装对 Namesrv，Broker 的 API调用，提供给 Producer、Consumer 使用。
RebalanceImpl：消费端负载均衡的逻辑。该类的调用轨迹如下：(MQClientInstance start --> (this.rebalanceService.start()) --->  RebalanceService.run（this.mqClientFactory.doRebalance()） ---> MQConsumerInner.doRebalance(DefaultMQPushConsumerImpl)  --->RebalanceImpl.doRebalance
在这里着重说明一点：消息队列数量与消费者关系：1个消费者可以消费多个队列，但1个消息队列只会被一个消费者消费；如果消费者数量大于消息队列数量，则有的消费者会消费不到消息（集群模式）

1.2、消息消费的pull和push方式：

对于任何一款消息中间件而言，消费者客户端一般有两种方式从消息中间件获取消息并消费：
（1）Push方式：由消息中间件（MQ消息服务器代理）主动地将消息推送给消费者；采用Push方式，可以尽可能实时地将消息发送给消费者进行消费。但是，在消费者的处理消息的能力较弱的时候(比如，消费者端的业务系统处理一条消息的流程比较复杂，其中的调用链路比较多导致消费时间比较久。概括起来地说就是“慢消费问题”)，而MQ不断地向消费者Push消息，消费者端的缓冲区可能会溢出，导致异常；
（2）Pull方式：由消费者客户端主动向消息中间件（MQ消息服务器代理）拉取消息；采用Pull方式，如何设置Pull消息的频率需要重点去考虑，举个例子来说，可能1分钟内连续来了1000条消息，然后2小时内没有新消息产生（概括起来说就是“消息延迟与忙等待”）。如果每次Pull的时间间隔比较久，会增加消息的延迟，即消息到达消费者的时间加长，MQ中消息的堆积量变大；若每次Pull的时间间隔较短，但是在一段时间内MQ中并没有任何消息可以消费，那么会产生很多无效的Pull请求的RPC开销，影响MQ整体的网络性能；
 
从严格意义上说，RocketMQ并没有实现真正的消息消费的Push模式，而是对Pull模式进行了一定的优化，
一方面在Consumer端开启后台独立的线程—PullMessageService不断地从阻塞队列—pullRequestQueue中获取PullRequest请求并通过网络通信模块发送Pull消息的RPC请求给Broker端。
另外一方面，consumer端后台还有另外一个独立线程—RebalanceService根据Topic中消息队列个数和当前消费组内消费者个数进行负载均衡，将产生的对应PullRequest实例放入阻塞队列—pullRequestQueue中。这里算是比较典型的生产者-消费者模型，实现了准实时的自动消息拉取。然后，再根据业务反馈是否成功消费来推动消费进度。
在Broker端，PullMessageProcessor业务处理器收到Pull消息的RPC请求后，通过MessageStore实例从commitLog获取消息。如1.2节内容所述，如果第一次尝试Pull消息失败（比如Broker端没有可以消费的消息），则通过长轮询机制先hold住并且挂起该请求，然后通过Broker端的后台线程PullRequestHoldService重新尝试和后台线程ReputMessageService的二次处理。

消费消息可以分成pull和push方式，push消息使用比较简单，因为RocketMQ已经帮助我们封装了大部分流程，我们只要重写回调函数即可。

下面我们就以push消费方式为例，分析下这部分源代码流程。

2. 消费者启动流程图

3.消费者类图

消费者类图

4. 消费者源代码流程

consumer启动的时候会启动两个service：
RebalanceService：主要实现consumer的负载均衡，但是并不会直接发送获取消息的请求，而是构造request之后放到PullMessageService中，等待PullMessageService的线程取出执行
PullMessageService：主要负责从broker获取message，包含一个需要获取消息的请求队列（是阻塞的），并不断依次从队列中取出请求向broker send Request

4.1 消费客户端启动

根据官方(https://github.com/apache/rocketmq)提供的例子，Consumer.java里面使用DefaultMQPushConsumer启动消息消费者，如下：

//初始化DefaultMQPushConsumer
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name");
//设置命名服务，参考namesrv的启动
consumer.setNamesrvAddr("localhost:9876");
//设置消费起始位置
consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
//订阅消费的主题和过滤符
consumer.subscribe("TopicTest", "*");
//设置消息回调函数
consumer.registerMessageListener(new MessageListenerConcurrently() {
      @Override
      public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
      ConsumeConcurrentlyContext context) {
      System.out.printf(Thread.currentThread().getName() + " Receive New Messages: " + msgs + "%n");
      return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
      }
});
//启动消费者
consumer.start();

4.2 消息者启动

我们接着看consumer.start()方法

@Override
public void start() throws MQClientException {
     this.defaultMQPushConsumerImpl.start();
}

DefaultMQPushConsumerImpl.java

    public synchronized void start() throws MQClientException {
        switch (this.serviceState) {
            case CREATE_JUST:
                ...

                this.checkConfig();//检查参数

                ...

                this.mQClientFactory = MQClientManager.getInstance().getAndCreateMQClientInstance(this.defaultMQPushConsumer, this.rpcHook);

                ...

                this.pullAPIWrapper = new PullAPIWrapper(
                    mQClientFactory,
                    this.defaultMQPushConsumer.getConsumerGroup(), isUnitMode());
                this.pullAPIWrapper.registerFilterMessageHook(filterMessageHookList);

                if (this.defaultMQPushConsumer.getOffsetStore() != null) {
                    this.offsetStore = this.defaultMQPushConsumer.getOffsetStore();
                } else {
                    //5、消费进度存储offsetStore，广播和集群不同
                    switch (this.defaultMQPushConsumer.getMessageModel()) {
                        case BROADCASTING:
                            this.offsetStore = new LocalFileOffsetStore(this.mQClientFactory, this.defaultMQPushConsumer.getConsumerGroup());
                            break;
                        case CLUSTERING:
                            this.offsetStore = new RemoteBrokerOffsetStore(this.mQClientFactory, this.defaultMQPushConsumer.getConsumerGroup());
                            break;
                        default:
                            break;
                    }
                    this.defaultMQPushConsumer.setOffsetStore(this.offsetStore);
                }
                this.offsetStore.load();

                ...

                this.consumeMessageService.start();

                boolean registerOK = mQClientFactory.registerConsumer(this.defaultMQPushConsumer.getConsumerGroup(), this);
                ...

                mQClientFactory.start();

                this.serviceState = ServiceState.RUNNING;
            ...
        }

        ...
    }

在初始化一堆参数之后，然后调用mQClientFactory.start();

private MQClientInstance mQClientFactory;

那继续看MQClientInstance的start

4.3 MQClientInstance

public void start() throws MQClientException {

        synchronized (this) {
            switch (this.serviceState) {
                case CREATE_JUST:
                    ...
                    // Start request-response channel
                    this.mQClientAPIImpl.start();
                    // Start various schedule tasks
                    this.startScheduledTask();
                    // Start pull service
                    this.pullMessageService.start();
                    // Start rebalance service
                    this.rebalanceService.start();
                    // Start push service
                    this.defaultMQProducer.getDefaultMQProducerImpl().start(false);
                    log.info("the client factory [{}] start OK", this.clientId);
                    this.serviceState = ServiceState.RUNNING;
                    break;
               ...
            }
        }
    }

各行代码的作用就像源代码里面的注释一样，重点看下pullMessageService.start()和rebalanceService.start()
pullMessageService.start()作用是不断从一个阻塞队列里面获取pullRequest请求，然后去RocketMQ broker里面获取消息。
如果没有pullRequest的话，那么它将阻塞。
那么，pullRequest请求是怎么放进去的呢？这个就要看rebalanceService了。

4.4 pullMessageService.start

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

@Override
public void run() {
        while (!this.isStopped()) {
            try {
                PullRequest pullRequest = this.pullRequestQueue.take();
                if (pullRequest != null) {
                    this.pullMessage(pullRequest);
                }
            } catch (InterruptedException e) {
            } catch (Exception e) {
                ..
            }
     }
}

顺便说一句，pullMessageService和rebalanceService都是继承自ServiceThread

public class PullMessageService extends ServiceThread {}

ServiceThread简单封装了线程的启动，调用start方法，就会调用它的run方法。

    public ServiceThread() {
        this.thread = new Thread(this, this.getServiceName()); //把当前对象作为runnable传入线程构造函数
    }

    public void start() {
        this.thread.start();
    }

这样启动线程就要方便一点，看起来舒服一点。

嗯，继续分析之前的分析。

从pullMessageService的run方法可以看出它是从阻塞队列pullRequestQueue里面获取pullRequest，如果没有那么将阻塞。(如果不清楚java阻塞的使用，百度)

执行完一次pullReqeust之后，再继续下一次获取阻塞队列，因为它是个while循环。

所以，我们需要分析下pullRequest放进队列的流程，也就是rebalanceService.

4.5 rebalanceService,消费端负载均衡

关于消费者的Rebalance过程，入口在RebalanceService，这是个线程，默认每隔20s做一次rebalance。

public class RebalanceService extends ServiceThread {
    private static long waitInterval =
        Long.parseLong(System.getProperty(
            "rocketmq.client.rebalance.waitInterval", "20000"));
    @Override
    public void run() {
        while (!this.isStopped()) {
            this.waitForRunning(waitInterval);
            this.mqClientFactory.doRebalance();
        }
    }
}

org.apache.rocketmq.client.impl.factory.MQClientInstance.java

    public void doRebalance() {
        for (Map.Entry<String, MQConsumerInner> entry : this.consumerTable.entrySet()) {
            MQConsumerInner impl = entry.getValue();
            if (impl != null) {
                try {
                    impl.doRebalance();
                } catch (Throwable e) {
                    log.error("doRebalance exception", e);
                }
            }
        }
    }

DefaultMQPushConsumerImpl.java

    @Override
    public void doRebalance() {
        if (!this.pause) {
            this.rebalanceImpl.doRebalance(this.isConsumeOrderly());
        }
    }

RebalanceImpl.java

一路跟下来，来到了RebalanceImpl.java的rebalanceByTopic方法，这个方法里面有两个case(Broadcasting和Clustering)也就是消息消费的两个模式，广播和集群消息。
广播的话，所有的监听者都会收到消息，集群的话，只有一个消费者可以收到，我们以集群消息为例。
先大概解释下在rebalanceByTopic里面要做什么。

  1. 从namesrv获取broker里面这个topic的消费者数量
  2. 从namesrv获取broker这个topic的消息队列数量
  3. 根据前两部获取的数据进行负载均衡计算，计算出当前消费者客户端分配到的消息队列。
  4. 按照分配到的消息队列，去broker请求这个消息队列里面的消息。

广播消息：

    private void rebalanceByTopic(final String topic, final boolean isOrder) {
        switch (messageModel) {
            case BROADCASTING: {
                Set<MessageQueue> mqSet = this.topicSubscribeInfoTable.get(topic);
                if (mqSet != null) {
                    //topicSubscribeInfoTable的更新操作（更新topic对应的MessageQueue）信息，发生在发送消息时（updateTopicRouteInfoFromNameServer方法）
                    boolean changed = this.updateProcessQueueTableInRebalance(topic, mqSet, isOrder);
                    if (changed) {
                        this.messageQueueChanged(topic, mqSet, mqSet);
                        log.info("messageQueueChanged {} {} {} {}",
                            consumerGroup,
                            topic,
                            mqSet,
                            mqSet);
                    }
                } else {
                    log.warn("doRebalance, {}, but the topic[{}] not exist.", consumerGroup, topic);
                }
                break;
            }

接着看

    private boolean updateProcessQueueTableInRebalance(final String topic, final Set<MessageQueue> mqSet,
        final boolean isOrder) {
        boolean changed = false;
        //移除 在processQueueTable && 不存在于 mqSet 里的消息队列
        Iterator<Entry<MessageQueue, ProcessQueue>> it = this.processQueueTable.entrySet().iterator();
        while (it.hasNext()) {
            Entry<MessageQueue, ProcessQueue> next = it.next();
            MessageQueue mq = next.getKey();
            ProcessQueue pq = next.getValue();

            if (mq.getTopic().equals(topic)) {
                if (!mqSet.contains(mq)) {//不包含的队列
                    pq.setDropped(true);
                    if (this.removeUnnecessaryMessageQueue(mq, pq)) {
                        it.remove();
                        changed = true;
                        log.info("doRebalance, {}, remove unnecessary mq, {}", consumerGroup, mq);
                    }
                } else if (pq.isPullExpired()) {//拉取的队列超时，同样清理
                    switch (this.consumeType()) {
                        case CONSUME_ACTIVELY:
                            break;
                        case CONSUME_PASSIVELY:
                            //PUSH模式下，移除拉取超时的
                            pq.setDropped(true);
                            if (this.removeUnnecessaryMessageQueue(mq, pq)) {
                                it.remove();
                                changed = true;
                                log.error("[BUG]doRebalance, {}, remove unnecessary mq, {}, because pull is pause, so try to fixed it",
                                    consumerGroup, mq);
                            }
                            break;
                        default:
                            break;
                    }
                }
            }
        }

继续往下看updateProcessQueueTableInRebalance方法的后半部分，下面是把远端新增的队列加入到processQueueTable中：

for (MessageQueue mq : mqSet) {
    //如果processQueueTable不包括这个mq
    if (!this.processQueueTable.containsKey(mq)) {
        if (isOrder && !this.lock(mq)) {
            log.warn("doRebalance, {}, add a new mq failed, {}, because lock failed", consumerGroup, mq);
            continue;
        }
        //把这个mq的offset先干掉，再添加
        this.removeDirtyOffset(mq);
        ProcessQueue pq = new ProcessQueue();
        long nextOffset = this.computePullFromWhere(mq);
        if (nextOffset >= 0) {
            ProcessQueue pre = this.processQueueTable.putIfAbsent(mq, pq);
            if (pre != null) {
                log.info("doRebalance, {}, mq already exists, {}", consumerGroup, mq);
            } else {
                log.info("doRebalance, {}, add a new mq, {}", consumerGroup, mq);
                PullRequest pullRequest = new PullRequest();
                pullRequest.setConsumerGroup(consumerGroup);
                pullRequest.setNextOffset(nextOffset);
                pullRequest.setMessageQueue(mq);
                pullRequest.setProcessQueue(pq);
                pullRequestList.add(pullRequest);
                //返回是否有变化
                changed = true;
            }
        } else {
            log.warn("doRebalance, {}, add new mq failed, {}", consumerGroup, mq);
        }
    }
}

最后，将pullRequest放到pullRequestQueue中等待去取数据

    // 将pullRequest放在pullRequestQueue中等待去取数据
    this.dispatchPullRequest(pullRequestList);

    return changed;

集群模式的更新队列方式使用的同样是updateProcessQueueTableInRebalance。

消费进度

首先消费者订阅消息消费队列（MessageQueue),当生产者将消息负载发送到MessageQueue中时，消费订阅者开始消费消息，消息消费过程中，为了避免重复消费，需要一个地方存储消费进度（消费偏移量）。
广播模式：每条消息都被每一个消费者消费,使用本地文件的消费进度。
集群模式：一条消息被集群中任何一个消费者消费,使用Broker的消费进度。

广播模式使用本地的消费进度即可，因为消费者之间互相独立，集群模式则不是，正常情况下，一条消息在一个消费者上消费成功(一条消息只能被集群内的一个消费者消费)，则不会发送到其他消费者，所以，进度不能保存在消费端，只能集中保存在一个地方，比较合适的是在Broker端。接下来我们先分析一下消息消费进度接口：OffsetStore.java

在入口代码：DefaultMQPushConsumerImpl#start()的第5点里

 

根据消息消费模式（集群模式、广播模式）会创建不同的OffsetStore方式。
由于上篇文章，谈到广播模式消息，如果返回CONSUME_LATER,竟然不会重试，而是直接丢弃，为什么呢？由于这个原因，这次破天荒的从广播模式的OffsetStore开始学习。
1、LocalFileOffsetStore (广播模式)
消息进度以本地文件方式保存。
源码路径：org.apache.rocketmq.client.consumer.store.LocalFileOffsetStore
1.1、核心属性与构造函数

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");
    private final static InternalLogger log = ClientLogger.getLog();
    private final MQClientInstance mQClientFactory;
    private final String groupName;
    private final String storePath;
    private ConcurrentMap<MessageQueue, AtomicLong> offsetTable =
        new ConcurrentHashMap<MessageQueue, AtomicLong>();

    public LocalFileOffsetStore(MQClientInstance mQClientFactory, String groupName) {

LOCAL_OFFSET_STORE_DIR : offset存储根目录，默认为用户主目录，例如 /home/dingw,可以在消费者启动的JVM参数中，通过-Drocketmq.client.localOffsetStoreDir=路径
groupName : 消费组名称
storePath : 具体的消费进度保存文件名（全路径）
offsetTable 内存中的offfset进度保持，以MessageQueue为键，偏移量为值
继续看一下构造函数：

    public LocalFileOffsetStore(MQClientInstance mQClientFactory, String groupName) {
        this.mQClientFactory = mQClientFactory;
        this.groupName = groupName;
        this.storePath = LOCAL_OFFSET_STORE_DIR + File.separator +
            this.mQClientFactory.getClientId() + File.separator +
            this.groupName + File.separator +
            "offsets.json";
    }

LocalFileOffsetStore 首先在DefaultMQPushConsumerImpl#start方法中创，并执行load方法加载消费进度。

接下来结束一下几个关键的实现方法

1.2 load()方法 
    @Override
    public void load() throws MQClientException {
        OffsetSerializeWrapper offsetSerializeWrapper = this.readLocalOffset();
        if (offsetSerializeWrapper != null && offsetSerializeWrapper.getOffsetTable() != null) {
            offsetTable.putAll(offsetSerializeWrapper.getOffsetTable());

            for (MessageQueue mq : offsetSerializeWrapper.getOffsetTable().keySet()) {
                AtomicLong offset = offsetSerializeWrapper.getOffsetTable().get(mq);
                log.info("load consumer's offset, {} {} {}",
                    this.groupName,
                    mq,
                    offset.get());
            }
        }
    }

 该方法，主要就是读取offsets.json或offsets.json.bak中的内容，然后将json转换成map：

然后更新或获取消息队列的消费进度，就是从内存(Map)或store中获取，接下来看一下初次保存offsets.json文件

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

保存逻辑很简单，就没必要一一分析，重点看一下，该方法在什么时候调用：【MQClientInstance#startScheduledTask】

 

顺藤摸瓜，原因是一个定时任务，默认消费端启动10秒后，每隔5s的频率持久化一次

广播模式消费进度存储容易，但其实还是不明白为什么RocketMQ广播模式，如果消费失败，则丢弃，因为广播模式有时候也必须确保每个消费者都成功消费，，通常的场景为，通过MQ刷新本地缓存等。

2、集群模式消费进度存储（RemoteBrokerOffsetStore）
在阅读RemoteBrokerOffsetStore之前，我们先思考一下如下几个问题：
在集群模式下，多个消费者会负载到不同的消费队列上，因为消息消费进度是基于消息队列进行保存的，也就是不同的消费者之间的消费进度保存是不会存在并发的，但是在同一个消费者，非顺序消息消费时，一个消费者（多个线程）并发消费消息，比如m1 < m2,,但m2先消费完，此时是如何保存的消费进度呢？举个例子，如果m2的offset为5，而m1的offset为4，如果m2先消费完，保存进度为5，那m1消息消费完，保存进度为4，这样岂不乱来了。
2.1 RemoteBrokerOffsetStore 核心属性

public class RemoteBrokerOffsetStore implements OffsetStore {
    private final static Logger log = ClientLogger.getLog();
    private final MQClientInstance mQClientFactory;      // MQ客户端实例，该实例被同一个客户端的消费者、生产者共用
    private final String groupName;                      // MQ消费组
    private ConcurrentMap<MessageQueue, AtomicLong> offsetTable =
        new ConcurrentHashMap<MessageQueue, AtomicLong>();    // 消费进度存储（内存中）
 
    public RemoteBrokerOffsetStore(MQClientInstance mQClientFactory, String groupName) {   // 构造方法
        this.mQClientFactory = mQClientFactory;
        this.groupName = groupName;
    }

2.2 updateOffset 更新offset

@Override
    public void updateOffset(MessageQueue mq, long offset, boolean increaseOnly) {
        if (mq != null) {
            AtomicLong offsetOld = this.offsetTable.get(mq);
            if (null == offsetOld) {     // @1
                offsetOld = this.offsetTable.putIfAbsent(mq, new AtomicLong(offset));    // @2
            }
 
            if (null != offsetOld) {   // @3
                if (increaseOnly) {          
                    MixAll.compareAndIncreaseOnly(offsetOld, offset);   // @4
                } else {
                    offsetOld.set(offset);    // @5
                }
            }
        }
    }

代码@1：如果当前并没有存储该mq的offset,则把传入的offset放入内存中（map)
代码@3：如果offsetOld不为空，这里如果不为空，说明同时对一个MQ消费队列进行消费，并发执行
代码@4，@5，根据increaseOnly更新原先的offsetOld的值，这个值是个局部变量，但这里到底有什么用呢？
2.3 readOffset 根据读取来源，读取消费队列的消费进度

public long readOffset(final MessageQueue mq, final ReadOffsetType type) {
        if (mq != null) {
            switch (type) {
                case MEMORY_FIRST_THEN_STORE:                        // 先从内存中读取，如果内存中不存在，再尝试从磁盘中读取                     
                case READ_FROM_MEMORY: {                             // 从内存中读取
                    AtomicLong offset = this.offsetTable.get(mq);
                    if (offset != null) {
                        return offset.get();
                    } else if (ReadOffsetType.READ_FROM_MEMORY == type) {
                        return -1;
                    }
                }
                case READ_FROM_STORE: {                              // 从磁盘中读取
                    try {
                        long brokerOffset = this.fetchConsumeOffsetFromBroker(mq);     
                        AtomicLong offset = new AtomicLong(brokerOffset);
                        this.updateOffset(mq, offset.get(), false);
                        return brokerOffset;
                    }
                    // No offset in broker
                    catch (MQBrokerException e) {
                        return -1;
                    }
                    //Other exceptions
                    catch (Exception e) {
                        log.warn("fetchConsumeOffsetFromBroker exception, " + mq, e);
                        return -2;
                    }
                }
                default:
                    break;
            }
        }
 
        return -1;
    }

这里主要关注从磁盘中读取消费进度，核心入口方法：fetchConsumeOffsetFromBroker

private long fetchConsumeOffsetFromBroker(MessageQueue mq) throws RemotingException, MQBrokerException,
        InterruptedException, MQClientException {
        FindBrokerResult findBrokerResult = this.mQClientFactory.findBrokerAddressInAdmin(mq.getBrokerName());
        if (null == findBrokerResult) {
 
            this.mQClientFactory.updateTopicRouteInfoFromNameServer(mq.getTopic());
            findBrokerResult = this.mQClientFactory.findBrokerAddressInAdmin(mq.getBrokerName());
        }
 
        if (findBrokerResult != null) {
            QueryConsumerOffsetRequestHeader requestHeader = new QueryConsumerOffsetRequestHeader();
            requestHeader.setTopic(mq.getTopic());
            requestHeader.setConsumerGroup(this.groupName);
            requestHeader.setQueueId(mq.getQueueId());
 
            return this.mQClientFactory.getMQClientAPIImpl().queryConsumerOffset(
                findBrokerResult.getBrokerAddr(), requestHeader, 1000 * 5);
        } else {
            throw new MQClientException("The broker[" + mq.getBrokerName() + "] not exist", null);
        }
    }

这里，主要是首先根据mq的broker名称获取broker地址，然后发送请求，我们重点关注一下消费进度是保存在broker哪个地方：
Broker端的offset管理参照 ConsumerOffsetManager，，保存逻辑其实与广播模式差不多，就不深入研究了，重点说一下offset保存的路径：
/rocketmq_home/store/config/consumerOffset.json

综上所述，我们了解到的情况是，广播模式，存放在消费者本地，集群模式，存储在Broker,存储文件，存放的是JSON。
也就是OffsetStore提供保存消费进度方法，也就是 {“consumeGroup" : [ {”ConsumeQueue1“:offset} ] }

现在我们思考如下问题：下面讨论还是基于非顺序消息：
1、集群模式，一个消费组是多个线程消费该队列中的消息，并发执行，例如在q1中存在 m1,m2,m3,m4,m5
最后消费成功的顺序有可能是 m1,m3,m2,m5,m4,如果消费消息，就将该消息的offset存入offset中，岂不是会乱，如果一批拉取了多条消息，消费进度是如何保存的。要解决上述问题，我们移步到到调用offsetStore.updateStore方法，重点看一下那块逻辑：
ConsumeMessageConcurrentlyService#processConsumeResult

也就是消息处理后，然后移除该批处理消息，然后返回要更新的offset。那我们重点看一下removeMessage方法：

public long removeMessage(final List<MessageExt> msgs) {
        long result = -1;
        final long now = System.currentTimeMillis();
        try {
            this.lockTreeMap.writeLock().lockInterruptibly();
            this.lastConsumeTimestamp = now;
            try {
                if (!msgTreeMap.isEmpty()) {
                    result = this.queueOffsetMax + 1;
                    int removedCnt = 0;
                    for (MessageExt msg : msgs) {
                        MessageExt prev = msgTreeMap.remove(msg.getQueueOffset());
                        if (prev != null) {
                            removedCnt--;
                        }
                    }
                    msgCount.addAndGet(removedCnt);
 
                    if (!msgTreeMap.isEmpty()) {
                        result = msgTreeMap.firstKey();
                    }
                }
            } finally {
                this.lockTreeMap.writeLock().unlock();
            }
        } catch (Throwable t) {
            log.error("removeMessage exception", t);
        }
 
        return result;
    }

主要一下，msgTreeMap的类型，TreeMap,按消息的offset升序排序，返回的result,如果treemap中不存在任何消息，那就返回该处理队列最大的偏移量+1，如果移除自己本批消息后，处理队列中，还存在消息，则返回该处理队列中最小的偏移量，也就是此时返回的偏移量有可能不是消息本身的偏移量，而是处理队列中最小的偏移量。
有点：防止消息丢失（也就是没有消费到）
缺点：会造成消息重复消费


 回来

上面代码里的mqset就是这个topic的消费队列，一般是4个，但是这个值是可以修改的，存储的位置在~/store/config/topics.json里面，比如：

"TopicTest":{
        "order":false,
        "perm":6,
        "readQueueNums":4,
        "topicFilterType":"SINGLE_TAG",
        "topicName":"TopicTest",
        "topicSysFlag":0,
        "writeQueueNums":4
}

可以修改readQueueNums和writeQueueNums为其他值

try {
     allocateResult = strategy.allocate(
                 this.consumerGroup,
         this.mQClientFactory.getClientId(),
         mqAll,
         cidAll);
  } catch (Throwable e) {
         return;
  }

这段代码就是客户端根据获取到的这个topic消费者数量和消息队列数量，使用负载均衡策略计算出当前客户端能够使用的消息队列。
负载均衡策略代码在这个位置。

consumer负载均衡有6种模式：

  • 分页模式(随机分配模式)
  • 手动配置模式
  • 指定机房模式
  • 就近机房模式
  • 统一哈希模式
  • 环型模式

那我们继续4.4 pullMessageService.start分析，因为rebalanceService已经把pullRequest放到了阻塞队列。

4.6 PullMessageService.run

@Override
public void run() {
        while (!this.isStopped()) {
            try {
                PullRequest pullRequest = this.pullRequestQueue.take();
                if (pullRequest != null) {
                    this.pullMessage(pullRequest);
                }
            } catch (InterruptedException e) {
            } catch (Exception e) {

            }
        }
    }

private void pullMessage(final PullRequest pullRequest) {
        final MQConsumerInner consumer = this.mQClientFactory.selectConsumer(pullRequest.getConsumerGroup());
        if (consumer != null) {
            DefaultMQPushConsumerImpl impl = (DefaultMQPushConsumerImpl) consumer;
            impl.pullMessage(pullRequest);
        } else {

        }
    }

调用到DefaultMQPushConsumerImpl.pullMessage(pullRequest)这个方法里面。

4.6.1

public void pullMessage(final PullRequest pullRequest) {
        ...

        final long beginTimestamp = System.currentTimeMillis();

        PullCallback pullCallback = new PullCallback() {
            @Override
            public void onSuccess(PullResult pullResult) {
                System.out.printf("pullcallback onsuccess: " + pullResult + " %n");
                if (pullResult != null) {
                    pullResult = DefaultMQPushConsumerImpl.this.pullAPIWrapper.processPullResult(pullRequest.getMessageQueue(), pullResult,
                        subscriptionData);

                    switch (pullResult.getPullStatus()) {
                        case FOUND:
                            long firstMsgOffset = Long.MAX_VALUE;
                            if (pullResult.getMsgFoundList() == null || pullResult.getMsgFoundList().isEmpty()) {
                                DefaultMQPushConsumerImpl.this.executePullRequestImmediately(pullRequest);
                            } else {

                                DefaultMQPushConsumerImpl.this.consumeMessageService.submitConsumeRequest(
                                    pullResult.getMsgFoundList(),
                                    processQueue,
                                    pullRequest.getMessageQueue(),
                                    dispathToConsume);
                            }
                            break;
                    }
                }
            }

            @Override
            public void onException(Throwable e) {
                                        DefaultMQPushConsumerImpl.this.executePullRequestLater(pullRequest, PULL_TIME_DELAY_MILLS_WHEN_EXCEPTION);
            }
        };

        try {
            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, PULL_TIME_DELAY_MILLS_WHEN_EXCEPTION);
        }
    }

上面这段代码主要就是设置消息获取后的回调函数PullCallback pullCallback，然后调用pullAPIWrapper.pullKernelImpl去Broker里面获取消息。

获取成功后，就会回调pullCallback的onSuccess方法的FOUND case分支。

在pullCallback的onSucess方法的FOUND case分支，会根据回调是同步还是异步，分为两种情况，如下：

同步消息和异步消息区别的源代码实现以后再讲。

参考：https://blog.csdn.net/prestigeding/article/details/79090848

参考：https://www.jianshu.com/p/f071d5069059

【推广】 免费学中医，健康全家人
原文地址：https://www.cnblogs.com/duanxz/p/4706352.html