RocketMQ源码15- consumer启动流程

落日映苍穹つ 2023-10-08 23:50 121阅读 0赞

本系列RocketMQ4.8注释github地址，希望对大家有所帮助，要是觉得可以的话麻烦给点一下Star哈

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

  1. public class Consumer {
  3.     public static void main(String[] args) throws InterruptedException, MQClientException {
  5.         // todo
  6.         DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name_4");
  8.         consumer.setNamesrvAddr("127.0.0.1:9876");
  10.         consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
  12.         consumer.subscribe("TopicTest", "*");
  14.         // 注册监听器，监听消息
  15.         consumer.registerMessageListener(new MessageListenerConcurrently() {
  17.             @Override
  18.             public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
  19.                 ConsumeConcurrentlyContext context) {
  20.                 // 这里获得了消息
  21.                 System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);
  22.                 return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
  23.             }
  24.         });
  25.         // todo 启动
  26.         consumer.start();
  28.         System.out.printf("Consumer Started.%n");
  29.     }
  30. }
  31. 复制代码

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

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

1. 构造方法：DefaultMQPushConsumer

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

  1. public class DefaultMQPushConsumer extends ClientConfig implements MQPushConsumer {
  3.     private final InternalLogger log = ClientLogger.getLog();
  5.     protected final transient DefaultMQPushConsumerImpl defaultMQPushConsumerImpl;
  7.     // 消费者所属组
  8.     private String consumerGroup;
  9.     // 消息消费模式，分为集群模式、广播模式，默认为集群模式
  10.     private MessageModel messageModel = MessageModel.CLUSTERING;
  11.     // 第一次消费时指定消费策略
  12.     private ConsumeFromWhere consumeFromWhere = ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET;
  14.     private String consumeTimestamp = UtilAll.timeMillisToHumanString3(System.currentTimeMillis() - (1000 * 60 * 30));
  16.     // 集群模式下消息队列的负载策略
  17.     private AllocateMessageQueueStrategy allocateMessageQueueStrategy;
  19.     /**
  20.      * Subscription relationship 订阅信息
  21.      */
  22.     private Map<String /* topic */, String /* sub expression */> subscription = new HashMap<String, String>();
  24.     /**
  25.      * Message listener 消息业务监听器
  26.      */
  27.     private MessageListener messageListener;
  29.     /**
  30.      * Offset Storage 消息消费进度存储器
  31.      */
  32.     private OffsetStore offsetStore;
  34.     /**
  35.      * Minimum consumer thread number 消费者最小线程数
  36.      */
  37.     private int consumeThreadMin = 20;
  39.     /**
  40.      * Max consumer thread number  消费者最大线程数，因为消费
  41.      * 者线程池使用无界队列，所以此参数不生效
  42.      */
  43.     private int consumeThreadMax = 20;
  45.     /**
  46.      * Threshold for dynamic adjustment of the number of thread pool
  47.      */
  48.     private long adjustThreadPoolNumsThreshold = 100000;
  50.     /**
  51.      * Concurrently max span offset.it has no effect on sequential consumption
  52.      * 并发消息消费时处理队列最大跨度，默认2000，表示如果消息处理队列中偏移量最大的消息
  53.      * 与偏移量最小的消息的跨度超过2000，则延迟50ms后再拉取消息
  54.      */
  55.     private int consumeConcurrentlyMaxSpan = 2000;
  57.     /**
  58.      * Flow control threshold on queue level, each message queue will cache at most 1000 messages by default,
  59.      * Consider the {@code pullBatchSize}, the instantaneous value may exceed the limit
  60.      * 默认1000，表示每1000
  61.      * 次流控后打印流控日志
  62.      */
  63.     private int pullThresholdForQueue = 1000;
  64.     private int pullThresholdSizeForTopic = -1;
  66.     /**
  67.      * Message pull Interval
  68.      * 推模式下拉取任务的间隔时间，默
  69.      * 认一次拉取任务完成后继续拉取
  70.      */
  71.     private long pullInterval = 0;
  73.     /**
  74.      * Batch consumption size 消息并发消费时一次
  75.      * 消费消息的条数，通俗点说，就是每次传入
  76.      * MessageListener#consumeMessage中的消息条数
  77.      */
  78.     private int consumeMessageBatchMaxSize = 1;
  80.     /**
  81.      * Batch pull size 每次消息拉取的条数，默认32条
  82.      */
  83.     private int pullBatchSize = 32;
  85.     /**
  86.      * Whether update subscription relationship when every pull
  87.      * 是否每次拉取消息都更
  88.      * 新订阅信息，默认为false
  89.      */
  90.     private boolean postSubscriptionWhenPull = false;
  92.     private boolean unitMode = false;
  93.     /**
  94.      *
  95.      * 最大消费重试次数。如果消息消费次数超过maxReconsumeTimes还未成功，则将该消息转移到一个失败
  96.      * 队列，等待被删除
  97.      */
  98.     private int maxReconsumeTimes = -1;
  100.     /**
  101.      * 延迟将该队列的消
  102.      * 息提交到消费者线程的等待时间，默认延迟1s
  103.      */
  104.     private long suspendCurrentQueueTimeMillis = 1000;
  106.     /**
  107.      * Maximum amount of time in minutes a message may block the consuming thread.
  108.      * 消息消费超时时间，默认为15，
  109.      * 单位为分钟
  110.      */
  111.     private long consumeTimeout = 15;
  112.     ...
  114.     public DefaultMQPushConsumer(final String consumerGroup) {
  115.         // 指定了队列分配策略 AllocateMessageQueueAveragely
  116.         this(null, consumerGroup, null, new AllocateMessageQueueAveragely());
  117.     }
  119.     ...
  120.     public DefaultMQPushConsumer(final String namespace, final String consumerGroup, RPCHook rpcHook,
  121.         AllocateMessageQueueStrategy allocateMessageQueueStrategy) {
  122.         this.consumerGroup = consumerGroup;
  123.         this.namespace = namespace;
  124.         this.allocateMessageQueueStrategy = allocateMessageQueueStrategy;
  125.         defaultMQPushConsumerImpl = new DefaultMQPushConsumerImpl(this, rpcHook);
  126.     }
  127. }
  128. 复制代码

在构造方法中，就只是做了一些成员变量的赋值操作，比较关键的是分配消息队列的策略：allocateMessageQueueStrategy，如果指定，默认就使用AllocateMessageQueueAveragely，即从各队列平均获取消息。

2.客户端start()方法都干了什么？

首先，暴露给开发者的 DefaultMQPushConsumer是一个外观类，真正工作的是其内部的DefaultMQPushConsumerImpl,所以我们看下DefaultMQPushConsumerImpl#start()的逻辑

  1. public synchronized void start() throws MQClientException {
  2.     switch (this.serviceState) {
  3.         case CREATE_JUST:
  4.             log.info("the consumer [{}] start beginning. messageModel={}, isUnitMode={}", this.defaultMQPushConsumer.getConsumerGroup(),
  5.                 this.defaultMQPushConsumer.getMessageModel(), this.defaultMQPushConsumer.isUnitMode());
  6.             this.serviceState = ServiceState.START_FAILED;
  8.             //TODO:检查配置，比如消费者组是否为空，消费模式是否为空，订阅信息是否为空等等
  9.             this.checkConfig();
  11.             // TODO 构建主题订阅信息SubscriptionData并加入RebalanceImpl的订阅消息中
  12.             this.copySubscription();
  14.             if (this.defaultMQPushConsumer.getMessageModel() == MessageModel.CLUSTERING) {
  15.                 this.defaultMQPushConsumer.changeInstanceNameToPID();
  16.             }
  18.             // todo 初始化MQClientInstance、RebalanceImpl（消息重新负载实现类）等
  19.             this.mQClientFactory = MQClientManager.getInstance().getOrCreateMQClientInstance(this.defaultMQPushConsumer, this.rpcHook);
  21.             // 设置负载均衡相关属性
  22.             this.rebalanceImpl.setConsumerGroup(this.defaultMQPushConsumer.getConsumerGroup());
  23.             this.rebalanceImpl.setMessageModel(this.defaultMQPushConsumer.getMessageModel());
  24.             this.rebalanceImpl.setAllocateMessageQueueStrategy(this.defaultMQPushConsumer.getAllocateMessageQueueStrategy());
  25.             this.rebalanceImpl.setmQClientFactory(this.mQClientFactory);
  27.             //TODO: 创建拉取对象的核心类，后面去broker拉取消息时会看到
  28.             this.pullAPIWrapper = new PullAPIWrapper(
  29.                 mQClientFactory,
  30.                 this.defaultMQPushConsumer.getConsumerGroup(), isUnitMode());
  31.             this.pullAPIWrapper.registerFilterMessageHook(filterMessageHookList);
  33.             // todo 初始化消息进度。如果消息消费采用集群模式，那么消
  34.             //息进度存储在Broker上，如果采用广播模式，那么消息消费进度存储
  35.             //在消费端
  36.             if (this.defaultMQPushConsumer.getOffsetStore() != null) {
  37.                 this.offsetStore = this.defaultMQPushConsumer.getOffsetStore();
  38.             } else {
  39.                 switch (this.defaultMQPushConsumer.getMessageModel()) {
  40.                     //TODO: 广播模式创建LocalFileOffsetStore，保存offset到本地文件中
  41.                     case BROADCASTING:
  42.                         this.offsetStore = new LocalFileOffsetStore(this.mQClientFactory, this.defaultMQPushConsumer.getConsumerGroup());
  43.                         break;
  44.                     //TODO:集群模式创建 RemoteBrokerOffsetStore，保存offset到broker文件中
  45.                     case CLUSTERING:
  46.                         this.offsetStore = new RemoteBrokerOffsetStore(this.mQClientFactory, this.defaultMQPushConsumer.getConsumerGroup());
  47.                         break;
  48.                     default:
  49.                         break;
  50.                 }
  51.                 this.defaultMQPushConsumer.setOffsetStore(this.offsetStore);
  52.             }
  53.             // 加载消费信息的偏移量
  54.             this.offsetStore.load();
  56.             // todo 如果是顺序消费，创建消费端消费线程服务。
  57.             //ConsumeMessageService主要负责消息消费，在内部维护一个线程池
  58.             if (this.getMessageListenerInner() instanceof MessageListenerOrderly) {
  59.                 this.consumeOrderly = true;
  60.                 //TODO:创建顺序消费消息服务类
  61.                 this.consumeMessageService =
  62.                     new ConsumeMessageOrderlyService(this, (MessageListenerOrderly) this.getMessageListenerInner());
  63.             } else if (this.getMessageListenerInner() instanceof MessageListenerConcurrently) {
  64.                 this.consumeOrderly = false;
  65.                 //TODO:创建其他消费消息服务类，其内部维护了一个线程池，后面会用到
  66.                 this.consumeMessageService =
  67.                     new ConsumeMessageConcurrentlyService(this, (MessageListenerConcurrently) this.getMessageListenerInner());
  68.             }
  70.             this.consumeMessageService.start();
  72.             // todo 向MQClientInstance注册消费者并启动MQClientInstance，JVM中的所有消费者、生产者持有同一个MQClientInstance，MQClientInstance只会启动一次
  73.             boolean registerOK = mQClientFactory.registerConsumer(this.defaultMQPushConsumer.getConsumerGroup(), this);
  74.             if (!registerOK) {
  75.                 this.serviceState = ServiceState.CREATE_JUST;
  76.                 this.consumeMessageService.shutdown(defaultMQPushConsumer.getAwaitTerminationMillisWhenShutdown());
  77.                 throw new MQClientException("The consumer group[" + this.defaultMQPushConsumer.getConsumerGroup()
  78.                     + "] has been created before, specify another name please." + FAQUrl.suggestTodo(FAQUrl.GROUP_NAME_DUPLICATE_URL),
  79.                     null);
  80.             }
  82.             // todo 启动客户端实例，这个方法非常重要，内部做了很多事情
  83.             mQClientFactory.start();
  84.             log.info("the consumer [{}] start OK.", this.defaultMQPushConsumer.getConsumerGroup());
  85.             this.serviceState = ServiceState.RUNNING;
  86.             break;
  87.         case RUNNING:
  88.         case START_FAILED:
  89.         case SHUTDOWN_ALREADY:
  90.             throw new MQClientException("The PushConsumer service state not OK, maybe started once, "
  91.                 + this.serviceState
  92.                 + FAQUrl.suggestTodo(FAQUrl.CLIENT_SERVICE_NOT_OK),
  93.                 null);
  94.         default:
  95.             break;
  96.     }
  98.     // 更新 topic 的信息，从nameServer获取数据
  99.     this.updateTopicSubscribeInfoWhenSubscriptionChanged();
  100.     this.mQClientFactory.checkClientInBroker();
  101.     // todo 发送心跳，发送到所有的broker
  102.     this.mQClientFactory.sendHeartbeatToAllBrokerWithLock();
  103.     // todo 负载均衡 立即重平衡
  104.     this.mQClientFactory.rebalanceImmediately();
  105. }
  106. 复制代码

2.1 检查配置

  1. 校验GroupName是否为空;校验GroupName是否等于DEFAULT_CONSUMER(等于的话直接抛出异常)
  2. 校验消费模式:集群/广播
  3. 校验ConsumeFromWhere
  4. 校验开始消费的指定时间
  5. 校验AllocateMessageQueueStrategy
  6. 校验订阅关系
  7. 校验是否注册消息监听
  8. 校验消费线程数,consumeThreadMin和 consumeThreadMax 默认值都是20，取值区间都是 [1, 1000]
  9. 校验本地队列缓存消息的最大数，默认是1000，取值范围是[1, 1024], 主要是做流控用的
  10. 校验拉取消息的时间间隔，pullInterval参数，默认是不存在间隔，取值范围是[0, 65535]。当消费速度比生产速度快，可以设置这个参数，避免花费大概率从broker拉取空消息
  11. 校验单次拉取的最大消息数，consumeMessageBatchMaxSize 参数，默认是1，取值范围是[1, 1024]
  12. 校验单次消费的最大消息数, pullBatchSize 参数，默认是32，取值范围是[1, 1024]。

2.2 拷贝订阅关系

将订阅关系设置到重平衡服务类RebalanceImpl中，订阅重试主题消息：

  1. //TODO:key=topic, value=订阅数据（就是tag信息）
  2. protected final ConcurrentMap<String /* topic */, SubscriptionData> subscriptionInner =
  3.     new ConcurrentHashMap<String, SubscriptionData>();
  4. 复制代码

2.3 创建客户端实例 MQClientInstance

  1. // todo 初始化MQClientInstance、RebalanceImpl（消息重新负载实现类）等
  2. this.mQClientFactory = MQClientManager.getInstance().getOrCreateMQClientInstance(this.defaultMQPushConsumer, this.rpcHook);
  3. 复制代码

这个对象在前面分析Producer启动时也看到了，因为Prodcuer/Consumer 都是客户端，所以都会根据这个实例来创建对象。 我们看下这个类的结构：

  1. public class MQClientInstance {
  3.         //TODO:生产者表，producer启动时创建一个新的MQClientInstance实例对象，将生产者信息注册到这里。生产者实例对象中消费者信息是空
  4.         private final ConcurrentMap<String/* group */, MQProducerInner> producerTable = new ConcurrentHashMap<String, MQProducerInner>();
  6.         //TODO:消费者表，consumer启动时创建一个新的MQClientInstance实例对象，将消费者信息注册到这里。消费者实例对象中生产者信息是空
  7.         private final ConcurrentMap<String/* group */, MQConsumerInner> consumerTable = new ConcurrentHashMap<String, MQConsumerInner>();
  9.         //TODO:topic路由信息，producer和consumer都会使用
  10.         private final ConcurrentMap<String/* Topic */, TopicRouteData> topicRouteTable = new ConcurrentHashMap<String, TopicRouteData>();
  12.         //TODO:broker信息，producer和consumer都会用到
  13.         private final ConcurrentMap<String/* Broker Name */, HashMap<Long/* brokerId */, String/* address */>> brokerAddrTable =
  14.             new ConcurrentHashMap<String, HashMap<Long, String>>();
  16.         //TODO......
  19.        /**
  20.         * TODO:构造器
  21.         */
  22.        public MQClientInstance(ClientConfig clientConfig, int instanceIndex, String clientId, RPCHook rpcHook) {
  23.         //TODO:客户端处理器,比如在集群消费模式下，有新的消费者加入，则通知消费者客户端重平衡,是给消费者用的(分析生产者时，我们直接忽略了它）
  24.         this.clientRemotingProcessor = new ClientRemotingProcessor(this);
  26.         //TODO:它的内部会创建netty客户端对象（NettyRemotingClient），用于和broker通信
  27.         this.mQClientAPIImpl = new MQClientAPIImpl(this.nettyClientConfig, this.clientRemotingProcessor, rpcHook, clientConfig);
  29.         //TODO.......
  31.         //TODO:拉取消息的服务，和消费者相关
  32.         this.pullMessageService = new PullMessageService(this);
  34.         //TODO:重平衡服务，和消费者相关
  35.         this.rebalanceService = new RebalanceService(this);
  37.         //TODO:other......
  38.     }
  39. }        
  40. 复制代码

所谓的”客户端”，实际上是在MQClientAPIImpl对象的内部的NettyRemotingClient：

  1. public class MQClientAPIImpl {
  2.     //TODO:.....
  4.     private final RemotingClient remotingClient;
  5.     private final TopAddressing topAddressing;
  6.     private final ClientRemotingProcessor clientRemotingProcessor;
  7.     private String nameSrvAddr = null;
  8.     private ClientConfig clientConfig;
  10.     public MQClientAPIImpl(final NettyClientConfig nettyClientConfig,
  11.         final ClientRemotingProcessor clientRemotingProcessor,
  12.         RPCHook rpcHook, final ClientConfig clientConfig) {
  13.         this.clientConfig = clientConfig;
  14.         topAddressing = new TopAddressing(MixAll.getWSAddr(), clientConfig.getUnitName());
  16.         //TODO:所谓的客户端实际上就是这个，netty客户端 
  17.         this.remotingClient = new NettyRemotingClient(nettyClientConfig, null);
  18.         this.clientRemotingProcessor = clientRemotingProcessor;
  20.         this.remotingClient.registerRPCHook(rpcHook);
  22.         //TODO:注册处理器
  23.         this.remotingClient.registerProcessor(RequestCode.CHECK_TRANSACTION_STATE, this.clientRemotingProcessor, null);
  24.         this.remotingClient.registerProcessor(RequestCode.NOTIFY_CONSUMER_IDS_CHANGED, this.clientRemotingProcessor, null);
  26.      //TODO:....注册其他处理器......
  27.     }
  28. 复制代码

这里我们关注下 NettyRemotingClient，其实前面分析Broker,Producer时，我们经常看到这个，因为RocketMQ是使用netty作为通信的。

NettyRemotingClient：netty的客户端对象
NettyRemotingServer：netty的服务端对象

2.4 创建拉取消息的核心类 PullAPIWrapper

  1. this.pullAPIWrapper = new PullAPIWrapper(
  2.     mQClientFactory,
  3.     this.defaultMQPushConsumer.getConsumerGroup(), isUnitMode());
  4. 复制代码

2.5 创建 offset存储服务 OffsetStore

  • 如果是广播模式(BROADCASTING), 则创建 LocalFileOffsetStore 对象，将消费者的offset存储到本地的，默认文件路径为当前用户主目录下的 .rocketmq_offsets/clientId/{clientId}/clientId/{group}/Offsets.json。其中clientId为当前消费者id，默认为ip@default,{clientId}为当前消费者id，默认为ip@default, clientId为当前消费者id，默认为ip@default,{group}为消费者组名称
  • 如果是集群模式(CLUSTERING), 则创建 RemoteBrokerOffsetStore对象，将消费者的offset存储到broker中，文件路径为当前用户主目录下的store/config/consumerOffset.json

2.6 创建消费消息的服务类 ConsumeMessageService

  • 如果是顺序消费，则创建 ConsumeMessageOrderlyService 对象
  • 如果是其他消费，则创建 ConsumeMessageConcurrentlyService 对象，同时内部也会创建一个ThreadPoolExecutor线程池，这个线程池非常的重要，拉取到消息后会将消息提交到这个线程池中给消费者消费

2.7 将consumer注册到本地

  1. boolean registerOK = mQClientFactory.registerConsumer(this.defaultMQPushConsumer.getConsumerGroup(), this);
  2. 复制代码

将消费者组信息添加到本地客户端实例MQClientInstance的 consumerTable Map中，key=groupName; value=DefaultMQPushConsumerImpl,就是消费者对象

2.8 启动客户端实例（重要）

  1. mQClientFactory.start();
  2. 复制代码

其内部做了很多事情,主要如下： org.apache.rocketmq.client.impl.factory.MQClientInstance#start：

  1. public void start() throws MQClientException {
  3.     synchronized (this) {
  4.         switch (this.serviceState) {
  5.             case CREATE_JUST:
  6.                 // 先设置 失败
  7.                 this.serviceState = ServiceState.START_FAILED;
  8.                 // If not specified,looking address from name server
  9.                 // 判断namesrv 是否为null
  10.                 if (null == this.clientConfig.getNamesrvAddr()) {
  11.                     this.mQClientAPIImpl.fetchNameServerAddr();
  12.                 }
  13.                 // todo Start request-response channel
  14.                 // 启动远程服务，这个方法只是装配了netty客户端相关配置
  15.                 // todo 注意：1. 这里是netty客户端，2. 这里并没有创建连接
  16.                 this.mQClientAPIImpl.start();
  17.                 // Start various schedule tasks
  18.                 // todo 开启任务调度
  19.                 this.startScheduledTask();
  20.                 // Start pull service
  21.                 // todo 开启 拉取服务 仅对consumer启作用
  22.                 this.pullMessageService.start();
  23.                 // Start rebalance service
  24.                 // todo 开启平衡服务 仅对consumer启作用
  25.                 this.rebalanceService.start();
  26.                 // Start push service
  27.                 // todo 启用内部的 producer
  28.                 this.defaultMQProducer.getDefaultMQProducerImpl().start(false);
  29.                 log.info("the client factory [{}] start OK", this.clientId);
  30.                 // 设置状态为running
  31.                 this.serviceState = ServiceState.RUNNING;
  32.                 break;
  33.             case START_FAILED:
  34.                 throw new MQClientException("The Factory object[" + this.getClientId() + "] has been created before, and failed.", null);
  35.             default:
  36.                 break;
  37.         }
  38.     }
  39. }
  40. 复制代码

在producer的启动过程中，也会调用这个方法，前面我们已经分析过了一波了，这次我们在consumer的角度再来分析这个方法。

该方法所做的工作如下：

  1. 获取 nameServer 的地址
  2. 启动客户端的远程服务，这个方法会配置netty客户端
  3. 启动定时任务
  4. 启动拉取消息服务
  5. 启动负载均衡服务

上面的1，2与producer的流程并无区别，就不再分析了，我们来看看接下来的任务

2.8.1 启动远程netty客户端

  1. // Start request-response channel
  2. this.mQClientAPIImpl.start();
  3. 复制代码
  4. 复制代码

就是启动其内部的 NettyRemotingClient,用于和broker通信。

2.8.2 启动各种定时任务

  1. // Start various schedule tasks
  2. this.startScheduledTask();
  3. 复制代码

那么有哪些定时任务呢？继续往里看,罗列几个特别关注的

2.8.2.1 发送心跳到Broker

  1. this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
  3.     @Override
  4.     public void run() {
  5.         try {
  6.             // todo 清除下线broker
  7.             MQClientInstance.this.cleanOfflineBroker();
  8.             // todo 发送心跳到所有broker上面
  9.             MQClientInstance.this.sendHeartbeatToAllBrokerWithLock();
  10.         } catch (Exception e) {
  11.             log.error("ScheduledTask sendHeartbeatToAllBroker exception", e);
  12.         }
  13.     }
  14. }, 1000, this.clientConfig.getHeartbeatBrokerInterval(), TimeUnit.MILLISECONDS);
  15. 复制代码

延迟1s执行，每隔30s发送一次心跳包

2.8.2.2 持久化消费者的 offset

  1. // // 持久化消费者的消费偏移量，每5秒一次
  2. this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
  4.     @Override
  5.     public void run() {
  6.         try {
  7.             // todo 持久化consumer offset 可以放在本地文件，也可以推送到 broker
  8.             MQClientInstance.this.persistAllConsumerOffset();
  9.         } catch (Exception e) {
  10.             log.error("ScheduledTask persistAllConsumerOffset exception", e);
  11.         }
  12.     }
  13. }, 1000 * 10, this.clientConfig.getPersistConsumerOffsetInterval(), TimeUnit.MILLISECONDS);
  14. 复制代码

延迟10s执行，每隔5s持久化一次offset
这里的持久化是将本地Map中offset发送到broker中，然后broker中的定时任务写到文件中，完成真正的持久化，后面会看到。

2.8.3 启动拉取消息的服务PullMessageService(重要）

  1. // Start pull service
  2. this.pullMessageService.start();
  3. 复制代码

他是一个异步线程，其核心逻辑是

  1. public void run() {
  2.     log.info(this.getServiceName() + " service started");
  4.     // Stopped 声明为volatile，每执行一次业务逻辑，检测一下其运行状态，可以
  5.     //通过其他线程将Stopped设置为true，从而停止该线程
  6.     while (!this.isStopped()) {
  7.         try {
  8.             // 从pullRequestQueue中获取一个PullRequest消息拉取任务，
  9.             //如果pullRequestQueue为空，则线程将阻塞，直到有拉取任务被放入
  10.             PullRequest pullRequest = this.pullRequestQueue.take();
  11.             // todo 调用pullMessage方法进行消息拉取
  12.             this.pullMessage(pullRequest);
  13.         } catch (InterruptedException ignored) {
  14.         } catch (Exception e) {
  15.             log.error("Pull Message Service Run Method exception", e);
  16.         }
  17.     }
  19.     log.info(this.getServiceName() + " service end");
  20. }
  21. 复制代码

他会监听阻塞队列pullRequestQueue,当队列是空的时候，他会一直阻塞，如果不为空，则获取PullRequest对象，去拉取消息，这个逻辑后面再说。

刚开始肯定是阻塞的，我们要看什么时候往队列中放入值，以及放入值之后做什么

2.8.4 启动重平衡服务 RebalanceService

  1. // Start rebalance service
  2. this.rebalanceService.start();
  3. 复制代码
  4. 复制代码

它也是一个异步线程，其核心逻辑是

  1. public void run() {
  2.     log.info(this.getServiceName() + " service started");
  5.     while (!this.isStopped()) {
  6.         // todo 等待20s执行一次 内部使用了juc的CountDownLatch, 使得这里启动后仍然是阻塞的
  7.         this.waitForRunning(waitInterval);
  8.         // todo
  9.         this.mqClientFactory.doRebalance();
  10.     }
  12.     log.info(this.getServiceName() + " service end");
  13. }
  14. 复制代码

他是重平衡的核心逻辑，但是在启动时，由于使用了JUC的 CountDownLatch锁，使其不会立即重平衡，而是阻塞，什么时候出发重平衡呢？我们还是继续往后看

2.9 发送心跳到broker

  1. this.updateTopicSubscribeInfoWhenSubscriptionChanged();
  2. this.mQClientFactory.checkClientInBroker();
  3. //TODO: 发送心跳到broker
  4. this.mQClientFactory.sendHeartbeatToAllBrokerWithLock();
  5. this.mQClientFactory.rebalanceImmediately();
  6. 复制代码

是不是有些疑惑？前面的定时任务中不是已经启动了心跳服务吗，为什么这里还要启动呢？我也不清楚，猜测是因为担心网络问题导致没有及时发送给broker吧

2.10 立即启动重平衡服务RebalanceService

  1. //TODO:立即启动重平衡服务
  2. this.mQClientFactory.rebalanceImmediately();
  3. 复制代码

前面在 2.8.4的时候，启动了重平衡服务，但是因为 CountDownLatch 导致阻塞了，这里就是唤醒，可以执行重平衡的逻辑。

这里先不关注它的内部逻辑,后面在分析。

其实，到这里Consumer客户端启动流程就结束了，至于如何拉取消息，也是这些服务类的相互配合工作的，后面我们在分析中在仔细分析它是如何一步一步拉取消息并消费的。

3. 总结

消费者启动过程相比生产者启动过程要复杂一些，会启动很多对象（实际上生产者也启动了，因为他们都是基于客户端实例MQClientInstance去创建的对象，只不过生产者并不会使用某些服务类）

简单总结下：

  1. 创建客户端实例MQClientInstance（其内部要创建netty客户端对象NettyRemotingClient）
  2. 启动客户端实例，其内部要启动netty客户端，消息拉取服务，以及重平衡服务，还有各种定时任务
  3. 发送心跳到broker

好了，Consumer启动过程就分析到这里，接下来就分析下消费过程

发表评论

表情：
评论列表 （有 0 条评论，121人围观）

还没有评论，来说两句吧...

相关阅读