RocketMQ 学习之五 —— 消费者

素颜马尾好姑娘i 2022-02-15 19:07 241阅读 0赞

**目录**

1、DefaultMQPushConsumer

2、DefaultMQPullConsumer

--------------------

消费者分为两种类型。一种是 DefaultMQPushConsumer，由系统控制读取操作，收到消息后自动调用传入的处理方法来处理；另一个是 DefaultMQPullConsumer，读取操作中的大部分功能由使用者自主控制。

# 1、DefaultMQPushConsumer #

上一篇用到的就是这个 ： [https://blog.csdn.net/ruanhao1203/article/details/89496832][https_blog.csdn.net_ruanhao1203_article_details_89496832]

代码如下：

/*
     * Licensed to the Apache Software Foundation (ASF) under one or more
     * contributor license agreements.  See the NOTICE file distributed with
     * this work for additional information regarding copyright ownership.
     * The ASF licenses this file to You under the Apache License, Version 2.0
     * (the "License"); you may not use this file except in compliance with
     * the License.  You may obtain a copy of the License at
     *
     *     http://www.apache.org/licenses/LICENSE-2.0
     *
     * Unless required by applicable law or agreed to in writing, software
     * distributed under the License is distributed on an "AS IS" BASIS,
     * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
     * See the License for the specific language governing permissions and
     * limitations under the License.
     */
    package org.apache.rocketmq.example.quickstart;
     
    import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
    import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
    import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
    import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
    import org.apache.rocketmq.client.exception.MQClientException;
    import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
    import org.apache.rocketmq.common.message.MessageExt;
     
    import java.util.List;
     
    /**
     * This example shows how to subscribe and consume messages using providing {@link DefaultMQPushConsumer}.
     */
    public class TestConsumer {
     
        public static void main(String[] args) throws InterruptedException, MQClientException {
     
            /*
             * Instantiate with specified consumer group name.
             */
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("firstGroupName");
            consumer.setNamesrvAddr("10.24.54.241:9876;10.24.54.242:9876");
            /*
             * Specify name server addresses.
             * <p/>
             *
             * Alternatively, you may specify name server addresses via exporting environmental variable: NAMESRV_ADDR
             * <pre>
             * {@code
             * consumer.setNamesrvAddr("name-server1-ip:9876;name-server2-ip:9876");
             * }
             * </pre>
             */
     
            /*
             * Specify where to start in case the specified consumer group is a brand new one.
             */
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
     
            /*
             * Subscribe one more more topics to consume.
             */
            consumer.subscribe("TestTopic1", "*");
     
            /*
             *  Register callback to execute on arrival of messages fetched from brokers.
             */
            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;
                }
            });
     
            /*
             *  Launch the consumer instance.
             */
            consumer.start();
     
            System.out.printf("Consumer Started.%n");
        }
    }

DefaultMQPushConsumer 需要设置三个参数：

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

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

**Clustering 模式**下，同一个 ConsumerGroup（GroupName 相同）里的每个 Consumer 只消费所订阅消息的一部分内容，同一个消费组所有的消费者消费的内容合起来才是所订阅 Topic 内容的整体，从而达到负载均衡的目的。

**Broadcasting 模式**下，同一个消费组里的每个消费者都能消费所订阅 Topic 的全部信息，也就是一个消息会被多次分发，被多个 Consumer 消费。

2. NameServer 的地址和端口号，可以填写多个，用分号分隔，达到消除单点故障的目的。

3. Topic 名称用来标识消息类型，需要提前创建（也可以在启动broker时设置允许动态创建）。如果不需要消费某个 Topic 下的所有消息，可以通过指定消息的 **Tag ** 进行消息过滤，比如 Consumer.subscribe("TopicTest" , "tag1 || tag2 || tag3")，表示这个 Consumer 要消费 “TopicTest” 下带有 tag1、tag2、tag3的消息。Tag 参数设置 null 或 “\*” 表示消费这个 Topic 的所有消息。

点击发送消息

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3J1YW5oYW8xMjAz_size_16_color_FFFFFF_t_70][]

弹出，可以看到tag

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3J1YW5oYW8xMjAz_size_16_color_FFFFFF_t_70 1][]

# 2、DefaultMQPullConsumer #

消费端代码：

/*
     * Licensed to the Apache Software Foundation (ASF) under one or more
     * contributor license agreements.  See the NOTICE file distributed with
     * this work for additional information regarding copyright ownership.
     * The ASF licenses this file to You under the Apache License, Version 2.0
     * (the "License"); you may not use this file except in compliance with
     * the License.  You may obtain a copy of the License at
     *
     *     http://www.apache.org/licenses/LICENSE-2.0
     *
     * Unless required by applicable law or agreed to in writing, software
     * distributed under the License is distributed on an "AS IS" BASIS,
     * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
     * See the License for the specific language governing permissions and
     * limitations under the License.
     */
    package org.apache.rocketmq.example.quickstart;
    
    import org.apache.rocketmq.client.consumer.DefaultMQPullConsumer;
    import org.apache.rocketmq.client.consumer.PullResult;
    import org.apache.rocketmq.client.exception.MQClientException;
    import org.apache.rocketmq.common.message.MessageExt;
    import org.apache.rocketmq.common.message.MessageQueue;
    
    import java.util.HashMap;
    import java.util.List;
    import java.util.Map;
    import java.util.Set;
    
    public class TestPullConsumer {
        private static final Map<MessageQueue, Long> OFFSE_TABLE = new HashMap<MessageQueue, Long>();
    
        public static void main(String[] args) throws MQClientException {
            DefaultMQPullConsumer consumer = new DefaultMQPullConsumer("firstGroupName");
            consumer.setNamesrvAddr("10.24.54.241:9876;10.24.54.242:9876");
            consumer.start();
    
            Set<MessageQueue> mqs = consumer.fetchSubscribeMessageQueues("TestTopic1");
            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:
                                List<MessageExt> messageExtList = pullResult.getMsgFoundList();
                                for (MessageExt msg : messageExtList) {
                                    System.out.println(new String(msg.getBody()));
                                }
                                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);
        }
    
    }

代码解释：

1. 获取 Message Queue 并遍历

一个 Topic 包括多个 Message Queue，如果这个Consumer 需要获取Topic 下所有的消息，就要遍历所有的 Message Queue。

2. 维护 Offsetstore

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

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

拉取消息的请求发出后，会返回： FOUND（获取到消息）、NO\_MATCHED\_MSG、NO\_NEW\_MSG（没有新消息）、OFFSET\_ILLEGAL 四种状态，需要根据每个状态做不同的处理。

[https_blog.csdn.net_ruanhao1203_article_details_89496832]: https://blog.csdn.net/ruanhao1203/article/details/89496832
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3J1YW5oYW8xMjAz_size_16_color_FFFFFF_t_70]: /images/20220215/291aae40deac48c182ce25970b910809.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3J1YW5oYW8xMjAz_size_16_color_FFFFFF_t_70 1]: /images/20220215/b95c1d2d8fb042ee8904bfca337b09fb.png