rocketmq源码剖析之producer和consumer的启动关闭

浅浅的花香味﹌ 2023-06-09 10:05 146阅读 0赞

producer在rocketmq的作用是消息的生产者，consumer在rocketmq的作用是消息的消费者，它的生命周期是跟项目相关的，即是由使用者控制的。而为什么要将这两个角色的启动关闭流程放在一起剖析呢？是因为他们都是MQ的客户端，在启动和关闭的行为上，有着很多共同的地方。接下来便将会来仔细探究其启动和关闭的过程。

Producer

DefaultMQProducer

DefaultMQProducer的启动

producer在启动的时候会做一系列的内部初始化，其启动的源码如下所示：

  1. public void start() throws MQClientException { 
  2.         this.setProducerGroup(withNamespace(this.producerGroup));
  3.         this.defaultMQProducerImpl.start();
  4.         if (null != traceDispatcher) { 
  5.             try { 
  6.                 traceDispatcher.start(this.getNamesrvAddr(), this.getAccessChannel());
  7.             } catch (MQClientException e) { 
  8.                 log.warn("trace dispatcher start failed ", e);
  9.             }
  10.         }
  11.     }

从简短的代码可以看见其做的事情是对producer的group的设置、核心初始化、trace的初始化。接下来便仔细说明下核心初始化过程，其余是其和consumer所共有的，会在公共行为部分进行剖析。

producer核心初始化

producer启动初始化化部分核心逻辑如下，第一次启动的时候它会执行CREATE_JUST分支：

  1. public void start(final boolean startFactory) throws MQClientException { 
  2.         switch (this.serviceState) { 
  3.             case CREATE_JUST:
  4.                 this.serviceState = ServiceState.START_FAILED;
  5.                 // 校验group参数
  6.                 this.checkConfig();
  8.                 if (!this.defaultMQProducer.getProducerGroup().equals(MixAll.CLIENT_INNER_PRODUCER_GROUP)) { 
  9.                     // 设置producer的实例名
  10.                     this.defaultMQProducer.changeInstanceNameToPID();
  11.                 }
  12.                 // 创建MQClient实例
  13.                 this.mQClientFactory = MQClientManager.getInstance().getAndCreateMQClientInstance(this.defaultMQProducer, rpcHook);
  14.                 // 向MQ客户端的producer表中注册
  15.                 boolean registerOK = mQClientFactory.registerProducer(this.defaultMQProducer.getProducerGroup(), this);
  16.                 if (!registerOK) { 
  17.                     this.serviceState = ServiceState.CREATE_JUST;
  18.                     throw new MQClientException("The producer group[" + this.defaultMQProducer.getProducerGroup()
  19.                         + "] has been created before, specify another name please." + FAQUrl.suggestTodo(FAQUrl.GROUP_NAME_DUPLICATE_URL),
  20.                         null);
  21.                 }
  22.                 // 向当前的producer的topic表中注册一个用于测试或demo用的topic “TBW102”
  23.                 this.topicPublishInfoTable.put(this.defaultMQProducer.getCreateTopicKey(), new TopicPublishInfo());
  25.                 if (startFactory) { 
  26.                     // 启动MQClient
  27.                     mQClientFactory.start();
  28.                 }
  30.                 log.info("the producer [{}] start OK. sendMessageWithVIPChannel={}", this.defaultMQProducer.getProducerGroup(),
  31.                     this.defaultMQProducer.isSendMessageWithVIPChannel());
  32.                 this.serviceState = ServiceState.RUNNING;
  33.                 break;
  34.             case RUNNING:
  35.             case START_FAILED:
  36.             case SHUTDOWN_ALREADY:
  37.                 throw new MQClientException("The producer service state not OK, maybe started once, "
  38.                     + this.serviceState
  39.                     + FAQUrl.suggestTodo(FAQUrl.CLIENT_SERVICE_NOT_OK),
  40.                     null);
  41.             default:
  42.                 break;
  43.         }
  44.         // broker的心跳机制发送
  45.         this.mQClientFactory.sendHeartbeatToAllBrokerWithLock();
  46.     }

浏览过这段代码，便可以很容易看出其中的核心逻辑在于MQClient的启动，以及producer与broker之间的心跳触发，这两部分是与consumer所共有的，会在公共行为部分进行剖析。

DefaultMQProducer的关闭

producer的关闭行为是producer从RUNNING状态到SHUTDOWN_ALREADY状态的过程，主要做了三件事：

  • 从注册表中清除了本身实例；
  • 关闭了线程池资源；
  • 清楚了client的资源；

核心代码如下所示：

  1. this.mQClientFactory.unregisterProducer(this.defaultMQProducer.getProducerGroup());
  2. this.defaultAsyncSenderExecutor.shutdown();
  3. if (shutdownFactory) { 
  4.     this.mQClientFactory.shutdown();
  5. }

TransactionMQProducer

TransactionMQProducer的启动

事务生产者TransactionMQProducer是默认的生产者的一个增强，它在启动方面多了一步操作：

  1. public void start() throws MQClientException { 
  2.         this.defaultMQProducerImpl.initTransactionEnv();
  3.         super.start();
  4.     }

可以看出多的这一步则是在事务方面的一个初始化，即事务检查的线程池初始化，其源码如下所示：

  1. public void initTransactionEnv() { 
  2.         TransactionMQProducer producer = (TransactionMQProducer) this.defaultMQProducer;
  3.         if (producer.getExecutorService() != null) { 
  4.             this.checkExecutor = producer.getExecutorService();
  5.         } else { 
  6.             this.checkRequestQueue = new LinkedBlockingQueue<Runnable>(producer.getCheckRequestHoldMax());
  7.             this.checkExecutor = new ThreadPoolExecutor(
  8.                 producer.getCheckThreadPoolMinSize(),
  9.                 producer.getCheckThreadPoolMaxSize(),
  10.                 1000 * 60,
  11.                 TimeUnit.MILLISECONDS,
  12.                 this.checkRequestQueue);
  13.         }
  14.     }

TransactionMQProducer的关闭

同样，事务生产者在关闭的时候也要多一步关闭事务事务检查线程池资源的操作，具体就不在详细说明。

Consumer

PushConsumer

PushConsumer的启动

push模式的消费者同生产者一样，在启动的时候依然是先包装group，然后在执行核心流程，最后在根据情况初始化trace。其源码如下：

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

接下来便仔细说明下核心初始化过程，其余是其和producer所共有的，会在公共行为部分进行剖析。

PushConsumer核心初始化

这里直接附上核心的start过程源码。其一样的走CREATE_JUST分支，如下所示：

  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.                 // 校验group参数 
  8.                 this.checkConfig();
  9.                 // 进行拷贝注册（拷贝topic，监听器等）
  10.                 this.copySubscription();
  12.                 if (this.defaultMQPushConsumer.getMessageModel() == MessageModel.CLUSTERING) { 
  13.                     // 如果是集群模式则根据PID改变实例名
  14.                     this.defaultMQPushConsumer.changeInstanceNameToPID();
  15.                 }
  16.                 // 创建MQClient实例
  17.                 this.mQClientFactory = MQClientManager.getInstance().getAndCreateMQClientInstance(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);
  24.                 // pull-api包装
  25.                 this.pullAPIWrapper = new PullAPIWrapper(
  26.                     mQClientFactory,
  27.                     this.defaultMQPushConsumer.getConsumerGroup(), isUnitMode());
  28.                 // 注册消息过滤钩子
  29.                 this.pullAPIWrapper.registerFilterMessageHook(filterMessageHookList);
  31.                 // 复制offset-store
  32.                 if (this.defaultMQPushConsumer.getOffsetStore() != null) { 
  33.                     this.offsetStore = this.defaultMQPushConsumer.getOffsetStore();
  34.                 } else { 
  35.                     switch (this.defaultMQPushConsumer.getMessageModel()) { 
  36.                         case BROADCASTING:
  37.                             // 广播模式设置为本地文件存储模式
  38.                             this.offsetStore = new LocalFileOffsetStore(this.mQClientFactory, this.defaultMQPushConsumer.getConsumerGroup());
  39.                             break;
  40.                         case CLUSTERING:
  41.                             // 集群模式将偏移两设置为远程broker存储
  42.                             this.offsetStore = new RemoteBrokerOffsetStore(this.mQClientFactory, this.defaultMQPushConsumer.getConsumerGroup());
  43.                             break;
  44.                         default:
  45.                             break;
  46.                     }
  47.                     this.defaultMQPushConsumer.setOffsetStore(this.offsetStore);
  48.                 }
  49.                 // 加载消费偏移情况
  50.                 this.offsetStore.load();
  52.                 if (this.getMessageListenerInner() instanceof MessageListenerOrderly) { 
  53.                     // 有序消费监听器
  54.                     this.consumeOrderly = true;
  55.                     this.consumeMessageService =
  56.                         new ConsumeMessageOrderlyService(this, (MessageListenerOrderly) this.getMessageListenerInner());
  57.                 } else if (this.getMessageListenerInner() instanceof MessageListenerConcurrently) { 
  58.                     // 并发消费监听器
  59.                     this.consumeOrderly = false;
  60.                     this.consumeMessageService =
  61.                         new ConsumeMessageConcurrentlyService(this, (MessageListenerConcurrently) this.getMessageListenerInner());
  62.                 }
  64.                 // 启动消费消息的服务（这里的start其实是启动了一个定时任务用于清理消费超时的消息）
  65.                 this.consumeMessageService.start();
  67.                 //向MQ客户端的consumer表中注册
  68.                 boolean registerOK = mQClientFactory.registerConsumer(this.defaultMQPushConsumer.getConsumerGroup(), this);
  69.                 if (!registerOK) { 
  70.                     this.serviceState = ServiceState.CREATE_JUST;
  71.                     this.consumeMessageService.shutdown();
  72.                     throw new MQClientException("The consumer group[" + this.defaultMQPushConsumer.getConsumerGroup()
  73.                         + "] has been created before, specify another name please." + FAQUrl.suggestTodo(FAQUrl.GROUP_NAME_DUPLICATE_URL),
  74.                         null);
  75.                 }
  76.                 // mqclient启动
  77.                 mQClientFactory.start();
  78.                 log.info("the consumer [{}] start OK.", this.defaultMQPushConsumer.getConsumerGroup());
  79.                 this.serviceState = ServiceState.RUNNING;
  80.                 break;
  81.             case RUNNING:
  82.             case START_FAILED:
  83.             case SHUTDOWN_ALREADY:
  84.                 throw new MQClientException("The PushConsumer service state not OK, maybe started once, "
  85.                     + this.serviceState
  86.                     + FAQUrl.suggestTodo(FAQUrl.CLIENT_SERVICE_NOT_OK),
  87.                     null);
  88.             default:
  89.                 break;
  90.         }
  92.         // 更新topic注册数据
  93.         this.updateTopicSubscribeInfoWhenSubscriptionChanged();
  94.         // 检测topic所在的broker是否能够连接成功
  95.         this.mQClientFactory.checkClientInBroker();
  96.         // 触发想broker的心跳
  97.         this.mQClientFactory.sendHeartbeatToAllBrokerWithLock();
  98.         // 直接触发一次平衡操作
  99.         this.mQClientFactory.rebalanceImmediately();
  100.     }

这里来看一下消费偏移的load情况，消费偏移的加载分从远程broker和本地文件加载，远程由RemoteBrokerOffsetStore来实现，本地由LocalFileOffsetStore来实现，具体使用哪儿种取决于是集群模式还是广播模式，者同样是影响消费偏移是持久化到本地还是远程的问题。

远程的加载其实没有做任何事情，源码如下所示：

  1. @Override
  2.     public void load() { 
  3.     }

从本地文件加载则是会根据是否从文件中读取到内容决定，具体源码如下：

  1. public void load() throws MQClientException { 
  2.         // 从本地文件读取内容到包装器
  3.         OffsetSerializeWrapper offsetSerializeWrapper = this.readLocalOffset();
  4.         if (offsetSerializeWrapper != null && offsetSerializeWrapper.getOffsetTable() != null) { 
  5.             // 设置偏移量表
  6.             offsetTable.putAll(offsetSerializeWrapper.getOffsetTable());
  7.             // 打印日志
  8.             for (MessageQueue mq : offsetSerializeWrapper.getOffsetTable().keySet()) { 
  9.                 AtomicLong offset = offsetSerializeWrapper.getOffsetTable().get(mq);
  10.                 log.info("load consumer's offset, {} {} {}",
  11.                     this.groupName,
  12.                     mq,
  13.                     offset.get());
  14.             }
  15.         }
  16.     }

PushConsumer的关闭

push模式consumer在关闭的时候清理资源的情况如下所示：

  1. public synchronized void shutdown() { 
  2.         switch (this.serviceState) { 
  3.             case CREATE_JUST:
  4.                 break;
  5.             case RUNNING:
  6.                 // 关闭消费消息服务
  7.                 this.consumeMessageService.shutdown();
  8.                 // 持久化消息的消费情况
  9.                 this.persistConsumerOffset();
  10.                 // 从注册表移除当前的消费者
  11.                 this.mQClientFactory.unregisterConsumer(this.defaultMQPushConsumer.getConsumerGroup());
  12.                 // 关闭MQClient资源
  13.                 this.mQClientFactory.shutdown();
  14.                 log.info("the consumer [{}] shutdown OK", this.defaultMQPushConsumer.getConsumerGroup());
  15.                 this.rebalanceImpl.destroy();
  16.                 this.serviceState = ServiceState.SHUTDOWN_ALREADY;
  17.                 break;
  18.             case SHUTDOWN_ALREADY:
  19.                 break;
  20.             default:
  21.                 break;
  22.         }
  23.     }

PullConsumer

PullConsumer的启动

pull模式的消费者在启动的时候则只有包装group和执行核心流程，其源码如下：

  1. public void start() throws MQClientException { 
  2.         this.setConsumerGroup(NamespaceUtil.wrapNamespace(this.getNamespace(), this.consumerGroup));
  3.         this.defaultMQPullConsumerImpl.start();
  4.     }

接下来便仔细说明下核心初始化过程，包装group部分会在公共行为部分进行剖析。

PullConsumer核心初始化

这里直接附上核心的start过程源码，其一样的走CREATE_JUST分支，如下所示：

  1. public synchronized void start() throws MQClientException { 
  2.         switch (this.serviceState) { 
  3.             case CREATE_JUST:
  4.                 this.serviceState = ServiceState.START_FAILED;
  5.                 // 相关配置检查
  6.                 this.checkConfig();
  7.                 // 进行拷贝注册
  8.                 this.copySubscription();
  10.                 if (this.defaultMQPullConsumer.getMessageModel() == MessageModel.CLUSTERING) { 
  11.                     // 如果是集群模式则更改实例名
  12.                     this.defaultMQPullConsumer.changeInstanceNameToPID();
  13.                 }
  14.                 // 获取MQ客户端实例
  15.                 this.mQClientFactory = MQClientManager.getInstance().getAndCreateMQClientInstance(this.defaultMQPullConsumer, this.rpcHook);
  17.                 // 设置pull模式的均衡策略
  18.                 this.rebalanceImpl.setConsumerGroup(this.defaultMQPullConsumer.getConsumerGroup());
  19.                 this.rebalanceImpl.setMessageModel(this.defaultMQPullConsumer.getMessageModel());
  20.                 this.rebalanceImpl.setAllocateMessageQueueStrategy(this.defaultMQPullConsumer.getAllocateMessageQueueStrategy());
  21.                 this.rebalanceImpl.setmQClientFactory(this.mQClientFactory);
  23.                 // 构造pullAPI的包装器
  24.                 this.pullAPIWrapper = new PullAPIWrapper(
  25.                     mQClientFactory,
  26.                     this.defaultMQPullConsumer.getConsumerGroup(), isUnitMode());
  27.                 this.pullAPIWrapper.registerFilterMessageHook(filterMessageHookList);
  29.                 // pull模式的消息消费偏移存储
  30.                 if (this.defaultMQPullConsumer.getOffsetStore() != null) { 
  31.                     this.offsetStore = this.defaultMQPullConsumer.getOffsetStore();
  32.                 } else { 
  33.                     switch (this.defaultMQPullConsumer.getMessageModel()) { 
  34.                         case BROADCASTING:
  35.                             this.offsetStore = new LocalFileOffsetStore(this.mQClientFactory, this.defaultMQPullConsumer.getConsumerGroup());
  36.                             break;
  37.                         case CLUSTERING:
  38.                             this.offsetStore = new RemoteBrokerOffsetStore(this.mQClientFactory, this.defaultMQPullConsumer.getConsumerGroup());
  39.                             break;
  40.                         default:
  41.                             break;
  42.                     }
  43.                     this.defaultMQPullConsumer.setOffsetStore(this.offsetStore);
  44.                 }
  46.                 // 加载消费偏移情况
  47.                 this.offsetStore.load();
  49.                 // 注册pull消费者
  50.                 boolean registerOK = mQClientFactory.registerConsumer(this.defaultMQPullConsumer.getConsumerGroup(), this);
  51.                 if (!registerOK) { 
  52.                     this.serviceState = ServiceState.CREATE_JUST;
  54.                     throw new MQClientException("The consumer group[" + this.defaultMQPullConsumer.getConsumerGroup()
  55.                         + "] has been created before, specify another name please." + FAQUrl.suggestTodo(FAQUrl.GROUP_NAME_DUPLICATE_URL),
  56.                         null);
  57.                 }
  59.                 // MQClient的启动
  60.                 mQClientFactory.start();
  61.                 log.info("the consumer [{}] start OK", this.defaultMQPullConsumer.getConsumerGroup());
  62.                 this.serviceState = ServiceState.RUNNING;
  63.                 break;
  64.             case RUNNING:
  65.             case START_FAILED:
  66.             case SHUTDOWN_ALREADY:
  67.                 throw new MQClientException("The PullConsumer service state not OK, maybe started once, "
  68.                     + this.serviceState
  69.                     + FAQUrl.suggestTodo(FAQUrl.CLIENT_SERVICE_NOT_OK),
  70.                     null);
  71.             default:
  72.                 break;
  73.         }
  75.     }

走览玩这段源码，可以看见，pull模式和push模式的消费者核心启动流程中，唯一的区别就是push模式下会有注册消息消费的监听器功能和最后检测连接操作，从这便可以看出，其实push模式是在pull模式上做了enhance。

PullConsumer的关闭

pull模式consumer在关闭的时候清理资源的情况如下所示：

  1. public synchronized void shutdown() { 
  2.         switch (this.serviceState) { 
  3.             case CREATE_JUST:
  4.                 break;
  5.             case RUNNING:
  6.                 // 持久化消费偏移
  7.                 this.persistConsumerOffset();
  8.                 // 将消费者移除注册表
  9.                 this.mQClientFactory.unregisterConsumer(this.defaultMQPullConsumer.getConsumerGroup());
  10.                 // 关闭MQClient，清理资源
  11.                 this.mQClientFactory.shutdown();
  12.                 log.info("the consumer [{}] shutdown OK", this.defaultMQPullConsumer.getConsumerGroup());
  13.                 this.serviceState = ServiceState.SHUTDOWN_ALREADY;
  14.                 break;
  15.             case SHUTDOWN_ALREADY:
  16.                 break;
  17.             default:
  18.                 break;
  19.         }
  20.     }

公共行为

包装设置 group

对于group的设置，主要是在给予的group参数基础上加了一层名字空间域namespace，而在包装的过程中首先是要获取namespace的值，源码如下：

  1. public String getNamespace() { 
  2.         if (StringUtils.isNotEmpty(namespace)) { 
  3.             return namespace;
  4.         }
  5.         if (StringUtils.isNotEmpty(this.namesrvAddr)) { 
  6.             if (NameServerAddressUtils.validateInstanceEndpoint(namesrvAddr)) { 
  7.                 return NameServerAddressUtils.parseInstanceIdFromEndpoint(namesrvAddr);
  8.             }
  9.         }
  10.         return namespace;
  11.     }

这段源码来自ClientConfig，可以看见其是先尝试获取client中配置的值，如果值不存在的话便尝试从实例的Endpoint中获取，从端点中获取是通过匹正则来正则来判断的，如果匹配成功，则从InstanceEndpoint中截取namespace得值，正则表达式如下所示：

  1. ^http://MQ_INST_\w+_\w+\..*

只要没有做配置，那便获取不到值，因为不存在默认值。在执行完毕获取namespace值的操作后，便是真正的来包装group的值了。源码如下所示：

  1. public static String wrapNamespace(String namespace, String resourceWithOutNamespace) { 
  2.         // 如果namespace为空或本身的group值为空
  3.         if (StringUtils.isEmpty(namespace) || StringUtils.isEmpty(resourceWithOutNamespace)) { 
  4.             return resourceWithOutNamespace;
  5.         }
  7.         // 如果是系统的group或者已经包装过namespace
  8.         if (isSystemResource(resourceWithOutNamespace) || isAlreadyWithNamespace(resourceWithOutNamespace, namespace)) { 
  9.             return resourceWithOutNamespace;
  10.         }
  12.         String resourceWithoutRetryAndDLQ = withOutRetryAndDLQ(resourceWithOutNamespace);
  13.         StringBuffer strBuffer = new StringBuffer();
  15.         // RetryTopic
  16.         if (isRetryTopic(resourceWithOutNamespace)) { 
  17.             strBuffer.append(MixAll.RETRY_GROUP_TOPIC_PREFIX);
  18.         }
  20.         // DLQTopic
  21.         if (isDLQTopic(resourceWithOutNamespace)) { 
  22.             strBuffer.append(MixAll.DLQ_GROUP_TOPIC_PREFIX);
  23.         }
  25.         return strBuffer.append(namespace).append(NAMESPACE_SEPARATOR).append(resourceWithoutRetryAndDLQ).toString();
  27.     }

消息轨迹功能开启

消息轨迹功能是异步的，其的核心也是一个producer，具体初始化话过程如下：

  1. public void start(String nameSrvAddr, AccessChannel accessChannel) throws MQClientException { 
  2.         if (isStarted.compareAndSet(false, true)) { 
  3.             traceProducer.setNamesrvAddr(nameSrvAddr);
  4.             traceProducer.setInstanceName(TRACE_INSTANCE_NAME + "_" + nameSrvAddr);
  5.             traceProducer.start();
  6.         }
  7.         this.accessChannel = accessChannel;
  8.         this.worker = new Thread(new AsyncRunnable(), "MQ-AsyncTraceDispatcher-Thread-" + dispatcherId);
  9.         this.worker.setDaemon(true);
  10.         this.worker.start();
  11.         this.registerShutDownHook();
  12.     }

MQClient启动

这里直接来看MQClient的启动，如果是第一次启动则走CREATE_JUST分支，核心源码如下：

  1. public void start() throws MQClientException { 
  3.         synchronized (this) { 
  4.             switch (this.serviceState) { 
  5.                 case CREATE_JUST:
  6.                     this.serviceState = ServiceState.START_FAILED;
  7.                     // If not specified,looking address from name server
  8.                     if (null == this.clientConfig.getNamesrvAddr()) { 
  9.                         this.mQClientAPIImpl.fetchNameServerAddr();
  10.                     }
  11.                     // Start request-response channel
  12.                     this.mQClientAPIImpl.start();
  13.                     // Start various schedule tasks
  14.                     this.startScheduledTask();
  15.                     // Start pull service
  16.                     this.pullMessageService.start();
  17.                     // Start rebalance service
  18.                     this.rebalanceService.start();
  19.                     // Start push service
  20.                     this.defaultMQProducer.getDefaultMQProducerImpl().start(false);
  21.                     log.info("the client factory [{}] start OK", this.clientId);
  22.                     this.serviceState = ServiceState.RUNNING;
  23.                     break;
  24.                 case RUNNING:
  25.                     break;
  26.                 case SHUTDOWN_ALREADY:
  27.                     break;
  28.                 case START_FAILED:
  29.                     throw new MQClientException("The Factory object[" + this.getClientId() + "] has been created before, and failed.", null);
  30.                 default:
  31.                     break;
  32.             }
  33.         }
  34.     }

可以看见，其主要是做了如下6件事：

  • 再次检查namesrv地址
  • 打开client的交互信道
  • 启动pull-message服务
  • 启动push-message服务
  • 启动均衡负载服务
  • 启动一系列定时任务

其中最核心的便是启动的这些定时任务，具体定时任务如下：

  • 定时刷新namesrv地址；本地的topic路由；
  • 定时处理broker（触发心跳，清理下线的）；
  • 定时持久化消费者消费位置；

  • 定时处理消费者的线程池；

    private void startScheduledTask() { 

    1.     if (null == this.clientConfig.getNamesrvAddr()) { 
    2.         this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() { 
    4.             @Override
    5.             public void run() { 
    6.                 try { 
    7.                     MQClientInstance.this.mQClientAPIImpl.fetchNameServerAddr();
    8.                 } catch (Exception e) { 
    9.                     log.error("ScheduledTask fetchNameServerAddr exception", e);
    10.                 }
    11.             }
    12.         }, 1000 * 10, 1000 * 60 * 2, TimeUnit.MILLISECONDS);
    13.     }
    15.     this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() { 
    17.         @Override
    18.         public void run() { 
    19.             try { 
    20.                 MQClientInstance.this.updateTopicRouteInfoFromNameServer();
    21.             } catch (Exception e) { 
    22.                 log.error("ScheduledTask updateTopicRouteInfoFromNameServer exception", e);
    23.             }
    24.         }
    25.     }, 10, this.clientConfig.getPollNameServerInterval(), TimeUnit.MILLISECONDS);
    27.     this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() { 
    29.         @Override
    30.         public void run() { 
    31.             try { 
    32.                 MQClientInstance.this.cleanOfflineBroker();
    33.                 MQClientInstance.this.sendHeartbeatToAllBrokerWithLock();
    34.             } catch (Exception e) { 
    35.                 log.error("ScheduledTask sendHeartbeatToAllBroker exception", e);
    36.             }
    37.         }
    38.     }, 1000, this.clientConfig.getHeartbeatBrokerInterval(), TimeUnit.MILLISECONDS);
    40.     this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() { 
    42.         @Override
    43.         public void run() { 
    44.             try { 
    45.                 MQClientInstance.this.persistAllConsumerOffset();
    46.             } catch (Exception e) { 
    47.                 log.error("ScheduledTask persistAllConsumerOffset exception", e);
    48.             }
    49.         }
    50.     }, 1000 * 10, this.clientConfig.getPersistConsumerOffsetInterval(), TimeUnit.MILLISECONDS);
    52.     this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() { 
    54.         @Override
    55.         public void run() { 
    56.             try { 
    57.                 MQClientInstance.this.adjustThreadPool();
    58.             } catch (Exception e) { 
    59.                 log.error("ScheduledTask adjustThreadPool exception", e);
    60.             }
    61.         }
    62.     }, 1, 1, TimeUnit.MINUTES);
    63. }

broker心跳触发

在触发心跳的操作中，具体分为两个部分，如下所示：

  1. // 触发心跳
  2. this.sendHeartbeatToAllBroker();
  3. // 上传加载消息过滤类
  4. this.uploadFilterClassSource();

上传消息过滤类资源是为了push模式下的消费者进行消息过滤用的，这里不做探究，直接来看启动触发心跳的核心代码，如下所示：

  1. private void sendHeartbeatToAllBroker() { 
  2.         final HeartbeatData heartbeatData = this.prepareHeartbeatData();
  3.         final boolean producerEmpty = heartbeatData.getProducerDataSet().isEmpty();
  4.         final boolean consumerEmpty = heartbeatData.getConsumerDataSet().isEmpty();
  5.         if (producerEmpty && consumerEmpty) { 
  6.             log.warn("sending heartbeat, but no consumer and no producer");
  7.             return;
  8.         }
  10.         if (!this.brokerAddrTable.isEmpty()) { 
  11.             long times = this.sendHeartbeatTimesTotal.getAndIncrement();
  12.             Iterator<Entry<String, HashMap<Long, String>>> it = this.brokerAddrTable.entrySet().iterator();
  13.             while (it.hasNext()) { 
  14.                 Entry<String, HashMap<Long, String>> entry = it.next();
  15.                 String brokerName = entry.getKey();
  16.                 HashMap<Long, String> oneTable = entry.getValue();
  17.                 if (oneTable != null) { 
  18.                     for (Map.Entry<Long, String> entry1 : oneTable.entrySet()) { 
  19.                         Long id = entry1.getKey();
  20.                         String addr = entry1.getValue();
  21.                         if (addr != null) { 
  22.                             if (consumerEmpty) { 
  23.                                 if (id != MixAll.MASTER_ID)
  24.                                     continue;
  25.                             }
  27.                             try { 
  28.                                 // 在触发心跳的时候获取broker的版本号
  29.                                 int version = this.mQClientAPIImpl.sendHearbeat(addr, heartbeatData, 3000);
  30.                                 if (!this.brokerVersionTable.containsKey(brokerName)) { 
  31.                                     this.brokerVersionTable.put(brokerName, new HashMap<String, Integer>(4));
  32.                                 }
  33.                                 this.brokerVersionTable.get(brokerName).put(addr, version);
  34.                                 if (times % 20 == 0) { 
  35.                                     log.info("send heart beat to broker[{} {} {}] success", brokerName, id, addr);
  36.                                     log.info(heartbeatData.toString());
  37.                                 }
  38.                             } catch (Exception e) { 
  39.                                 if (this.isBrokerInNameServer(addr)) { 
  40.                                     log.info("send heart beat to broker[{} {} {}] failed", brokerName, id, addr, e);
  41.                                 } else { 
  42.                                     log.info("send heart beat to broker[{} {} {}] exception, because the broker not up, forget it", brokerName,
  43.                                         id, addr, e);
  44.                                 }
  45.                             }
  46.                         }
  47.                     }
  48.                 }
  49.             }
  50.         }
  51.     }

发表评论

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

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

相关阅读