RocketMQ架构之消息收发过程详解

上一篇文章RocketMQ架构设计之启动过程解析分析了 NameServer, Broker, Producer, Consumer 四个部分的启动过程, 这边文章我们分析 消息的收发过程.

生产者发送消息过程解析

我们还是看上篇文章中的例子:

// 使用GroupName初始化Producer
DefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name");
// 指定NameSrv的地址: 也可以通过环境变量NAMESRV_ADDR来指定，则不需要下面这一行。
producer.setNamesrvAddr("name-server1-ip:9876;name-server2-ip:9876");
// 启动实例
producer.start();
try {
    // 创建消息实例,指定 topic, tag, message body.
    Message msg = new Message("TopicTest" /* Topic */,
        "TagA" /* Tag */,
        ("Hello RocketMQ !!").getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */
    );
    // 发送消息给Broker
    SendResult sendResult = producer.send(msg);
    System.out.printf("%s%n", sendResult);
} catch (Exception e) {
    e.printStackTrace();
    Thread.sleep(1000);
}
// 关闭生产者
producer.shutdown();

在上一篇文中，我们是看到了 producer.start();, 现在我们接着来看 producer.send(msg); 的过程。上一篇文章中，有简单的介绍过消息的发送分为三步：

获取 topic 的元数据信息,包括：路由信息，是否为有序队列，MessageQueue列表，消息会发送的队列的索引。
选择 MessageQueue
发送消息

具体实现的代码在 org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl#sendDefaultImpl方法中。

在消息的发送过程中，首先会校验当前Producer的运行状态，然后判断消息的group,topic等等是否为空。合法之后，会获取topic的MessageQueue列表。从messageQueue列表中查找出一个发送的messageQueue. 通过什么的机制选取MessageQueue呢？这里会涉及到一个 Broker的延迟容错机制，下面我们会详解的介绍这个机制。当消息第一次过来的时候，会通过Random随机选择一个messageQueue，进行发送，如果发送失败，会将该messageQueue 隔离起来，重试的时候，不会给这个 messageQueue 投递消息。选择出来了MessageQueue之后，进行发送消息，发送消息首先会根据BrokerName获取broker的地址，执行几个 RpcHook, 然后封装请求头，根据我们发送方式的不同，会选择对应的请求命令(Send_Message,Send_Reply_Message,Send_Reply_Message_V2等等)发送，消息发送完成之后，会更新Producer端的 broker容错表。这样消息就发送出去了。

延迟容错机制

RocketMQ 发送消息的时候，由于 nameServer 检测 Broker 是否存活是有延迟的，在选择消息队列的时候，也可能遇到宕机的Broker, 因网络问题发送失败的不时会发生，因次 RocketMQ 实现这种高可用的设计方案。

到底是怎么回事儿呢?

我们从代码中看看端倪。

文中图片我标注了两处代码

重试

当消息是同步方式发送的时候, 根据配置的属性进行重试，默认为3次。如果是单向方式或者异步的方式进行发送时，则就会发送一次，不会重试。可以看到确实可以从一定程度上提高消息发送成功的几率，但是，如果是broker宕机，重试只会增加时长。

先不急着解决，我们看一下消息第一次发送是怎样的过程。图中2处，会根据 topicPublishInfo 取出一个MessageQueue。跟进代码：

然后接着进入代码

我们先假设 sendLatencyFaultEnable = false. 我们再看一下具体实现:

其中 sendWhichQueue 字段就是上一次消息发送的MessageQueue List的下标。

最后就是根据 selectOneMessage() 方法通过轮询的方式来选MessageQueue了。第一次是通过随机的方式来选出第一个 MessageQueue的，重试的都是轮询的去取了。

那当: sendLatencyFaultEnable = true 的时候呢?

sendLatencyFaultEnable 是什么意思呢？它表示是否启用 Broker故障延迟机制。看下图中的代码，我们假设启用了 Broker故障延迟机制。

图中标注的部分上面的逻辑和刚才选择MessageQueue的机制是一样的。那奥秘肯定是在下面的代码中了(图中框选的部分)

latencyFaultTolerance 是什么呢? latencyFaultTolerance 是一个接口，直译就是：延迟容错。

/**
 * 延迟容错
 *
 * @param <T>
 */
public interface LatencyFaultTolerance<T> {
    /**
     * 更新失败的Broker。
     *
     * @param name                 brokerName
     * @param currentLatency       本次消息发送的时长
     * @param notAvailableDuration 规避broker的时长
     */
    void updateFaultItem(final T name, final long currentLatency, final long notAvailableDuration);

    /**
     * 判断Broker是否可以用
     *
     * @param name broker的名称
     * @return
     */
    boolean isAvailable(final T name);

    /**
     * 从Broker故障列表中移除
     * @param name
     */
    void remove(final T name);

    /**
     * 弹出最后一个 Broker
     * @return
     */
    T pickOneAtLeast();
}

我们来看一下 latencyFaultTolerance.isAvailable(mq.getBrokerName()) 的实现

@Override
public boolean isAvailable(final String name) {
    // faultItemTable 是一个 ConcurrentHashMap
    final FaultItem faultItem = this.faultItemTable.get(name);
    if (faultItem != null) {
        // 判断该Broker(messageQueue)是否在规避时间内。
        return faultItem.isAvailable();
    }
    return true;
}

faultItem.isAvailable() 的实现如下所示：

public boolean isAvailable() {
    return (System.currentTimeMillis() - startTimestamp) >= 0;
}

从这两段代码中可以看出，判断 Broker是否可用的判断依据就是：当前时间是否大于 startTimeStamp。startTimeStamp标识该Broker可以开始使用的时间。

看到这里，一头雾水。分区容错到底是什么东西呢?

我们返回上一层，从框选的下面接着看：

如果还没有选出一个可用 MessageQueue，则会中故障的Broker列表中选出一个Broker，判断这个broker是否有MessageQueue，如果有则随机(轮询)的选一个，如果没有则从故障的Broker列表中移除一个，使用默认的机制选择一个MessageQueue投送消息。

选择出来MessageQueue之后，就会发送消息了。

发送完消息还进行了一个操作

 // 更新topic的一个MessageQueue 是否可用
this.updateFaultItem(mq.getBrokerName(), endTimestamp - beginTimestampPrev, false);

这是干什么呢？我们直译过来更新故障(错误)的条目。去看一下代码实现:

可以看到最终是调用了 MqFaultStrategty 的 updateFaultItem() 方法。这个方法中执行了两个方法，computeNotAvailableDuration 计算不可用的时长，和 latencyFaultTolerance的updateFaultItem方法。

上图中贴出了 computeNotAvailableDuration 方法的实现。方法中使用了两个数组 latencyMax 和 notAvailableDuration ,这两个数组就是延迟机制的核心了. 数组的定义如下:

/**
  * 延迟机制 - 本次消息发送时长区间，用于映射计算出需要规避Broker的时长。
  */
private long[] latencyMax = {50L, 100L, 550L, 1000L, 2000L, 3000L, 15000L};
/**
  * 延迟机制 - 规避Broker的时长
  */
private long[] notAvailableDuration = {0L, 0L, 30000L, 60000L, 120000L, 180000L, 600000L};

方法中的形参 currentlatency 为本次消息发送的延迟时间。

调用computeNotAvailableDuration方法的实参 isolation 表示该Broker是否需要规避。如果消息发送成功，isolation就是false, 反之，就是代表Broker需要被规避。从 computeNotAvailableDuration 方法的实现可以看出: 其会根据本校消息发送的延迟时间从 latencyMax 数组尾部找到第一个比 currentlatency 小的数组下标i，没有则返回0，将 notAvailableDuration[i] 进行返回。当isolation=true的时候，就会默认有10分钟的规避时长，反之，规避时长就会消息发送的延时是多少了。

我们接着看 latencyFaultTolerance的updateFaultItem方法

图中对代码进行注释，不过多解释了。其中 FaultItem 是存储故障broker的类，可以称为失败条目，每个FaultItem 存储了broker的名称，消息发送延迟的时长，故障规避结束的开始时间。

那么更新这两个字段的值有什么作用呢? 不知你是否有印象，刚才说有一头雾水的地方。

在选择MessageQueue的时候，会判断messageQueue是否可用，调用了故障容器接口的 isAvalilable 方法，方法就是根据 startTimestamp 来进行判断改队列是否可用的。

public boolean isAvailable() {
  return (System.currentTimeMillis() - startTimestamp) >= 0;
}

如果当前的系统时间大于故障规避开始时间，说明broker可以加入轮询的队伍里。

以上就是生产者发送消息整个过程了。

生产者总结

我们知道了生产者发送消息，三步完成，1.获取topic的元数据信息，比如路由信息，MessageQueue队列列表等， 2.选择合适的MessageQueue，我们也着重介绍了 Broker的故障延迟机制，这也是RocketMQ实现高可用的方式之一。3.发送消息，这里我们并没有过多的介绍，使用nettyClient 将 Message 发送过 Broker端。

Broker 存储消息

上面我们分析了生产者发送消息的整体过程，尤其着重的分析了生产者选择MessageQueue时候的容错机制。生产者会把消息发送给Broker，进行存储，存储是一个怎样的流程呢?

我直接进入主题，从 SendMessageProcessor 说起。

为什么上来就说这个类呢? 上篇文章中，我们分析了Broker的启动过程中，可以知道的 Broker启动了NettyServer等待生产者，消费者发送来请求。这个 SendMessageProcessor 就是Broker接受生产者发送消息的处理器。

具体的代码在 org.apache.rocketmq.broker.BrokerController#registerProcessor中。

根据之前的内容，我们已知发送的消息有事务消息，批量消息，单条的普通消息三种类型，并且 RocketMQ 支持同步存储和异步存储两种方式。所以肯定是Broker端提供了这六种消息的存储方式。

实际上，也确实支持了这6种方式。我们先从 SendMessageProcessor 的类继承关系来看一下 Broker是如何进行消息存储，我也会按照这样的思路来行文。

接收同步消息

同步消息处理的本质上还是异步的处理方式。

 response = asyncProcessRequest(ctx, request).get();

接着往下看

异步消息处理

public CompletableFuture<RemotingCommand> asyncProcessRequest(ChannelHandlerContext ctx,
        RemotingCommand request) throws RemotingCommandException {
        final SendMessageContext mqtraceContext;
        switch (request.getCode()) {
            // 消费者发送回来的消息(可能是消费失败之后的?)
            case RequestCode.CONSUMER_SEND_MSG_BACK:
                return this.asyncConsumerSendMsgBack(ctx, request);
            default:
                // 处理生产者发送来的消息
                SendMessageRequestHeader requestHeader = parseRequestHeader(request);
                if (requestHeader == null) {
                    return CompletableFuture.completedFuture(null);
                }
                mqtraceContext = buildMsgContext(ctx, requestHeader);
                // 消息发送前的钩子。在消息发送之前，执行一些操作。
                this.executeSendMessageHookBefore(ctx, request, mqtraceContext);
                if (requestHeader.isBatch()) {
                    // 如果是批量消息
                    return this.asyncSendBatchMessage(ctx, request, mqtraceContext, requestHeader);
                } else {
                    // 不是批量消息
                    return this.asyncSendMessage(ctx, request, mqtraceContext, requestHeader);
                }
        }
    }

可以看到，Broker不仅处理来自生产者发送来的消息，还需要处理消费者消费失败重试的消息。消费者重试的消息这部分比较复杂，我们后面再分析。我们先看处理生产者发送来的消息这部分。

代码中显示，会判断消息的类型，是否是批量消息，分别调用不同的方法，进行存储。如下图。

上图中不是完整的代码，我删除了一些不重要的代码，可以看出来，不管是批量消息还是单条消息(除事务消息)都是交给了 MessageStore 进行存储的。当然，事务消息是交给了 TransactionalMessageService 进行存储的，这个内容我们在分析RocketMQ事务消息的时候，会着重的分析，本文中就不做分析了。

继续跟进代码

MessageStore 只是对消息和 Broker 进行了简单的校验，并没有真正的去存储消息。而是把消息交给了 commitLog 进行存储消息。

我们以单条消息为例，来看一下 commitLog 将消息存储到文件中的具体流程。

如上图中， commitLog 收到消息，首先会选择一个MappedFile, 这个 MappedFile 代表磁盘上的物理文件在内容中的映射。指代 $HOME/store/commitlog 目录下的文件。也代表了消息要写入的文件。MappedFile 会把消息写到缓冲区中(即右上图)，消息顺利写完之后，就触发刷盘动作，刷盘有两种方式：同步写盘和异步刷盘。同步写盘是指等待 FlushCommitLogService 把消息真正写到物理磁盘上，然后再返回。异步刷盘是指唤醒 FlushCommitLogService 执行的线程，不等待其是否真正把消息写入了物理磁盘中。最后，会把消息同步到其他的副本服务中。

存储消息流程总结

Broker收到生产者发送的消息，会由 SendMessageProcessor 进行接收，然后会根据消息的类型单条消息，批量消息，还是事务消息，分别进行不同的处理。当然也会区分异步存储还是同步的存储消息。方法实现上本质还是调用异步的方法来实现的，SendMessageProcessor 会把消息转交给 CommitLog 进行校验，会检查当前Broker和当前消息的状态，然后交给MappedFile真正的存储消息，MappedFile表示存储消息的物理文件在内存中的映射，首先会把消息的内容写到缓冲区中，然后调用刷盘的方式(同步写盘或者异步刷盘)将消息真正的写到物理磁盘上。然后，将消息同步给其他副本。

再探讨一点细节

获取MappedFile

在Broker将消息写入 MappedFile 之前，就需要需要获取到一个MappedFile 进行写入。如何获取的呢?

在 asyncPutMessages 方法中，有两处获取 MappedFile 的代码。

第一处代码，倒是没有复杂的逻辑

public MappedFile getLastMappedFile() {
    MappedFile mappedFileLast = null;
    while (!this.mappedFiles.isEmpty()) {
        try {
            mappedFileLast = this.mappedFiles.get(this.mappedFiles.size() - 1);
            break;
        } catch (IndexOutOfBoundsException e) {
            //continue;
        } catch (Exception e) {
            log.error("getLastMappedFile has exception.", e);
            break;
        }
    }
    return mappedFileLast;
}

第二处代码，稍有猫腻。

public MappedFile getLastMappedFile(final long startOffset) {
    return getLastMappedFile(startOffset, true);
}

创建文件的代码和执行过程如上图所示，代码有删减。

解释一下猫腻: 创建 MappedFile 的时候，一次会创建两个MappedFile, 这样下次就可以直接通过一处的代码获取到 MappedFile，而不用再去创建 MappedFile 了。由于真实的创建MappedFile 是由后台 AllocateMappedFileService 线程来创建的，所以也不会浪费时间。

代码显示，会把 AllocateRequest 分别放到 requestTable 和 requestQueue 中。AllocateRequest 是创建 MappedFile 文件请求， requesetTable 是一个 concurrentHashMap , key是文件的路径，Value是 AllocateRequest。存储了路径和AllocateRequest的映射关系。requestQueue，是请求的队列，PriorityBlockingQueue，后台AllocateMappedFileService线程会requestQueue取出一个请求，分配的时候有两种策略，一种是使用Mmap的方式来构建MappedFile实例，另外一种是从TransientStorePool堆外内存池中获取相应的DirectByteBuffer来构建MappedFile.

Consumer 消费消息

在上一篇文章中，我们介绍了消费者有两种实现: DefaultLitePullConsumerImpl 和 DefaultMQPushConsumerImpl. 我们先暂时抛开这两种具体的实现方式。从稍微高一点的角度来看一下消息者消费消息的逻辑。从 消费者注册, 负载均衡，拉取消息，消费消息 这个思路来行文。

消息者注册

消费者启动后，每隔10ms会从 NameServer 查询一次用户订阅的所有话题路由信息.

/** 每10ms从NameServer拉取一下 TopicRouteInfo*/
this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
    @Override
    public void run() {
        try {
            MQClientInstance.this.updateTopicRouteInfoFromNameServer();
        } catch (Exception e) {
            log.error("ScheduledTask updateTopicRouteInfoFromNameServer exception", e);
        }
    }
}, 10, this.clientConfig.getPollNameServerInterval(), TimeUnit.MILLISECONDS);

更新topic路由信息在上篇文中已经介绍过了，这里就不详细的去介绍了，只是简单的串一下消息者消费消息的流程。

除此之外，还有一下操作：

每秒发送一次心跳检测.
每10s 持久化一次 Consumer 的offset. 说到这类，在启动的时候也会加载指定目录下的 consumeOffset.
每分钟扫描 consumerTable 自动调整线程池大小

消费者负载均衡

在消费者启动过程中，会启动 RebalenceService. 看名字就知道这是一个负载均衡服务。什么作用呢? 在任何一个消费者启动，关闭，重置消费偏移的时候，都会触发这个这个服务的重平衡操作。

// 本机 启动时=>其他的消费者会受到 consumerIds发生变更的请求。
/** 5.启动负载均衡服务 org.apache.rocketmq.client.impl.factory.MQClientInstance#start*/
this.rebalanceService.start();
// 其他的消息者启动或者关闭时,即 consumerId 发生变动的时候 org.apache.rocketmq.client.impl.ClientRemotingProcessor#processRequest
case RequestCode.NOTIFY_CONSUMER_IDS_CHANGED:
    return this.notifyConsumerIdsChanged(ctx, request);
// 控制端 重置消费偏移量时 org.apache.rocketmq.client.impl.ClientRemotingProcessor#processRequest
case RequestCode.RESET_CONSUMER_CLIENT_OFFSET:
    return this.resetOffset(ctx, request);

重平衡操作发生了什么?

我们都知道在Broker端管理消息的最小单位是MessageQueue ,而不是 topic。比如当你发送了 10 条相同话题的消息，这 10 条话题可能存储在了不同 Broker 服务器的不同队列中。所以当我们讨论消息队列负载均衡的时候，就是在讨论服务器端的所有队列如何给所有消费者消费的问题。在 RocketMQ 中，客户端有两种消费模式，一种是广播模式，另外一种是集群模式。

我们现在假设总共有两台 Broker 服务器，假设用户使用 Producer 已经发送了 8 条消息，这 8 条消息现在均衡的分布在两台 Broker 服务器的 8 个队列中，每个队列中有一个消息。现在有 3 台消费者都订阅了 Test 话题的消费者实例，我们来看在不同消费模式下，不同的消费者会收到哪几条消息。

广播模式

这种方式下，每台消费者都会收到8条消息。所以，在广播的模式下，重平衡操作执行如下所示:

// 消费者的模式: 广播还是集群模式
switch (messageModel) {
    // 广播方式。
    case BROADCASTING: {
        // topicSubscribeInfoTable: 存储topic 对应的 MessageQueue 列表.
        Set<MessageQueue> mqSet = this.topicSubscribeInfoTable.get(topic);
        if (mqSet != null) {
            // 更新 消费者消费订阅表。返回是否发生了变动。
            boolean changed = this.updateProcessQueueTableInRebalance(topic, mqSet, isOrder);
            if (changed) {
                // 如果发生了变动，则根据不同的消费消费方式处理：
                // push,则需要通知给Broker，更新各个消费者的订阅关系。
                // pull,则会通知消费者，执行消费逻辑。
                this.messageQueueChanged(topic, mqSet, mqSet);
                log.info("messageQueueChanged {} {} {} {}", consumerGroup, topic, mqSet, mqSet);
            }
        } else {
            log.warn("doRebalance, {}, but the topic[{}] not exist.", consumerGroup, topic);
        }
        break;
    }
    case ....:
    // 省略..
}

如上面代码中的注释，主要注意的是，广播模式下，messageQueue发生变动，需要根据不同的消费方式(pull/push),执行不同的逻辑。

push方式：会更新消息消费订阅表的版本，然后会通知Broker，触发consumerId变动。
pull方式: 会重新pull消息，执行pullTask.

集群模式

集群模式是指所有的消息队列会按照某种分配策略来分给不同的消费者客户端。

集群方式，不就是多个消费者协同的消费一个topic的数据吗，也就是说多个消费者协同的订阅topic下的多个消息队列(MessageQueue). 那就需要一种策略去分配多个MessageQueue给每个消费者。

RocktMQ 提供了六种MessageQueue的消费策略。

// 当前消费者 需要消费哪几个消息队列中消息的分配策略。
AllocateMessageQueueStrategy strategy = this.allocateMessageQueueStrategy;

List<MessageQueue> allocateResult = null;
try {
    // mqAll:当前topic和group下的消费队列列表.   cidAll: 所有消费队列的id.
    // 两者一一对应。
    allocateResult = strategy.allocate(this.consumerGroup, this.mQClientFactory.getClientId(),
        mqAll, cidAll);
}catch(Throwable e){}

根据分配策略分配好自己要订阅的MessageQueue之后,就会更新消费者消费订阅表。然后需要根据不同的消费方式(pull/push),执行不同的逻辑。

push方式：会更新消息消费订阅表的版本，然后会通知Broker，触发consumerId变动。
pull方式: 会重新pull消息，执行pullTask.

我们就一起看下 AllocateMessageQueueStrategy 的六种实现。

平均哈希队列算法(AllocateMessageQueueAveragely)

这种策略比较简单。举个例子来说明:

假设有5个mq, 3个消费者，那么消费者id为1的C1会订阅mq1,mq2. C2=>mq3,mq4. C3=>mq5. 假设有5个mq, 6个消费者，那么订阅关系则为: C1=>mq1,C2=>mq2,C3=>mq3,C4=>mq4,C5=>mq5,C6=>[]

// cidAll: 消费者id列表
int index = cidAll.indexOf(currentCID);
// mqAll: 消息队列的列表
// mod: 表示 有多个队列无法平均分配。
int mod = mqAll.size() % cidAll.size();

// 4 <= 5 ? 1 : (4 > 0 && 1 < 4 ? 4 / 5 + 1 : 4 / 5)
int averageSize;
if (mqAll.size() <= cidAll.size()) {
    // 队列比消费者少，每个消费者最大消费1个队列。
    averageSize = 1;
} else {
    if (mod > 0 && index < mod) {
        // 不能平均分, 并且消费者的id比余数小。那么该消费者要多消费一个MessageQueue.
        averageSize = mqAll.size() / cidAll.size() + 1;
    } else {
        // 正好平均分 或者 当前消费者的id大于mod, 那么该消费者就平均分.
        // 会有两种情况: 1、mq有5个,consumer有5个， cid=5,则每个消费者都正好订阅一个mq
        // 2、mq有5个，consumer有6个,cid=6, 则该消费者不会订阅mq
        averageSize = mqAll.size() / cidAll.size();
    }
}
//        int averageSize =
//            mqAll.size() <= cidAll.size() ? 1 : (mod > 0 && index < mod ? mqAll.size() / cidAll.size()
//                + 1 : mqAll.size() / cidAll.size());

// 订阅mq 的起始 索引. 比如5个mq，3个consumer, 那么,c1=>m1,m2; c2=>m3,m4; c3=>m5
int startIndex = (mod > 0 && index < mod) ? index * averageSize : index * averageSize + mod;
int range = Math.min(averageSize, mqAll.size() - startIndex);
for (int i = 0; i < range; i++) {
    result.add(mqAll.get((startIndex + i) % mqAll.size()));
}
return result;

循环平均哈希队列算法(AllocateMessageQueueAveragelyByCircle)

这种实现页是很简单的实现。举例来说明，不过多介绍了。

假设有5个mq, 3个消费者，那么订阅关系则为:C1=>mq1,mq4.
C2=>mq2,mq5.C3=>mq3.

假设有5个mq, 6个消费者，那么订阅关系则为: C1=>mq1,C2=>mq2,C3=>mq3,C4=>mq4,C5=>mq5,C6=>[]

// index: 表示第index个消费者
int index = cidAll.indexOf(currentCID);
for (int i = index; i < mqAll.size(); i++) {
    // 循环分配给消费者。
    if (i % cidAll.size() == index) {
        result.add(mqAll.get(i));
    }
}

一致性哈希队列算法(AllocateMessageQueueConsistentHash)

这个算法实现起来比较复杂，涉及到了一个分布式算法，一致性哈希算法。

具体的算法实现内容可以参考: 哈希算法

// 哈希环的节点
Collection<ClientNode> cidNodes = new ArrayList<>();
for (String cid : cidAll) {
    cidNodes.add(new ClientNode(cid));
}

// 创建一致性哈希的Router
final ConsistentHashRouter<ClientNode> router;
if (customHashFunction != null) {
    router = new ConsistentHashRouter<>(cidNodes, virtualNodeCnt, customHashFunction);
} else {
    router = new ConsistentHashRouter<>(cidNodes, virtualNodeCnt);
}

// 将MQf通过Router分配给消费者
List<MessageQueue> results = new ArrayList<>();
for (MessageQueue mq : mqAll) {
    ClientNode clientNode = router.routeNode(mq.toString());
    if (clientNode != null && currentCID.equals(clientNode.getKey())) {
        results.add(mq);
    }
}

return results;

这种算法的好处也特别明显: 当有消费者加入或者退出的时候，不会移动跨节点的MQ分配情况。

通过配置的属性分配队列(AllocateMessageQueueByConfig)

这种算法就是根据配置的MQ,进行消费。

通过 set方法设置该消费者要消费的MQ,

public void setMessageQueueList(List<MessageQueue> messageQueueList) {
    this.messageQueueList = messageQueueList;
}

分配方法则，直接将 messageQueueList 返回即可。

@Override
public List<MessageQueue> allocate(String consumerGroup, String currentCID, List<MessageQueue> mqAll,
    List<String> cidAll) {
    return this.messageQueueList;
}

机房Hash队列算法(AllocateMessageQueueByMachineRoom)

这个算法和循环平均哈希队列算法是一样的。只不过是根据 brokerName 进行Hash计算的。

举个例子来说:

假设5个mq (fxb-test01@fxb.com,fxb-test02@fxb.com,fxb-test03@fxb.com,fxb-test04@fxb.com,fxb-test05@fxb.com),3个消费者。则订阅关系如下:C1=>mq1,mq4.
C2=>mq2,mq5.C3=>mq3.

代码实现如下:

List<MessageQueue> premqAll = new ArrayList<MessageQueue>();
for (MessageQueue mq : mqAll) {
    String[] temp = mq.getBrokerName().split("@");
    if (temp.length == 2 && consumeridcs.contains(temp[0])) {
        premqAll.add(mq);
    }
}

int mod = premqAll.size() / cidAll.size();
int rem = premqAll.size() % cidAll.size();
int startIndex = mod * currentIndex;
int endIndex = startIndex + mod;
for (int i = startIndex; i < endIndex; i++) {
    result.add(premqAll.get(i));
}
if (rem > currentIndex) {
    result.add(premqAll.get(currentIndex + mod * cidAll.size()));
}
return result;

基于机房远近优先(AllocateMachineRoomNearby)

这种算法会根据消费者的命名来区分。比如相同机房的消费者有相同的命名格式:beijing-mxy-001,beijing-mxy-002... ，那么这类的消费者和MQ会被认为属于同一个机房。在同一个机房内的MQ会优先被本机房的消费者进行消费 如果本机房内没有存活的消费者，那该机房中的MQ会被所有的消费者所共享。至于某一个消费者会订阅哪个MQ，则由具体指定的分配策略来分配。具体如下代码。

//  根据机房分组 broker,
Map<String/*machine room */, List<MessageQueue>> mr2Mq = new TreeMap<String, List<MessageQueue>>();
for (MessageQueue mq : mqAll) {
    String brokerMachineRoom = machineRoomResolver.brokerDeployIn(mq);
    if (StringUtils.isNoneEmpty(brokerMachineRoom)) {
        if (mr2Mq.get(brokerMachineRoom) == null) {
            mr2Mq.put(brokerMachineRoom, new ArrayList<MessageQueue>());
        }
        mr2Mq.get(brokerMachineRoom).add(mq);
    } else {
        throw new IllegalArgumentException("Machine room is null for mq " + mq);
    }
}

//根据机房分组consumer
Map<String/*machine room */, List<String/*clientId*/>> mr2c = new TreeMap<String, List<String>>();
for (String cid : cidAll) {
    // 根据cid判断出所属的机房==> 相同机房的消费需要有相同的命名格式.
    String consumerMachineRoom = machineRoomResolver.consumerDeployIn(cid);
    if (StringUtils.isNoneEmpty(consumerMachineRoom)) {
        if (mr2c.get(consumerMachineRoom) == null) {
            mr2c.put(consumerMachineRoom, new ArrayList<String>());
        }
        mr2c.get(consumerMachineRoom).add(cid);
    } else {
        throw new IllegalArgumentException("Machine room is null for consumer id " + cid);
    }
}

List<MessageQueue> allocateResults = new ArrayList<MessageQueue>();

// 计算当前consumer可以消费的MQ. 当前机房有多个消费者的时候，则使用真正分配策略(其他的分配策略)进行分配。
// 1.allocate the mq that deploy in the same machine room with the current consumer
String currentMachineRoom = machineRoomResolver.consumerDeployIn(currentCID);
List<MessageQueue> mqInThisMachineRoom = mr2Mq.remove(currentMachineRoom);
List<String> consumerInThisMachineRoom = mr2c.get(currentMachineRoom);
if (mqInThisMachineRoom != null && !mqInThisMachineRoom.isEmpty()) {
    allocateResults.addAll(allocateMessageQueueStrategy.allocate(consumerGroup, currentCID, mqInThisMachineRoom, consumerInThisMachineRoom));
}

// 2.allocate the rest mq to each machine room if there are no consumer alive in that machine room
// 如果当前机房没有存活的消费者，那么当前机房的MQ会被所有的消费者共享。根据真实的分配策略进行分配
for (String machineRoom : mr2Mq.keySet()) {
    if (!mr2c.containsKey(machineRoom)) { // no alive consumer in the corresponding machine room, so all consumers share these queues
        allocateResults.addAll(allocateMessageQueueStrategy.allocate(consumerGroup, currentCID, mr2Mq.get(machineRoom), cidAll));
    }
}

return allocateResults;

以上就是RocketMQ使用的负载均衡的内容了，当然，我们也可以自定义分配策略。只需要实现 AllocateMessageQueueStrategy , 在创建消费者实例的时候使用就好了。

拉取消息

我们知道的是， RocketMQ支持两种消费者类型，一种是推送方式的，一种是主动拉取消息。

这里我们先从拉取消息说起。你懂的，我们是要讲的是: DefaultLitePullConsumer.

DefaultLitePullConsumer

DefaultLitePullConsumer，会为每个 messageQueue 创建一个PullTaskImpl. 它会定时的从 Broker端拉取消息。然后封装成 PullRequest, 放到 consumeRequestCache 这个阻塞队列中，然后 DefaultLitePullConsumer 会调用poll方法，获取 List, 这样就算是消费完成了。

是客户端定时任务主动想Broker端发送请求，拉取消息。

而推的方式，就不是这么简单了。

DefaultMQPushConsumer

这还是要从Broker端存储完消息之后说起，Broker会运行一个 ReputMessageService. 这是一个线程，有什么作用呢? 它是推送消息的线程，负责将写入CommitLog的消息推送给对应的消费者。

这里会有两个问题:

1、ReputMessageService 怎么知道要去推送消息给消费者呢

ReputMessageService每隔 1 秒会检查一下这个 CommitLog 是否有新的数据写入。ReputMessageService 自身维护了一个偏移量 reputFromOffset，用以对比和 CommitLog 文件中的消息总偏移量的差距。当这两个偏移量不同的时候，就代表有新的消息到来了。

如下图:

ReputMessageService 是怎样推送消息的

从上图中可以看见， ReputMessageService 封装了一个 DispatchRequest，然后通过 DefaultMessageStore.doDispatch 方法，转发了出去。虽说是转发了出去，但是并非直接发送给了消费者。

而是转发给了 CommitLogDispatcherBuildConsumeQueue. 它主要是根据这条请求按照不同的队列 ID 创建不同的消费队列文件，并在内存中维护一份消费队列列表。然后将 DispatchRequest 请求中这条消息的消息偏移量、消息大小以及消息在发送时候附带的标签的 Hash 值写入到相应的消费队列文件中去。

public void putMessagePositionInfo(DispatchRequest dispatchRequest) {
    ConsumeQueue cq = this.findConsumeQueue(dispatchRequest.getTopic(), dispatchRequest.getQueueId());
    cq.putMessagePositionInfoWrapper(dispatchRequest);
}

后台通过FlushConsumeQueueService线程来定时的持久化到磁盘中，写文件和 Broker写入MappedLogFile一样。只是位置不同。

好家伙，到现在还没有给消息推送消息，自己先存起来了。

而事实上，RocketMQ 并没有实现 Broker端发送消息给消费者。推模型的消息消费模式，是通过拉模式实现的。

这里需要区分广播模式和集群模型分开来说了。

广播模式

每个消费队列的偏移量肯定不能存储在 Broker 服务器端，因为多个消费者对于同一个队列的消费可能不一致，偏移量会互相覆盖掉。因此，在广播模式下，每个客户端的消费偏移量是存储在本地的，然后每隔 5 秒将内存中的 offsetTable 持久化到磁盘中。当首次从服务器获取可消费队列的时候，偏移量是直接从本地文件中读取的。

集群模式

在集群模式下，由于每个客户端所消费的消息队列不同，所以每个消息队列已经消费到哪里的消费偏移量是记录在 Broker 服务器端的。

消费者客户端在内存中维护了一个 offsetTable 表，在 Broker 服务器端也维护了一个偏移量表。在消费者客户端，RebalanceService 服务会定时地 (默认 20 秒) 从 Broker 服务器获取当前客户端所需要消费的消息队列，并与当前消费者客户端的消费队列进行对比，看是否有变化。对于每个消费队列，会从 Broker 服务器查询这个队列当前的消费偏移量。然后根据这几个消费队列，创建对应的拉取请求 PullRequest 准备从 Broker 服务器拉取消息。当从 Broker 服务器拉取下来消息以后，只有当用户成功消费的时候，才会更新本地的偏移量表。本地的偏移量表再通过定时服务每隔 5 秒同步到 Broker 服务器端,维护在 Broker 服务器端的偏移量表也会每隔 5 秒钟序列化到磁盘中.

那么重点来了，创建的PullRequest就会被放到 pullRequestQueue 中。

拉取消息

在消费者启动一文中，可以知道消费者启动后，后台会一直运行一个 PullMessageService 线程。它会阻塞的从 pullRequestQueue中获取 PullRequest 向Broker端发送请求。

@Override
public void run() {

    while (!this.isStopped()) {
        try {
            // 出队
            PullRequest pullRequest = this.pullRequestQueue.take();
            // 拉取消息
            this.pullMessage(pullRequest);
        }
    }
}

pullRequestQueue 便是 pull Message Request Queue了。(拉取消息请求的队列了)

当真正尝试拉取消息之前，其会检查当前请求的内部缓存的消息数量、消息大小、消息阈值跨度是否超过了某个阈值，如果超过某个阈值，则推迟 50 毫秒重新执行这个请求，当执行完一些必要的检查之后，客户端会将用户指定的过滤信息以及一些其它必要消费字段封装到请求信息体中，然后才开始从 Broker 服务器拉取这个请求从当前偏移量开始的消息，默认一次性最多拉取 32 条，服务器返回的响应会告诉客户端这个队列下次开始拉取时的偏移量。客户端每次都会注册一个 PullCallback 回调，用以接受服务器返回的响应信息，根据响应信息的不同状态信息，然后修正这个请求的偏移量，并进行下次请求。

对了，都有些什么时候会将 PullRequest 放入 pullRequestQueue 中的呢?

重平衡的时候

有消费者加入，发生重平衡
定时任务，发送重平衡。

消息消费失败，重试。
启动的时候。

这些都是会触发拉取消息的。

最后，终于到最后一步了, 消费消息。

消费消息

当把消息放在 PullRequestQueue 这个阻塞队列中之后，后台线程会不断从这个阻塞队列中获取 PullRequest. 如下图代码中所示。拿到了 PullRequest 则会提交了一个 ConsumeRequest. 我想这又被你猜到了。哈哈。这里将 ConsumeRequest submit 给了一个后台线程 consumeExecutor。接着往下看吧。该线程会每分钟执行一次。进行消息消费。

依赖于用户指定的消息回调函数的不同，消息的消费分为两种: 并发消费和有序消费。

并发消费没有考虑消息发送的顺序，客户端从服务器获取到消息就会直接回调给用户。而有序消费会考虑每个队列消息发送的顺序，注意此处并不是每个话题消息发送的顺序，一定要记住 RocketMQ 控制消息的最细粒度是消息队列。当我们讲有序消费的时候，就是在说对于某个话题的某个队列，发往这个队列的消息，客户端接受消息的顺序与发送的顺序完全一致。

并发消费

当用户注册消息回调类的时候，如果注册的是 MessageListenerConcurrently 回调类，那么就认为用户不关心消息的顺序问题。我们在上文提到过每个 PullRequest 都关联了一个处理队列 ProcessQueue，而每个处理队列又都关联了一颗消息树 msgTreeMap。当客户端拉取到新的消息以后，其先将消息放入到这个请求所关联的处理队列的消息树中，然后提交一个消息消费请求，用以回调用户端的代码消费消息.

提交给 consumeExecutor 之后， ConsumeRequest 实现逻辑如下:

有序消费

RocketMQ 的有序消费主要依靠两把锁，一把是维护在 Broker 端，一把维护在消费者客户端。Broker 端有一个 RebalanceLockManager 服务，其内部维护了一个 mqLockTable 消息队列锁表:

在有序消费的时候，Broker 需要确保任何一个队列在任何时候都只有一个客户端在消费它，都在被一个客户端所锁定。当客户端在本地根据消息队列构建 PullRequest 之前，会与 Broker 沟通尝试锁定这个队列，另外当进行有序消费的时候，客户端也会周期性地 (默认是 20 秒) 锁定所有当前需要消费的消息队列.

代码逻辑如下

/**
* 顺序消费
* 每秒 定时多实例消费的所有队列，上锁成功将 ProcessQueue的lock属性设置为true
*/
@Override
public void start() {
    if (MessageModel.CLUSTERING.equals(ConsumeMessageOrderlyService.this.defaultMQPushConsumerImpl.messageModel())) { /// 只支持集群模式
        this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                ConsumeMessageOrderlyService.this.lockMQPeriodically();
            }
        }, 1000 * 1, ProcessQueue.REBALANCE_LOCK_INTERVAL, TimeUnit.MILLISECONDS);
    }
}

而在 Broker 这端，每个客户端所锁定的消息队列对应的锁项 LogEntry 有一个上次锁定时的时间戳，当超过锁的超时时间 (默认是 60 秒) 后，也会判定这个客户端已经不再持有这把锁，以让其他客户端能够有序消费这个队列。

在前面我们说到过 RebalanceService 均衡服务会定时地依据不同消费者数量分配消费队列。我们假设 Consumer-1 消费者客户端一开始需要消费 3 个消费队列，这个时候又加入了 Consumer-2 消费者客户端，并且分配到了 MessageQueue-2 消费队列。当 Consumer-1 内部的均衡服务检测到当前消费队列需要移除 MessageQueue-2 队列，这个时候，会首先解除 Broker 端的锁，确保新加入的 Consumer-2 消费者客户端能够成功锁住这个队列，以进行有序消费。

消费者客户端每一次拉取消息请求，如果有发现新的消息，那么都会将这些消息封装为 ConsumeRequest 来喂给消费线程池，等待消费。如果消息特别多，这样一个队列可能有多个消费请求正在等待客户端消费，用户可能会先消费偏移量大的消息，后消费偏移量小的消息。所以消费同一队列的时候，需要一把锁以消费请求顺序化。

代码如下：


public class ConsumeMessageOrderlyService implements ConsumeMessageService {
    class ConsumeRequest implements Runnable {
        private final ProcessQueue processQueue;
        private final MessageQueue messageQueue;

        public ConsumeRequest(ProcessQueue processQueue, MessageQueue messageQueue) {
            this.processQueue = processQueue;
            this.messageQueue = messageQueue;
        }

        public ProcessQueue getProcessQueue() {
            return processQueue;
        }

        public MessageQueue getMessageQueue() {
            return messageQueue;
        }

        @Override
        public void run() {
            final Object objLock = messageQueueLock.fetchLockObject(this.messageQueue);
            synchronized (objLock) {
                if (MessageModel.BROADCASTING.equals(ConsumeMessageOrderlyService.this.defaultMQPushConsumerImpl.messageModel())|| (this.processQueue.isLocked() && !this.processQueue.isLockExpired())) {
                    for (boolean continueConsume = true; continueConsume; ) {
                        // 每次获取消息的数量
                        final int consumeBatchSize =
                            ConsumeMessageOrderlyService.this.defaultMQPushConsumer.getConsumeMessageBatchMaxSize();
                        // 拿到可以消费的消息列表。默认为 1 条。
                        List<MessageExt> msgs = this.processQueue.takeMessages(consumeBatchSize);
                        defaultMQPushConsumerImpl.resetRetryAndNamespace(msgs, defaultMQPushConsumer.getConsumerGroup());
                        if (!msgs.isEmpty()) {
                            final ConsumeOrderlyContext context = new ConsumeOrderlyContext(this.messageQueue);
                            ConsumeOrderlyStatus status = null;
                            ConsumeMessageContext consumeMessageContext = null;
                            try {
                                // 消费者加锁
                                this.processQueue.getConsumeLock().lock();
                                // 回调消费者消息
                                status = messageListener.consumeMessage(Collections.unmodifiableList(msgs), context);
                            } catch (Throwable e) {
                            } finally {
                                this.processQueue.getConsumeLock().unlock();
                            }
                            continueConsume = ConsumeMessageOrderlyService.this.processConsumeResult(msgs, status, context, this);
                        } else {
                            continueConsume = false;
                        }
                    }
                } else {
                    ConsumeMessageOrderlyService.this.tryLockLaterAndReconsume(this.messageQueue, this.processQueue, 100);
                }
            }
        }
    }
}