【RocketMq实战第四篇】-不同类型消费者（DefaultMQPushConsumer&DefaultMQPullConsumer）

Dear 丶 2022-03-22 06:37 423阅读 0赞

## 前言 ##

生产者和消费者是消息队列的两个重要角色，生产者向消息队列写人数据， 消费者从消息队列里读取数据。本篇讲解两种类型的消费者，一个是 DefaultMQPushConsumer，由系统控制读取操作，收到消息后自动调用传人的 处理方法来处理;另 一个是 DefaultMQPullConsumer，读取操作中的大部分功 能由使用者自主控制 。

## 正文 ##

### DefaultMQPushConsumer  ###

使用 DefaultMQPushConsumer 主要是设置好各种参数和传人处理消息的函数 。 系统收到消息后自动调用处理函数来处理消息，自动保存 Offset，而且加入新的 DefaultMQPushConsumer后会自动做负载均衡。

我们在安装目录下找到example项目的  package org.apache.rocketmq.example.quickstart;目录下源码介绍。

public class Consumer {
    
        public static void main(String[] args) throws InterruptedException, MQClientException {
          
            DefaultMQPushConsumer consumer = new  DefaultMQPushConsumer("please_rename_unique_group_name_4");
    
            consumer.setNamesrvAddr("name-server1-ip:9876;name-server2-ip: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("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });
     
            consumer.start();
            System.out.printf("Consumer Started.%n");
        }
    }

**讲解源码**

DefaultMQPushConsumer需要设置三 个 参数 : 一 是这个 Consumer 的 GroupName，二是 NameServer 的地址和端 口号，三是 Topic 的名称 ，下面将分 别进行详细介绍 。

（1) Consumer的 GroupName用于把多个 Consumer组织到一起， 提高并发 处理能力， GroupName需要和消息模式 (MessageModel)配合使用。

RocketMQ支持两种消息模式: **Clustering**和**Broadcasting**。

*  在 Clustering模式下，同一个 ConsumerGroup(GroupName相同) 里的每 个 Consumer 只消费所订阅消 息的一部分 内容， 同一个 ConsumerGroup 里所有的 Consumer消费的内容合起来才是所订阅 Topic 内容的整体， 从而达到负载均衡的目的 。
 *  在 Broadcasting模式下，同一个 ConsumerGroup里的每个 Consumer都 能消费到所订阅 Topic 的全部消息，也就是一个消息会被多次分发，被 多个 Consumer消费。

（2）**NameServer** 的地址和端口 号，可以填写多个 ，用分号隔开，达到消除单点故障的目的，比如 “ip1:port;ip2:port;ip3 :port” 。

（3）**Topic**名称用来标识消息类型， 需要提前创建。如果不需要消费某 个 Topic 下的所有消息，可以通过指定消息的 Tag 进行消息过滤，比如: Consumer.subscribe (”TopicTest”，’tag1 || tag2 || tag3”)， 表示这个 Consumer要 消费“ TopicTest”下带有 tag1 或 tag2 或 tag3 的消息。在填写 Tag 参数的位置，用 null 或者“ \*“ 表示要消费这个 Topic 的所有消息 。

**处理流程**

网上有很多文章讲处理流程的，个人觉得这个比较详细，[https://blog.csdn.net/panxj856856/article/details/80776032][https_blog.csdn.net_panxj856856_article_details_80776032]把defaultMQPushConsumerImpl.start()方法按顺序都分析了。

不过我要说的重点是消息处理逻辑是在pullMessage这个函数的PullCallBack中。PullCallBack函数里有个 switch 语句，根据从 Broker 返回的消息类型做相应的 处理

switch (pullResult.getPullStatus ()) {
        case FOUND:
           ……
          break;
        case NO_NEW_MSG
           ……
           break;
         case OFFSET_ILLEGAL :
           ……
           break;
         default:
           break;
        }

pullMessage函数的参数是 final PullRequest pullRequest ，这是通过“长轮询”方式达到 Push效果的方法，**长轮询方式既有 Pull 的优点，又兼具 Push方式的实时性。**

**Push的方式是 Server端接收到消息后，主动把消息推送给 Client端，主动权在Server端，实时性高。用 Push方式主动推送有很多弊 端:首先是加大 Server 端的 工作量，进而影响 Server 的性能;其次，Client 的处理能力各不相同， Client 的状态不受 Server 控制，**

**Pull方式是 Client端循环地从 Server端拉取消息，主动权在 Client手里， 自己拉取到一定量消息后，处理妥当了再接着取。Pull 方式的问题是循环拉取 消息的间隔不好设定，间隔太短就处在一个 “忙等”的状态，浪费资源;  Pull 的时间间隔太长 Server 端有消息到来时 有可能没有被及时处理。**

下面就来介绍**“长轮询“**

“长轮询”方式通过 Client端和 Server端的配合，达到既拥有 Pull 的优 点，又能达到保证实时性的目的 。

**大家只需记住长轮询就是在Broker在没有新消息的时候才阻塞，阻塞时间默认设置是 15秒，有消息会立刻返回**

有兴趣的可以看一下长轮询源码在package org.apache.rocketmq.broker.longpolling.PullRequestHoldService.java的 run方法，里面是有三次Check，每次5s。

**“长轮询”的 主动权还是 掌 握在 Consumer 手中， Broker 即使有大 量 消息积 压 ，也不会主动推 送给 Consumer 。**

**DefaultMQPushConsumer 的流量控制**

上面我们分析得知PushConsumer的核心还是 Pull 方式。PushConsumer有个线程池 ， 消息处理逻辑在各个线程里同时执行。

多线程处理业务是很麻烦的，所以RocketMQ定义了一个快照类 ProcessQueue来解决 堆积的数量 ?如何重复处理某些消息? 如何延迟处理某些消息? 等问题。每个 Message Queue 都会有个对应的 ProcessQueue 对象，保存了这个 Message Queue 消息处理状态的快照 。

ProcessQueue对象里主要的内容是一个 TreeMap 和一个读写锁。 TreeMap 里以 Message Queue 的 Offset作为 Key，以消息内容的引用为 Value，保存了 所有从 MessageQueue 获取到，但是还未被处理的消息; 读写 锁控制着多个线程对 TreeMap 对象的并发访 问 。

有 了 ProcessQueue 对象，流量控 制 就方便和灵活多了 ，客 户 端在每次 Pull请求前会做几个判断，分别取但还未处理的消息个数、消 息总大小、 Offset 的跨度，任何一个值超过设定的大小就隔一段时间再拉取消 息，从而达到流量控制的目的 。 此外 ProcessQueue 还可以辅助实现顺序消费的 逻辑。

有兴趣的可以翻看源码位置在org.apache.rocketmq.client.impl.consumer.pullMessage()

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODAwMzM4OQ_size_16_color_FFFFFF_t_70][]

## ##

### DefaultMQPullConsumer ###

我们来看一下包中示例代码，位置在package org.apache.rocketmq.example.simple.PullConsumer;

public class PullConsumer {
        private static final Map<MessageQueue, Long> OFFSE_TABLE = new HashMap<MessageQueue, Long>();
    
        public static void main(String[] args) throws MQClientException {
            DefaultMQPullConsumer consumer = new DefaultMQPullConsumer("please_rename_unique_group_name_5");
    
            consumer.start();
    
            Set<MessageQueue> mqs = consumer.fetchSubscribeMessageQueues("TopicTest1");
            for (MessageQueue mq : mqs) {
                System.out.printf("Consume from the queue: %s%n", mq);
                SINGLE_MQ:
                while (true) {
                    try {
                        PullResult pullResult =
                            consumer.pullBlockIfNotFound(mq, null, getMessageQueueOffset(mq), 32);
                        System.out.printf("%s%n", pullResult);
                        putMessageQueueOffset(mq, pullResult.getNextBeginOffset());
                        switch (pullResult.getPullStatus()) {
                            case FOUND:
                                break;
                            case NO_MATCHED_MSG:
                                break;
                            case NO_NEW_MSG:
                                break SINGLE_MQ;
                            case OFFSET_ILLEGAL:
                                break;
                            default:
                                break;
                        }
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            }
    
            consumer.shutdown();
        }
    
        private static long getMessageQueueOffset(MessageQueue mq) {
            Long offset = OFFSE_TABLE.get(mq);
            if (offset != null)
                return offset;
    
            return 0;
        }
    
        private static void putMessageQueueOffset(MessageQueue mq, long offset) {
            OFFSE_TABLE.put(mq, offset);
        }
    
    }

**讲解源码**

处理 逻辑是逐个 读 取某 Topic 下所有 Message Queue 的内容， 读完一遍后退出， 主要处理额外的三件事情:

( 1 )获取 Message Queue 并遍历

一 个 Topic 包括多个 Message Queue，如果这个 Consumer 需要获取 Topic 下所有的消息，就 要遍历多有的 Message Queue。 如果有特殊情况，也可以选 择某些特定的 Message Queue 来读取消息 。

( 2 )维护 Offsetstore

从一个 Message Queue 里拉取消息的时候，要传人 Offset参数( long类型 的值)，随着不断读取消息 ， Offset会不断增长 。 这个时候由用户负责把 Offset 存储下来，根据具体情况可以存到内存里、写到磁盘或者数据库里等 。

( 3 )根据不同的消息状态做不同的处理

拉取消息的请求发出后，会返回: FOUND、 NO\_MATCHED\_MSG、 NO\_NEW\_MSG、 OFFSET\_ILLEGAL 四种状态，需要根据每个状态做不同的处理 。**比较重要的两个状态是 FOUNT 和 NO NEW MSG ，分别表示获取到消息和没 有新的消息 。**

## ##

### Consumer 的启动、关闭流程 ###

对于 PullConsumer来说，使用者主动权很高，可以根据实际需要暂停、停止、启动消费过程 。 需要注意的是 Offset 的保存，要在程序的异常处理部分增加把 Offset 写人磁盘方 面的处理，记准了每个 Message Queue 的 Offset，才能保证消息消 费 的准确性 。

DefaultMQPushConsumer 的退出， 要调用 shutdown() 函数， 以便 释放资 源、保存 Offset 等 。 这个调用要加到 Consumer 所在应用的退出逻辑中 。

**启动 DefaultMQPushConsumer 时, NameServer 地址填错，程序仍然 可以正常启动，但是不会收到消息 。**

解决启动时NameServer填写错误报错：

可以在 Consumer.start()语句后调用: Consumer.fetchSubscribeMessageQueues(”TopicName”)，这 时如果配 置信息写得不准确，或者当 前服务不可 用，这个语句会报 MQClientException 异 常 。

[https_blog.csdn.net_panxj856856_article_details_80776032]: https://blog.csdn.net/panxj856856/article/details/80776032
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODAwMzM4OQ_size_16_color_FFFFFF_t_70]: /images/20220322/5ab0e8979f9247e28380b491797dddb5.png