RocketMQ源码解析五（Consumer启动流程）

骑猪看日落 2022-09-12 07:46 76阅读 0赞

RocketMQ版本4.6.0，记录自己看源码的过程

Consumer示例：

  1. public class Consumer { 
  3.     public static void main(String[] args) throws InterruptedException, MQClientException { 
  5.         DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumerGroup3");
  6.         consumer.setNamesrvAddr("localhost:9876");
  7.         consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
  8.         consumer.subscribe("TestTopic", "*");
  9.         consumer.registerMessageListener(new MessageListenerConcurrently() { 
  11.             @Override
  12.             public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
  13.                 ConsumeConcurrentlyContext context) { 
  14.                 System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);
  15.                 if (msgs.get(0).getReconsumeTimes() >= 5) { 
  16.                     return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
  17.                 }
  18.                 return ConsumeConcurrentlyStatus.RECONSUME_LATER;
  19.             }
  20.         });
  21.         consumer.start();
  23.         System.out.printf("Consumer Started.%n");
  24.     }
  25. }

先创建消费者DefaultMQPushConsumer

  1. public DefaultMQPushConsumer(final String consumerGroup) { 
  2.     // 这里可以看出负载均衡策略使用平均算法
  3.     this(null, consumerGroup, null, new AllocateMessageQueueAveragely());
  4. }
  6. public DefaultMQPushConsumer(final String namespace, final String consumerGroup, RPCHook rpcHook,
  7.     AllocateMessageQueueStrategy allocateMessageQueueStrategy) { 
  8.     this.consumerGroup = consumerGroup;
  9.     this.namespace = namespace;
  10.     this.allocateMessageQueueStrategy = allocateMessageQueueStrategy;
  11.     defaultMQPushConsumerImpl = new DefaultMQPushConsumerImpl(this, rpcHook);
  12. }

创建过程没什么操作，指定了负载均衡策略以及也跟生产者一样实例化了一个内部消费者实例DefaultMQPushConsumerImpl，将消费者的功能委托给这个类。
接着设置Namesrv地址，从哪开始消费，订阅主题（同时会将订阅信息复制到重平衡组件）以及注册一个消息监听器，用来消费消息。最后启动。
DefaultMQPushConsumer

  1. @Override
  2. public void start() throws MQClientException { 
  3.     setConsumerGroup(NamespaceUtil.wrapNamespace(this.getNamespace(), this.consumerGroup));
  4.     this.defaultMQPushConsumerImpl.start();
  5.     if (null != traceDispatcher) { 
  6.         try { 
  7.             traceDispatcher.start(this.getNamesrvAddr(), this.getAccessChannel());
  8.         } catch (MQClientException e) { 
  9.             log.warn("trace dispatcher start failed ", e);
  10.         }
  11.     }
  12. }

可以看到也是通过DefaultMQPushConsumerImpl来启动

  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;
  7.             // 主要是对消费者的配置进行检查
  8.             this.checkConfig();
  9.             // 将订阅数据复制到重平衡组件中，并且如果是集群模式，也会订阅retryTopic，
  10.             // 然后等consumer向broker发送心跳时在broker中创建该重试主题
  11.             this.copySubscription();
  13.             if (this.defaultMQPushConsumer.getMessageModel() == MessageModel.CLUSTERING) { 
  14.                 this.defaultMQPushConsumer.changeInstanceNameToPID();
  15.             }
  16.             // 从缓存获取或创建一个新的MQClientInstance实例
  17.             this.mQClientFactory = MQClientManager.getInstance().getOrCreateMQClientInstance(this.defaultMQPushConsumer, this.rpcHook);
  18.             // 为重平衡组件设置必要的属性
  19.             this.rebalanceImpl.setConsumerGroup(this.defaultMQPushConsumer.getConsumerGroup());
  20.             this.rebalanceImpl.setMessageModel(this.defaultMQPushConsumer.getMessageModel());
  21.             this.rebalanceImpl.setAllocateMessageQueueStrategy(this.defaultMQPushConsumer.getAllocateMessageQueueStrategy());
  22.             this.rebalanceImpl.setmQClientFactory(this.mQClientFactory);
  23.             // 从这里也可以看出push底层也是pull
  24.             this.pullAPIWrapper = new PullAPIWrapper(
  25.                 mQClientFactory,
  26.                 this.defaultMQPushConsumer.getConsumerGroup(), isUnitMode());
  27.             this.pullAPIWrapper.registerFilterMessageHook(filterMessageHookList);
  29.             if (this.defaultMQPushConsumer.getOffsetStore() != null) { 
  30.                 this.offsetStore = this.defaultMQPushConsumer.getOffsetStore();
  31.             } else { 
  32.                 switch (this.defaultMQPushConsumer.getMessageModel()) { 
  33.                     // 广播模式，消费组里的每个消费者都会收到全部消息，offset消费端自己管理
  34.                     case BROADCASTING:
  35.                         this.offsetStore = new LocalFileOffsetStore(this.mQClientFactory, this.defaultMQPushConsumer.getConsumerGroup());
  36.                         break;
  37.                     // 集群模式，消息平均分配给消费组中的每个消费者，offset由broker管理
  38.                     case CLUSTERING:
  39.                         this.offsetStore = new RemoteBrokerOffsetStore(this.mQClientFactory, this.defaultMQPushConsumer.getConsumerGroup());
  40.                         break;
  41.                     default:
  42.                         break;
  43.                 }
  44.                 this.defaultMQPushConsumer.setOffsetStore(this.offsetStore);
  45.             }
  46.             // 集群模式是个空实现，只有广播模式有用，实现从本地文件加载进度
  47.             this.offsetStore.load();
  48.             // 根据并发消费还是顺序消费创建不同的 消费消息服务
  49.             if (this.getMessageListenerInner() instanceof MessageListenerOrderly) { 
  50.                 this.consumeOrderly = true;
  51.                 this.consumeMessageService =
  52.                     new ConsumeMessageOrderlyService(this, (MessageListenerOrderly) this.getMessageListenerInner());
  53.             } else if (this.getMessageListenerInner() instanceof MessageListenerConcurrently) { 
  54.                 this.consumeOrderly = false;
  55.                 this.consumeMessageService =
  56.                     new ConsumeMessageConcurrentlyService(this, (MessageListenerConcurrently) this.getMessageListenerInner());
  57.             }
  58.             //
  59.             this.consumeMessageService.start();
  60.             // 注册消费者缓存
  61.             boolean registerOK = mQClientFactory.registerConsumer(this.defaultMQPushConsumer.getConsumerGroup(), this);
  62.             if (!registerOK) { 
  63.                 this.serviceState = ServiceState.CREATE_JUST;
  64.                 this.consumeMessageService.shutdown();
  65.                 throw new MQClientException("The consumer group[" + this.defaultMQPushConsumer.getConsumerGroup()
  66.                     + "] has been created before, specify another name please." + FAQUrl.suggestTodo(FAQUrl.GROUP_NAME_DUPLICATE_URL),
  67.                     null);
  68.             }
  70.             // 启动一些功能组件和后台任务，比如netty客户端启动，拉取消息服务，重平衡服务
  71.             mQClientFactory.start();
  72.             log.info("the consumer [{}] start OK.", this.defaultMQPushConsumer.getConsumerGroup());
  73.             this.serviceState = ServiceState.RUNNING;
  74.             break;
  75.         case RUNNING:
  76.         case START_FAILED:
  77.         case SHUTDOWN_ALREADY:
  78.             throw new MQClientException("The PushConsumer service state not OK, maybe started once, "
  79.                 + this.serviceState
  80.                 + FAQUrl.suggestTodo(FAQUrl.CLIENT_SERVICE_NOT_OK),
  81.                 null);
  82.         default:
  83.             break;
  84.     }
  86.     this.updateTopicSubscribeInfoWhenSubscriptionChanged();
  87.     this.mQClientFactory.checkClientInBroker();
  88.     // 发送心跳到每个broker
  89.     this.mQClientFactory.sendHeartbeatToAllBrokerWithLock();
  90.     // 启动消费者需立即发起一次重平衡，通过CountdownLatch -1将重平衡线程立刻唤醒，如果线程已经在执行状态，则这里不会再重平衡
  91.     this.mQClientFactory.rebalanceImmediately();
  92. }

Step1：检查消费者配置。
Step2：将订阅数据复制到重平衡组件中，并且如果是集群模式，也会订阅retryTopic，然后等consumer向broker发送心跳时在broker中创建该重试主题。
Step3：从缓存获取或创建一个新的MQClientInstance实例，这里面创建了包括拉取消息服务和重平衡服务等。
MQClientManager

  1. /** * 从缓存获取或创建一个新的MQClientInstance实例 */
  2. public MQClientInstance getOrCreateMQClientInstance(final ClientConfig clientConfig, RPCHook rpcHook) { 
  3.     String clientId = clientConfig.buildMQClientId();
  4.     MQClientInstance instance = this.factoryTable.get(clientId);
  5.     // 根据clientId从缓存中获取，没获取到则创建一个放到缓存中
  6.     if (null == instance) { 
  7.         instance =
  8.             new MQClientInstance(clientConfig.cloneClientConfig(),
  9.                 this.factoryIndexGenerator.getAndIncrement(), clientId, rpcHook);
  10.         MQClientInstance prev = this.factoryTable.putIfAbsent(clientId, instance);
  11.         if (prev != null) { 
  12.             instance = prev;
  13.             log.warn("Returned Previous MQClientInstance for clientId:[{}]", clientId);
  14.         } else { 
  15.             log.info("Created new MQClientInstance for clientId:[{}]", clientId);
  16.         }
  17.     }
  19.     return instance;
  20. }

整个JVM实例中只存在一个MQClientInstance实例，并维护一个MQClientInstance缓存factoryTable，一个clientId只会创建一个MQClientInstance。clientId = 客户端IP + “@” + instanceName，同一个JVM内，生产者和消费者都是用一个MQClientInstance实例。
Step4：为重平衡组件设置必要的属性。
Step5：创建Consumer端的消费进度管理组件。
Step6：从消费进度组件加载消费进度，只有广播模式有用，实现从本地文件加载进度。
Step7：根据并发消费还是顺序消费创建不同的消费消息服务。
Step8：向MQClientInstance注册消费者实例，如果相同消费组已经存在一个实例了，则会报错。
Step9：启动MQClientInstance实例，启动一些功能组件和后台任务，比如netty客户端启动，拉取消息服务，重平衡服务。

  1. /** * 一个实例只会启动一次 */
  2. public void start() throws MQClientException { 
  4.     synchronized (this) { 
  5.         switch (this.serviceState) { 
  6.             case CREATE_JUST:
  7.                 this.serviceState = ServiceState.START_FAILED;
  8.                 // If not specified,looking address from name server
  9.                 if (null == this.clientConfig.getNamesrvAddr()) { 
  10.                     this.mQClientAPIImpl.fetchNameServerAddr();
  11.                 }
  12.                 // Start request-response channel 启动netty客户端
  13.                 this.mQClientAPIImpl.start();
  14.                 // 启动一些定时任务，比如定时从NameServer拉取路由信息，定时向broker发送心跳等
  15.                 this.startScheduledTask();
  16.                 // 启动拉取消息服务
  17.                 this.pullMessageService.start();
  18.                 // 启动重平衡服务
  19.                 this.rebalanceService.start();
  20.                 // 用来消费者往broker发送重试消息？？
  21.                 this.defaultMQProducer.getDefaultMQProducerImpl().start(false);
  22.                 log.info("the client factory [{}] start OK", this.clientId);
  23.                 this.serviceState = ServiceState.RUNNING;
  24.                 break;
  25.             case START_FAILED:
  26.                 throw new MQClientException("The Factory object[" + this.getClientId() + "] has been created before, and failed.", null);
  27.             default:
  28.                 break;
  29.         }
  30.     }
  31. }

到这消费者就算启动完成了，接着就交给几个定时任务了。
在这里插入图片描述

参考资料
《儒猿技术窝——从 0 开始带你成为消息中间件实战高手》

发表评论

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

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

相关阅读