RocketMQ.0-术语、消费模式、应用场景

1. 什么是 RocketMQ

RocketMQ是一个低延迟、高并发、高可用、高可靠的分布式消息中间件。

从RocketMQ的架构图可以看到，它是由NameServer, Broker, Producer, Consumer四种角色组成的，每一种角色都可以进行水平扩展而不会出现单点故障，所以它天然的支持分布式。

2. RocketMQ 名词介绍

在刚开始使用RocketMQ时相信大家都会跟着官网的 Quick Start 等案例来实现一个demo，我也一样。但是完事之后，感觉好像跟之前没有什么区别，根本不知道demo里面每个步骤中启动的是个什么玩意儿。
所以在这里先记录一下RocketMQ各个角色及名词的介绍，在知道这个之后，或许会对RocketMQ的架构和执行流程有了更好的理解。

2.1 角色名词

先来看一个购物的例子：假设我在京东商城买了一个东西，仓库会把相应的商品打包好，交给京东物流，次日京东物流会指派快递点的某个快递员把商品交到我手上。

在这个过程中有这样几个角色：京东物流、快递点、寄件人仓库、收件人我，这些角色就分别代表了NameServer, Broker, Producer, Consumer。

NameServer: 在上述例子中就是京东物流，京东物流负责的工作是管理所有的快递点，可以把它看成是一个指挥部，它知道所有快递点的信息，它会把商品分发给适合的快递点，让快递点负责交付货物。
Broker: 作为快递点的Broker，负责的就是商品的暂存和传输，即消息暂存和消息传输，同时它还提供消息查询功能。
Producer: 消息生产者，即寄件人仓库。
Consumer: 消息消费者，即收件人我。

2.2 其他术语

在RocketMQ官网的Simple Example 案例中发送消息时会指定消息的Topic, Tag，像下面的代码那样。

//Create a message instance, specifying topic, tag and message body.
Message msg = new Message("TopicTest" /* Topic */,
    "TagA" /* Tag */,
    ("Hello RocketMQ " +i).getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */
);

而在发送消息之前，创建生产者时需要指定group name。

 //Instantiate with a producer group name.
DefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name");

2.2.1 Topic, Tag

刚开始用RocketMQ时我根本不知道所谓的Topic, Tag代表了什么意思，后来才渐渐明白，所以我想如果我在最开始就先去了解这些术语的意思，或许会对RocketMQ的入门有很大的帮助。

首先来想象一个电商网站的页面，它的主页一定有各种类别的商品，比如衣服、裤子、鞋子等等。而每种类别的商品下面还有子类别，假设是品牌的分类。在这个画面中，各种类别的商品就代表Topic，而商品的子类别品牌就代表Tag。

RocketMQ.0-术语、消费模式、应用场景

Topic: 主题，代表消息的类别，可以理解为消息类型的第一级划分。
Tag: 子主题，也属于消息的类别，但是它是Topic下的类别，可以理解为消息类型的第二级划分。

2.2.2 GroupName

所谓的GroupName是用来将相同角色（相同组）的生产者和消费者组合在一起的。

在同一个GroupName中的消费者会消费该组中生产者发送的消息。

3. RocketMQ 消费消息的方式

在Demo程序中创建消费者时，官方给出的是使用DefaultMQPushConsumer这个类来获取消息，其实在RocketMQ中还有一个用于获取消息的类——DefaultLitePullConsumer。

从RocketMQ 4.6.0版本开始DefaultMQPullConsumer类已被标注为弃用，使用DefaultLitePullConsumer类替代其用于主动拉取消息的场景。

可以很容易看出，这两者一个用于Push场景，一个用于Pull场景。前者是服务端主动推送消息给客户端，后者则是客户端需要到服务端拉取数据。

3.1 Pull 模式

先来说说Pull模式，客户端循环地从服务端拉取消息。客户端可以设定适合的“拉取消息等待时间”，等到自己处理完消息之后再拉取新的消息，能够有效防止消息堆积的情况出现。

但是这也是Pull模式的缺陷，即拉取消息的时间间隔较难以界定。如果时间间隔过长，这一批消息都处理完了，时间间隔还没到，必须要等到时间到了之后再拉取新消息，会造成资源的浪费；而如果时间间隔过短，消息未处理完就拉取新消息，容易造成消息堆积。

3.2 Push 模式

Push模式是由服务端在接收到消息后，主动推送到客户端，所以这种方式的实时性较高。但是如果服务端源源不断地推送消息而客户端消费能力不足，就会产生消息堆积的问题。

事实上，Push模式的实现也是一种客户端主动拉取，即“长轮询”。

// PullRequestHoldService#run
while (!this.isStopped()) {
    try {
        if (this.brokerController.getBrokerConfig().isLongPollingEnable()) {
            this.waitForRunning(5 * 1000);
        } else {
            this.waitForRunning(this.brokerController.getBrokerConfig().getShortPollingTimeMills());
        }

        long beginLockTimestamp = this.systemClock.now();
        this.checkHoldRequest();
        long costTime = this.systemClock.now() - beginLockTimestamp;
        if (costTime > 5 * 1000) {
            log.info("[NOTIFYME] check hold request cost {} ms.", costTime);
        }
    } catch (Throwable e) {
        log.warn(this.getServiceName() + " service has exception. ", e);
    }
}

在服务端接收到消息后，队列并不会直接返回，而是通过上面这个循环不断查看其状态，每次等待一段时间（默认5秒），然后调用PullRequestHoldService#checkHoldRequest方法来检查当前Broker的状态。当服务端一直没有新消息，且进行到第三次检查的时候，且超过了用户最大挂起时间brokerSuspendMaxTimeMillis时，才会返回空队列。

而如果在检查的过程中Broker接收到了新消息，就会通过调用PullRequestHoldService#notifyMessageArriving方法发送一个消息已到达的通知。

private void checkHoldRequest() {
    for (String key : this.pullRequestTable.keySet()) {
        String[] kArray = key.split(TOPIC_QUEUEID_SEPARATOR);
        if (2 == kArray.length) {
            String topic = kArray[0];
            int queueId = Integer.parseInt(kArray[1]);
            final long offset = this.brokerController.getMessageStore().getMaxOffsetInQueue(topic, queueId);
            try {
                this.notifyMessageArriving(topic, queueId, offset);
            } catch (Throwable e) {
                log.error("check hold request failed. topic={}, queueId={}", topic, queueId, e);
            }
        }
    }
}