源码版本号:版本号:4.9.4

部署架构

NameServer

它负责存储RocketMQ集群的元数据信息，例如某个Topic有多少个队列(MessageQueue)，又分别在哪些Broker上

定时任务每隔10秒去尝试清理一次已经120秒没有发送过心跳的Broker

Broker

存储数据的地方

定时任务每隔30秒向NameServer发送心跳，携带topic信息和更新心跳时间

Producer

生产者，每隔30秒向所有的Broker发送心跳信息

Consumer

消费者，每隔30秒向所有的Broker发送心跳信息，主要包括当前消费者订阅了哪些topic信息

每个消费者就可以从Broker拿到消费某个Topic的所有实例，每个消费者就可以通过具体的算法计算出自己应该消费哪些队列

某个Topic下的某个队列，在同一个消费者组内，只允许一个消费者对其进行消费，可以保证同一个队列的消息能够被顺序消费

MQClientInstance

每个生产者和消费者都有一个 MQClientInstance 实例

生产者和消费者启动时通过调用 MQClientManager.getInstance().getOrCreateMQClientInstance获取

重要字段分析

public class MQClientInstance {
 /** * 生产者启动时注册进来, key 为 groupName * 如果生产者groupName相同, 则会报错, 生产者实例里面会保存一份发送过消息的topic的队列信息
 */ private final ConcurrentMap<String, MQProducerInner> producerTable = new ConcurrentHashMap<String, MQProducerInner>(); /** * 消费者启动时注册进来, key 为 groupName * 保存消费者信息, 消费者实例里面能知道需要订阅哪些topic信息
 */ private final ConcurrentMap<String, MQConsumerInner> consumerTable = new ConcurrentHashMap<String, MQConsumerInner>(); /** * 保存所有的broker信息, 给Broker发送心跳就是通过这个拿到Broker地址
 * 有定时任务定时去查询Broker地址是否在topicRouteTable中, 如果不在则需要去除
 * 生产者发送消息时, 通过brokerName和brokerId=0来查找对应的brokerAddress
 */ private final ConcurrentMap<String/* Broker Name */, HashMap<Long/* brokerId */, String/* address */>> brokerAddrTable = new ConcurrentHashMap<String, HashMap<Long, String>>(); /** * 所有的topic元数据信息, TopicRouteData就是从NameServer拉回来的数据
 * TopicRouteData里面会有Topic的队列信息[brokerName、readQueueNums、writeQueueNums]
 * 和Broker信息[cluster、brokerName、brokerAddrs(key=brokerId, value=broker address)]
 */ private final ConcurrentMap<String/* Topic */, TopicRouteData> topicRouteTable = new ConcurrentHashMap<String, TopicRouteData>();}

启动

生产者和消费者启动的时候都会调用 MQClientInstance 的启动方法

代码入口: MQClientInstance#start

public class MQClientInstance {
 // 实例状态
 private ServiceState serviceState = ServiceState.CREATE_JUST; /** * 找到225行
 */ public void start() throws MQClientException { synchronized (this) { // 启动前 serviceState = ServiceState.CREATE_JUST switch (this.serviceState) { case CREATE_JUST: this.serviceState = ServiceState.START_FAILED; // If not specified,looking address from name server if (null == this.clientConfig.getNamesrvAddr()) { this.mQClientAPIImpl.fetchNameServerAddr(); } // 里面会启动一个Netty服务, 用来进行通信服务
 this.mQClientAPIImpl.start(); /** * 开启各种定时任务
 */ this.startScheduledTask(); /** * 开启消费者的消息拉取任务
 * 拉取消息的任务被放在一个队列里面, 消息队列负载的时候往里面放任务
 * 查看PullMessageService#run方法, 从队列里面拿拉取消息的任务
 * 拿到任务后会交给具体的消费者去broker拉取消息
 * 最终负责拉取消息的代码在DefaultMQPushConsumerImpl#pullMessage方法中
 */ this.pullMessageService.start(); /** * 开启消息队列负载
 * 查看RebalanceService#run方法
 * 可以发现默认每隔20s就会执行MQClientInstance#doRebalance方法
 * 然后调用每个消费者的DefaultMQPushConsumerImpl#doRebalance方法
 */ this.rebalanceService.start(); // Start push service this.defaultMQProducer.getDefaultMQProducerImpl().start(false); log.info("the client factory [{}] start OK", this.clientId); this.serviceState = ServiceState.RUNNING; break; case START_FAILED: throw new MQClientException("The Factory object[" + this.getClientId() + "] has been created before, and failed.", null); default: break; } } }}

定时任务分析

MQClientInstance会启动多个定时任务

具体代码在这个方法里
MQClientInstance.startScheduledTask

public class MQClientInstance {
 /** * 找到256行
 */ private void startScheduledTask() { // 省略部分代码, 只贴上主要代码
 /** * 每隔30s执行一次
 * 从NameServer中获取最新的topic信息, 将最新的topic信息更新到每个生产者和消费者中
 * 1.遍历所有的生产者和消费者, 拿到所有的topic
 * 2.一个一个地去NameServer查询, 然后更新生产者和消费者里面的topic信息(topic对应的队列列表)
 * 消费者: 能够知道是否存在新增或者减少消费者实例, 消费者就能重新分配拉取消息的队列
 * 生产者: 如果存在Broker不可用, 这个时候生产者就不会往不可用Broker的队列里发送消息
 */ MQClientInstance.this.updateTopicRouteInfoFromNameServer(); /** * 每隔30s执行一次
 * 1.清理掉下线的Broker
 * 2.给每个Broker发送心跳信息
 * 这样Broker就能知道有哪些生产者、哪些消费者消费哪些topic
 */ MQClientInstance.this.cleanOfflineBroker(); MQClientInstance.this.sendHeartbeatToAllBrokerWithLock(); /** * 每个10s执行一次
 * 更新消费者的消费进度
 */ MQClientInstance.this.persistAllConsumerOffset(); }}

本作品采用《CC 协议》，转载必须注明作者和本文链接