RocketMQ源码： consumer启动流程

偏执的太偏执、 2024-03-24 14:47 78阅读 0赞

`org.apache.rocketmq.example.quickstart.Consumer`:

public class Consumer {
    
        public static void main(String[] args) throws InterruptedException, MQClientException {
    
            // todo
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name_4");
    
            consumer.setNamesrvAddr("127.0.0.1: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;
                }
            });
            // todo 启动
            consumer.start();
    
            System.out.printf("Consumer Started.%n");
        }
    }
    复制代码

`consumer`使用起来还是挺简单的，先是创建了一个`DefaultMQPushConsumer`对象，然后配置了一些属性，比较关键的就是注册消息监听器（在这个监听器里会获取消息），之后就调用`start()`方法启动`consumer`.

接下来我们就来分析这块的消费过程。

## 1. 构造方法：`DefaultMQPushConsumer` ##

`consumer`的处理类为`DefaultMQPushConsumer`，我们先来看看`DefaultMQPushConsumer`的属性值和构造方法：

public class DefaultMQPushConsumer extends ClientConfig implements MQPushConsumer {
    
        private final InternalLogger log = ClientLogger.getLog();
    
        protected final transient DefaultMQPushConsumerImpl defaultMQPushConsumerImpl;
    
        // 消费者所属组
        private String consumerGroup;
        // 消息消费模式，分为集群模式、广播模式，默认为集群模式
        private MessageModel messageModel = MessageModel.CLUSTERING;
        // 第一次消费时指定消费策略
        private ConsumeFromWhere consumeFromWhere = ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET;
    
        private String consumeTimestamp = UtilAll.timeMillisToHumanString3(System.currentTimeMillis() - (1000 * 60 * 30));
    
        // 集群模式下消息队列的负载策略
        private AllocateMessageQueueStrategy allocateMessageQueueStrategy;
    
        /**
         * Subscription relationship 订阅信息
         */
        private Map<String /* topic */, String /* sub expression */> subscription = new HashMap<String, String>();
    
        /**
         * Message listener 消息业务监听器
         */
        private MessageListener messageListener;
    
        /**
         * Offset Storage 消息消费进度存储器
         */
        private OffsetStore offsetStore;
    
        /**
         * Minimum consumer thread number 消费者最小线程数
         */
        private int consumeThreadMin = 20;
    
        /**
         * Max consumer thread number  消费者最大线程数，因为消费
         * 者线程池使用无界队列，所以此参数不生效
         */
        private int consumeThreadMax = 20;
    
        /**
         * Threshold for dynamic adjustment of the number of thread pool
         */
        private long adjustThreadPoolNumsThreshold = 100000;
    
        /**
         * Concurrently max span offset.it has no effect on sequential consumption
         * 并发消息消费时处理队列最大跨度，默认2000，表示如果消息处理队列中偏移量最大的消息
         * 与偏移量最小的消息的跨度超过2000，则延迟50ms后再拉取消息
         */
        private int consumeConcurrentlyMaxSpan = 2000;
    
        /**
         * Flow control threshold on queue level, each message queue will cache at most 1000 messages by default,
         * Consider the {@code pullBatchSize}, the instantaneous value may exceed the limit
         * 默认1000，表示每1000
         * 次流控后打印流控日志
         */
        private int pullThresholdForQueue = 1000;
        private int pullThresholdSizeForTopic = -1;
    
        /**
         * Message pull Interval
         * 推模式下拉取任务的间隔时间，默
         * 认一次拉取任务完成后继续拉取
         */
        private long pullInterval = 0;
    
        /**
         * Batch consumption size 消息并发消费时一次
         * 消费消息的条数，通俗点说，就是每次传入
         * MessageListener#consumeMessage中的消息条数
         */
        private int consumeMessageBatchMaxSize = 1;
    
        /**
         * Batch pull size 每次消息拉取的条数，默认32条
         */
        private int pullBatchSize = 32;
    
        /**
         * Whether update subscription relationship when every pull
         * 是否每次拉取消息都更
         * 新订阅信息，默认为false
         */
        private boolean postSubscriptionWhenPull = false;
    
        private boolean unitMode = false;
        /**
         *
         * 最大消费重试次数。如果消息消费次数超过maxReconsumeTimes还未成功，则将该消息转移到一个失败
         * 队列，等待被删除
         */
        private int maxReconsumeTimes = -1;
    
        /**
         * 延迟将该队列的消
         * 息提交到消费者线程的等待时间，默认延迟1s
         */
        private long suspendCurrentQueueTimeMillis = 1000;
    
        /**
         * Maximum amount of time in minutes a message may block the consuming thread.
         * 消息消费超时时间，默认为15，
         * 单位为分钟
         */
        private long consumeTimeout = 15;
        ...
        
        public DefaultMQPushConsumer(final String consumerGroup) {
            // 指定了队列分配策略 AllocateMessageQueueAveragely
            this(null, consumerGroup, null, new AllocateMessageQueueAveragely());
        }
    
        ...
        public DefaultMQPushConsumer(final String n