RocketMQ原理学习--消费者消费消息

迈不过友情╰ 2022-04-11 04:23 277阅读 0赞

在之前的一篇博客[《RocketMQ原理学习--消息类型》][RocketMQ_--]中我们有介绍过RocketMQ的消息类型，这篇博客我们简单介绍一下RocketMQ消费者是如何消费消息的。

### 一、Pull or Push ###

简单来说RocketMQ给我们提供了两种消息消费方式，Pull模式和Push模式，简单理解我们可能会认为Pull模式是消费者主动去拉取消息，Push模式是RocketMQ的Broker主动将消息推送过来，其实RocketMQ对于这两种方式都是采用的Pull拉取的方式，Push模式不过是通过回调来实现的，让我们理解为推送模式

1、Push示例：

提供一个MessageListenerConcurrently监听器，当存在消息时会回调这个类的consumeMessage方法。

public class PushConsumer {
    
        public static void main(String[] args) throws InterruptedException, MQClientException {
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("rmq-group");
            consumer.setNamesrvAddr("localhost:9876");
            //consumer.setInstanceName("rmq-instance2");
            consumer.subscribe("TopicA-test", "TagA");
    
            consumer.registerMessageListener(new MessageListenerConcurrently() {
                public ConsumeConcurrentlyStatus consumeMessage(
                        List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                    for (MessageExt msg : msgs) {
                        System.out.println(new String(msg.getBody()));
                    }
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }
            });
            consumer.start();
            System.out.println("Consumer Started.");
        }

2、Pull模式：

维护MessageQueue和消息offset不断的去拉取消息。

public class PullConsumer {
    
    
        private static final Map<MessageQueue, Long> OFFSE_TABLE = new HashMap<>();
    
        public static void main(String [] args) throws Exception{
    
    
            DefaultMQPullConsumer consumer = new DefaultMQPullConsumer("rmq-group");
    
            consumer.start();
    
            Set<MessageQueue> mqs = consumer.fetchSubscribeMessageQueues("TopicA-test");
            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);
        }
    
    
    }

### 二、执行流程 ###

1、Push模式：

对于Push模式，RocketMQ提供了PullMessageService线程，定时不断的从RocketMQ中拉取消息，最终来回调MessageListener的consumeMessage方法来消费消息，如下执行流程图：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxOTI0ODYyMDc3_size_16_color_FFFFFF_t_70][]

2、Pull模式：

Pull模式理解起来就比较简单了，就是不断的去拉取消息即可

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxOTI0ODYyMDc3_size_16_color_FFFFFF_t_70 1][]

### 三、定时任务 ###

消费者更新本地broker信息、心跳发送等操作都是通过定时任务来完成的，以下是消费者的一些定时任务在MQClientInstance的startScheduledTask方法中启动。

private void startScheduledTask() {
            if (null == this.clientConfig.getNamesrvAddr()) {
                this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
    
                    @Override
                    public void run() {
                        try {
    					    //更新nameserver地址
                            MQClientInstance.this.mQClientAPIImpl.fetchNameServerAddr();
                        } catch (Exception e) {
                            log.error("ScheduledTask fetchNameServerAddr exception", e);
                        }
                    }
                }, 1000 * 10, 1000 * 60 * 2, TimeUnit.MILLISECONDS);
            }
    
            this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
    
                @Override
                public void run() {
                    try {
    					//从nameserver中获取topic信息
                        MQClientInstance.this.updateTopicRouteInfoFromNameServer();
                    } catch (Exception e) {
                        log.error("ScheduledTask updateTopicRouteInfoFromNameServer exception", e);
                    }
                }
            }, 10, this.clientConfig.getPollNameServerInterval(), TimeUnit.MILLISECONDS);
    
            this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
    
                @Override
                public void run() {
                    try {
    					//清除下线的broker
                        MQClientInstance.this.cleanOfflineBroker();
    					//发送心跳
                        MQClientInstance.this.sendHeartbeatToAllBrokerWithLock();
                    } catch (Exception e) {
                        log.error("ScheduledTask sendHeartbeatToAllBroker exception", e);
                    }
                }
            }, 1000, this.clientConfig.getHeartbeatBrokerInterval(), TimeUnit.MILLISECONDS);
    
            this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
    
                @Override
                public void run() {
                    try {
    					//持久化消息消费
                        MQClientInstance.this.persistAllConsumerOffset();
                    } catch (Exception e) {
                        log.error("ScheduledTask persistAllConsumerOffset exception", e);
                    }
                }
            }, 1000 * 10, this.clientConfig.getPersistConsumerOffsetInterval(), TimeUnit.MILLISECONDS);
    
            this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
    
                @Override
                public void run() {
                    try {
    					//调整线程池参数
                        MQClientInstance.this.adjustThreadPool();
                    } catch (Exception e) {
                        log.error("ScheduledTask adjustThreadPool exception", e);
                    }
                }
            }, 1, 1, TimeUnit.MINUTES);
        }

### 四、消费者消费策略 ###

RocketMQ提供了消费者的消费策略接口AllocateMessageQueueStrategy并给出了几种消费策略：

（1）AllocateMachineRoomNearby：相近机房策略

（2）AllocateMessageQueueConsistentHash：一致性hash策略

（3）AllocateMessageQueueByMachineRoom：根据机房hash

（4）AllocateMessageQueueByConfig：配置策略

（5）AllocateMessageQueueAveragelyByCircle：hash轮询策略

（6）AllocateMessageQueueAveragely：平均策略

### **五、消费 offset 信息存储** ###

对于消息的消费坐标 offset 信息 RocketMQ 对于集群消息和广播消息处理的方式是不同的：

（1）广播消息：每个消费者维护自己消费消息的 offset 坐标信息，将信息持久化到本地文件中，本地文件地址命名及内容示例： 文件地址：.rocketmq\_offsets\\192.168.182.1@DEFAULT\\rmq-group\\offsets.json 内容

{
        "offsetTable":{
       {
                "brokerName":"DESKTOP-TKINHU2",
                "queueId":0,
                "topic":"TopicA-test"
            }:7545,{
                "brokerName":"DESKTOP-TKINHU2",
                "queueId":3,
                "topic":"TopicA-test"
            }:7543,{
                "brokerName":"DESKTOP-TKINHU2",
                "queueId":1,
                "topic":"TopicA-test"
            }:7541,{
                "brokerName":"DESKTOP-TKINHU2",
                "queueId":2,
                "topic":"TopicA-test"
            }:7543
          }
        }

（2）集群消息：消费者每隔一段时间（5s）将消费的 offset 坐标信息上报到 Broker 节点，由Broker 节点统一维护管理，持久化到 config 目录下的consumerOffset.json文件中。

示例：

{
        "offsetTable":{
            # 记录每个queueId消费的offset信息
            "TopicA-test@rmq-group":{0:7540,1:7537,2:7538,3:7537
            },
            "%RETRY%rmq-group@rmq-group":{0:154
            }
         }
        }

由于offset信息持久化存在时间间隔，导致如果rocketMQ服务关闭时可能导致offset没有及时持久化，那样就很容易导致消息重复消费（业务要做好幂等性），rocketMQ还是提供消费者进行配置在消息启动时如何消费消息。

（1）CONSUME\_FROM\_LAST\_OFFSET：从最后offset坐标开始消费消息

（2）CONSUME\_FROM\_FIRST\_OFFSET：消费者从一开始offset持久化消费消息

（3）CONSUME\_FROM\_TIMESTAMP：从某个时间开始消费消息

[RocketMQ_--]: https://blog.csdn.net/qq924862077/article/details/84574455
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxOTI0ODYyMDc3_size_16_color_FFFFFF_t_70]: /images/20220411/f93cf778176c4f2e954da7842d7b567e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxOTI0ODYyMDc3_size_16_color_FFFFFF_t_70 1]: /images/20220411/24339ac43d0145abaf91b03148248584.png