9、RocketMQ的Consumer之启动流程

约定不等于承诺〃 2024-02-05 14:42 30阅读 0赞

Consumer之启动流程  
Producer将消息发送给Broker后Broker做好持久化这是就可以创建消费者来对消息进行消费，今天我们分析如何启动一个消费者代码，原理与启动生产者类似。

DefaultMQPushConsumer类的初始化

public DefaultMQPushConsumer(final String namespace, final String consumerGroup, RPCHook rpcHook,
        AllocateMessageQueueStrategy allocateMessageQueueStrategy) {
        
        // 消费者组
        this.consumerGroup = consumerGroup;
        // 命名空间
        this.namespace = namespace;
        // 分配策略
        this.allocateMessageQueueStrategy = allocateMessageQueueStrategy;
        // 创建DefaultMQPushConsumerImpl实例
        defaultMQPushConsumerImpl = new DefaultMQPushConsumerImpl(this, rpcHook);
    }

DefaultMQPushConsumerImpl作为DefaultMQPushConsumer具体实现，基本上和之前分析过的Producer类似，该方法初始化注册了钩子函数，注册messageListener消息监听器，赋值一些namesevAddr等参数。

**subscribe订阅**

Producer创建时可以去指定发送的Topic而消费端只要自己去订阅相关topic，同时也可以根据tag与Sql方式进行消息的过滤。

/**
     * 订阅topic支持消息过滤表达式
     * @param topic 订阅的topic
     * @param subExpression 只支持或（“||”）操作，如果是null或者“*”则表示订阅全部
     */
    public void subscribe(String topic, String subExpression) throws MQClientException {
        
        this.defaultMQPushConsumerImpl.subscribe(withNamespace(topic), subExpression);
    }
    /**
     *
     */
    public void  subscribe(String topic, String subExpression) throws MQClientException {
        
        try {
        
            // 解析订阅表达式生成SubscriptionData
            SubscriptionData subscriptionData = FilterAPI.buildSubscriptionData(topic, subExpression);
            // 将topic与SubscriptionData的关系维护到RebalanceImpl内部的subscriptionInner
            this.rebalanceImpl.getSubscriptionInner().put(topic, subscriptionData);
            if (this.mQClientFactory != null) {
        
                // 如果mQClientFactory不为null就发送心跳给所有broker
                this.mQClientFactory.sendHeartbeatToAllBrokerWithLock();
            }
        } catch (Exception e) {
        
            throw new MQClientException("subscription exception", e);
        }
    }

**DefaultMQPushConsumer的start方法**

正式开始启动消费者，调用DefaultMQPushConsumer的start启动。
    public void start() throws MQClientException {
        
      // 根据namespace和consumerGroup设置消费者组
        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);
            }
        }
    }

DefaultMQPushConsumer的start启动的逻辑主要就是根据namespace和ConsumerGroup去设置消费者组，主要的实现还是在其近视实现类DefaultMQPushConsumerImpl（门面模式）中。

public synchronized void start() throws MQClientException {
        
        // 根据服务状态实现不同的逻辑，只有当为CREATE_JUST才会正常启动，其他类型的都报错
        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();
              // 拷贝订阅关系，为集群模式的消费者配置其对应的重试队列
                this.copySubscription();
    
                if (this.defaultMQPushConsumer.getMessageModel() == MessageModel.CLUSTERING) {
        
                    this.defaultMQPushConsumer.changeInstanceNameToPID();
                }
    
                this.mQClientFactory = MQClientManager.getInstance().getOrCreateMQClientInstance(this.defaultMQPushConsumer, this.rpcHook);
                // 消费者组
                this.rebalanceImpl.setConsumerGroup(this.defaultMQPushConsumer.getConsumerGroup());
                // 消费者模式（广播/集群）
                this.rebalanceImpl.setMessageModel(this.defaultMQPushConsumer.getMessageModel());
                //Consumer负载均衡策略（集群模式才存在负载均衡）
                this.rebalanceImpl.setAllocateMessageQueueStrategy(this.defaultMQPushConsumer.getAllocateMessageQueueStrategy());
                this.rebalanceImpl.setmQClientFactory(this.mQClientFactory);
    
                this.pullAPIWrapper = new PullAPIWrapper(
                    mQClientFactory,
                    this.defaultMQPushConsumer.getConsumerGroup(), isUnitMode());
                this.pullAPIWrapper.registerFilterMessageHook(filterMessageHookList);
    
                if (this.defaultMQPushConsumer.getOffsetStore() != null) {
        
                    this.offsetStore = this.defaultMQPushConsumer.getOffsetStore();
                } else {
        
                    switch (this.defaultMQPushConsumer.getMessageModel()) {
        
                        // 广播模式（本地的Offset文件）
                        case BROADCASTING:
                            this.offsetStore = new LocalFileOffsetStore(this.mQClientFactory, this.defaultMQPushConsumer.getConsumerGroup());
                            break;
                        // 集群模式 （Broker里面的offset）
                        case CLUSTERING:
                            this.offsetStore = new RemoteBrokerOffsetStore(this.mQClientFactory, this.defaultMQPushConsumer.getConsumerGroup());
                            break;
                        default:
                            break;
                    }
                    this.defaultMQPushConsumer.setOffsetStore(this.offsetStore);
                }
                // 加载消息偏移量
                this.offsetStore.load();
                //Consumer中自行指定的回调函数。
                if (this.getMessageListenerInner() instanceof MessageListenerOrderly) {
        
                    this.consumeOrderly = true;
                    // MessageListenerOrderly服务
                    this.consumeMessageService =
                        new ConsumeMessageOrderlyService(this, (MessageListenerOrderly) this.getMessageListenerInner());
                } else if (this.getMessageListenerInner() instanceof MessageListenerConcurrently) {
        
                    this.consumeOrderly = false;
                    // MessageListenerConcurrently服务
                    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(defaultMQPushConsumer.getAwaitTerminationMillisWhenShutdown());
                    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);
                }
              // 启动CreateMQClientInstance客户端通信实例
                // 包括netty服务，各种定时任务，拉取消息服务，rebalanceService服务。
                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;
        }
        
        // 向NameServer拉取并更新当前消费者订阅的topic路由信息
        this.updateTopicSubscribeInfoWhenSubscriptionChanged();
        // 随机选择broker检查tags
        this.mQClientFactory.checkClientInBroker();
        // 发送心跳给所有Broker
        this.mQClientFactory.sendHeartbeatToAllBrokerWithLock();
        // 唤醒负载均衡服务rebalanceService，主动进行一次MessageQueue的重平衡。
        this.mQClientFactory.rebalanceImmediately();
    }

相关逻辑实现  
检查配置  
调用checkConfig()方法，对一系列参数进行校验。

private void checkConfig() throws MQClientException {
        
          // 检查消费组（不能为空，长度不能大于255，只能包含数字和字母）
            Validators.checkGroup(this.defaultMQPushConsumer.getConsumerGroup());
            if (null == this.defaultMQPushConsumer.getConsumerGroup()) {
        
                throw new MQClientException(
                    "consumerGroup is null"
                        + FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL),
                    null);
            }
          // 检查是否是默认值名，是则报错
            if (this.defaultMQPushConsumer.getConsumerGroup().equals(MixAll.DEFAULT_CONSUMER_GROUP)) {
        
                throw new MQClientException(
                    "consumerGroup can not equal "
                        + MixAll.DEFAULT_CONSUMER_GROUP
                        + ", please specify another one."
                        + FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL),
                    null);
            }
          // 检查消费模式（广播或集群）
            if (null == this.defaultMQPushConsumer.getMessageModel()) {
        
                throw new MQClientException(
                    "messageModel is null"
                        + FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL),
                    null);
            }
          //  ....
    }

首先检查消费组不能为空，长度不能大于255，只能包含数字和字母，同时消费组的名字不能和默认消费组(DEFAULT\_CONSUMER)名字一致，否则就会抛异常。

检查消费模式只能存在集群或者广播模式，其它的则会抛异常。

检查ConsumeFromWhere是否为空，ConsumeFromWhere表示消费策略，简单理解就是消息从什么地方开始消费，默认CONSUME\_FROM\_LAST\_OFFSET(队列的最后开始消费)，还可以指定CONSUME\_FROM\_FIRST\_OFFSET(队列的第一个开始消费)和CONSUME\_FROM\_TIMESTAMP（指定时间戳）。

校验开始消费时间，集群模式下负载策略allocateMessageQueueStrategy，订阅关系，是否注册消息监听，本地线程数，拉取消息的时间间隔，单次拉取消息的数量以及单次消费的数量。

**拷贝订阅关系**  
就是为集群模式下的消费者配置其对应的重试主题retryTopic并绑定，用于实现消息重试，而广播消息不会做任何消息。

private void copySubscription() throws MQClientException {
        
        try {
        
          // 订阅消息进行拷贝存入rebalanceImpl中
            Map<String, String> sub = this.defaultMQPushConsumer.getSubscription();
            if (sub != null) {
        
                for (final Map.Entry<String, String> entry : sub.entrySet()) {
        
                    final String topic = entry.getKey();
                    final String subString = entry.getValue();
                    SubscriptionData subscriptionData = FilterAPI.buildSubscriptionData(topic, subString);
                    this.rebalanceImpl.getSubscriptionInner().put(topic, subscriptionData);
                }
            }
    
            if (null == this.messageListenerInner) {
        
                this.messageListenerInner = this.defaultMQPushConsumer.getMessageListener();
            }
          // 消息类型，如果是集群消息进行绑定进行失败重试，如果是广播消息不做任何操作不重试直接丢弃
            switch (this.defaultMQPushConsumer.getMessageModel()) {
        
                case BROADCASTING:
                    break;
                case CLUSTERING:
                    final String retryTopic = MixAll.getRetryTopic(this.defaultMQPushConsumer.getConsumerGroup());
                    SubscriptionData subscriptionData = FilterAPI.buildSubscriptionData(retryTopic, SubscriptionData.SUB_ALL);
                    this.rebalanceImpl.getSubscriptionInner().put(retryTopic, subscriptionData);
                    break;
                default:
                    break;
            }
        } catch (Exception e) {
        
            throw new MQClientException("subscription exception", e);
        }
    }

注意：集群消息自动订阅该消费者的重试topic，当消息失败会进行重试，防止消息丢失，广播消息不会做任何操作，当消息失败直接丢弃，消息会产生丢失。

其它操作  
调用getOrCreateMQClientInstance方法获取CreateMQClientInstance实例，设置负载均衡服务rebalanceImpl的相关属性，创建拉取消息核心类PullAPIWrapper。

根据不同的消息模式设置不同位置想Offset文件即消息偏移量，广播模式下Offset文件保存在本地，集群模式下Offset文件保存在远程的Broker端，同时调用load去加载偏移量。

指定指定的MessageListener去创建不同的ConsumeMessageService消息服务，其中存在MessageListenerOrderly(顺序消费)与MessageListenerConcurrently(并发消费)两个类。

启动消费者服务this.consumeMessageService.start()，进行服务的注册，启动CreateMQClientInstance客户端通信实例，初始化netty服务，各种定时任务，拉取消息服务，rebalabceService服务等等，与之前的Producer类似。