RocketMQ源码解析-消费策略源码解析

柔情只为你懂 2024-03-26 22:04 29阅读 0赞

#### 1. 消费者的两种消费方式 ####

1.  **Push消费--底层通过长轮询来实现(DefaultMQPushConsumer来实现的)**
2.  **Pull消费--(4.7.0中代码已经用Deprecated标记了DefaultMQPullConsumer的实现)**

#### 2. 消费者消费模型 ####

![format_png][]

#### 3. 消费者并发消费数据 ####

public class Consumer {
    
      public static void main(String[] args) throws InterruptedException, MQClientException {
    
            // Instantiate with specified consumer group name.
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name");
             
            // Specify name server addresses.
            consumer.setNamesrvAddr("localhost:9876");
            
            // Subscribe one more more topics to consume.
            consumer.subscribe("TopicTest", "*");
            // 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");
        }
    }
    
    复制代码

代码来自官网的例子，这个就是并发消费MQ消息。通过设置MessageListenerConcurrently并发的监听器来实现监听消费的消息然后做后续的处理。通过调用 **`DefaultMQPushConsumer.start`** 方法来启动消费者消费。

public class DefaultMQPushConsumer extends ClientConfig implements MQPushConsumer {
        
        //消费实现类
        protected final transient DefaultMQPushConsumerImpl defaultMQPushConsumerImpl;
        
        //消费组
        private String consumerGroup;
        
        //消费模式--默认为集群消费，还有一种BROADCASTING 广播消费
        private MessageModel messageModel = MessageModel.CLUSTERING;
        
        //消费的起始位置--默认为末尾的offset
        private ConsumeFromWhere consumeFromWhere = ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET;
        
        //消息队列分配策略
        private AllocateMessageQueueStrategy allocateMessageQueueStrategy;
        
        //topic和订阅关系 
        private Map<String /* topic */, String /* sub expression */> subscription = new HashMap<String, String>();
        
        //消息监听器--并发消费和顺序消费 
        private MessageListener messageListener;
        
        //消费offset存储实现 
        private OffsetStore offsetStore;
        
        //最小消费线程数  
        private int consumeThreadMin = 20;
    
        //最大消费线程数
        private int consumeThreadMax = 20;
    
        //用于动态调整线程池数目的阈值
        private long adjustThreadPoolNumsThreshold = 100000;
    
        //并发同时最大跨度偏移。它对顺序消费没有影响
        private int consumeConcurrentlyMaxSpan = 2000;
    
        //流控制阈值在队列级别，每个消息队列默认最多缓存1000条消息
        private int pullThresholdForQueue = 1000;
    
        //在队列级别限制缓存的消息大小，默认情况下每个消息队列最多缓存100 MiB消息
        private int pullThresholdSizeForQueue = 100;
        
        //拉Topic的阈值--无限制
        private int pullThresholdForTopic = -1;
    
        限制主题级别上缓存的消息大小，默认值为-1 MiB(无限制)
        private int pullThresholdSizeForTopic = -1;
    
        //消息拉取间隔
        private long pullInterval = 0;
    
        //批量消费规模
        private int consumeMessageBatchMaxSize = 1;
    
        //批处理拉取大小
        private int pullBatchSize = 32;
    
        //是否每次拉取的时候更新订阅关系
        private boolean postSubscriptionWhenPull = false;
    
    
        private boolean unitMode = false;
    
        //最大重复消费次数- -1-16
        private int maxReconsumeTimes = -1;
    
        //对于需要缓慢拉动的情况，如流量控制情况，暂停拉动时间。
        private long suspendCurrentQueueTimeMillis = 1000;
    
        //以分钟为单位的最大时间量可能会阻塞正在使用的线程。
        private long consumeTimeout = 15;
    
        //异步传输数据的接口
        private TraceDispatcher traceDispatcher = null;
        
        //.........省略部分代码
    }
    复制代码

@Override
    public void start() throws MQClientException {
            setConsumerGroup(NamespaceUtil.wrapNamespace(this.getNamespace(), this.consumerGroup));
            this.defaultMQPushConsumerImpl.start();
            if (null != traceDispatcher) {
                try {
                    traceDispatcher.start(this.getNamesrvAddr(), this.getAccessChannel());
                } catch (MQClientException e) {
                    log.warn("trace dispatcher start failed ", e);
                }
            }
        }
    复制代码

通过调用 **`DefaultMQPushConsumerImpl.start`** 方法来启动消费。看一下DefaultMQPushConsumerImpl的创建

// this 为DefaultMQPushConsumer的实例
     defaultMQPushConsumerImpl = new DefaultMQPushConsumerImpl(this, rpcHook);
    复制代码

接下来看一下 start方法：

public synchronized void start() throws MQClientException {
            switch (this.serviceState) {
                case CREATE_JUST:
                    log.info("the consumer [{}] start beginning. messageModel={}, isUnitMode={}", this.defaultMQPushConsumer.getConsumerGroup(),
                        this.defaultMQPushConsumer.getMessageModel(), this.defaultMQPushConsumer.isUnitMode());
                    this.serviceState = ServiceState.START_FAILED;
    
                    //检查配置信息包括 是否设置了消费组，消费模式等等
                    this.checkConfig();
    
                    //拷贝订阅关系到RebalanceImpl中
                    this.copySubscription();
    
                    if (this.defaultMQPushConsumer.getMessageModel() == MessageModel.CLUSTERING) {
                        this.defaultMQPushConsumer.changeInstanceNameToPID();
                    }
    
                    //获取MQClientInstance
                    this.mQClientFactory = MQClientManager.getInstance().getOrCreateMQClientInstance(this.defaultMQPushConsumer, this.rpcHook);
    
                    //设置rebalanceImpl配置
                    this.rebalanceImpl.setConsumerGroup(this.defaultMQPushConsumer.getConsumerGroup());
                    this.rebalanceImpl.setMessageModel(this.defaultMQPushConsumer.getMessageModel());
                    this.rebalanceImpl.setAllocateMessageQueueStrategy(this.defaultMQPushConsumer.getAllocateMessageQueueStrategy());
                    this.rebalanceImpl.setmQClientFactory(this.mQClientFactory);
                                    //创建PullAPIWrapper--pull的api的包装类
                    this.pullAPIWrapper = new PullAPIWrapper(
                        mQClientFactory,
                        this.defaultMQPushConsumer.getConsumerGroup(), isUnitMode());
                    this.pullAPIWrapper.registerFilterMessageHook(filterMessageHookList);
    
                    //处理offerSet的存储
                    if (this.defaultMQPushConsumer.getOffsetStore() != null) {
                        this.offsetStore = this.defaultMQPushConsumer.getOffsetStore();
                    } else {
                        switch (this.defaultMQPushConsumer.getMessageModel()) {
                            case BROADCASTING:
                                this.offsetStore = new LocalFileOffsetStore(this.mQClientFactory, this.defaultMQPushConsumer.getConsumerGroup());
                                break;
                            case CLUSTERING:
                                this.offsetStore = new RemoteBrokerOffsetStore(this.mQClientFactory, this.defaultMQPushConsumer.getConsumerGroup());
                                break;
                            default:
                                break;
                        }
                        this.defaultMQPushConsumer.setOffsetStore(this.offsetStore);
                    }
                    this.offsetStore.load();
                                   //根据不的监听器创建不同的消息消费服务
                    if (this.getMessageListenerInner() instanceof MessageListenerOrderly) {
                        this.consumeOrderly = true;
                        this.consumeMessageService =
                            new ConsumeMessageOrderlyService(this, (MessageListenerOrderly) this.getMessageListenerInner());
                    } else if (this.getMessageListenerInner() instanceof MessageListenerConcurrently) {
                        this.consumeOrderly = false;
                        this.consumeMessageService =
                            new ConsumeMessageConcurrentlyService(this, (MessageListenerConcurrently) this.getMessageListenerInner());
                    }
    
                    //启动消费服务--定时清理过期的消息
                    this.consumeMessageService.start();
    
                    boolean registerOK = mQClientFactory.registerConsumer(this.defaultMQPushConsumer.getConsumerGroup(), this);
                    if (!registerOK) {
                        this.serviceState = ServiceState.CREATE_JUST;
                        this.consumeMessageService.shutdown();
                        throw new MQClientException("The consumer group[" + this.defaultMQPushConsumer.getConsumerGroup()
                            + "] has been created before, specify another name please." + FAQUrl.suggestTodo(FAQUrl.GROUP_NAME_DUPLICATE_URL),
                            null);
                    }
                                   //MQClientInstance启动
                    mQClientFactory.start();
                    log.info("the consumer [{}] start OK.", this.defaultMQPushConsumer.getConsumerGroup());
                    this.serviceState = ServiceState.RUNNING;
                    break;
                case RUNNING:
                case START_FAILED:
                case SHUTDOWN_ALREADY:
                    throw new MQClientException("The PushConsumer service state not OK, maybe started once, "
                        + this.serviceState
                        + FAQUrl.suggestTodo(FAQUrl.CLIENT_SERVICE_NOT_OK),
                        null);
                default:
                    break;
            }
    
            t

[format_png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/25/4def3cf31f5f4de0b9eac7bebb8bc614.png