RocketMQ如何维持心跳

首先放上RocketMQ网络结构图，如下所示：

Producer与NameSrv随机建立长连接，定期从NameSrv获取topic路由信息。然后Producer还与Broker的Master结点建立长连接，用于发送消息。此外Producer还与Master维持了一个心跳。
Conumser与NamseSrv随机建立长连接，定期从NameSrv获取topic路由信息。然后Consumer还与Broker的Master和Slave结点建立长连接，用于订阅消息。此外Consumer还与Master和lslave维持了一个心跳。
以上就是RocketMQ所有的心跳机制。

客户端发送心跳

在Producer和Consumer启动时，会通过 this.mQClientFactory.sendHeartbeatToAllBrokerWithLock();发起心跳请求。此外在MQClientInstance启动时，会启动一个定时任务 this.startScheduledTask();。里面包含了各种各样的定时任务，其中就包括定期发送心跳信息到Broker。

 this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {@Overridepublic void run() {try {MQClientInstance.this.cleanOfflineBroker();MQClientInstance.this.sendHeartbeatToAllBrokerWithLock(); // 发送心跳，默认时间间隔30秒} catch (Exception e) {log.error("ScheduledTask sendHeartbeatToAllBroker exception", e);}}}, 1000, this.clientConfig.getHeartbeatBrokerInterval(), TimeUnit.MILLISECONDS);

sendHeartbeanToAllBrokerWithLock在sendHeartbeatToAllBroker上加了锁，避免心跳混乱，没有什么特别之处。我们来看看sendHeartbeatToAllBroker做了什么：

 private void sendHeartbeatToAllBroker() {// 首先准备好心跳信息，主要是Producer和comsumer相关信息，内容后面会具体分析final HeartbeatData heartbeatData = this.prepareHeartbeatData();final boolean producerEmpty = heartbeatData.getProducerDataSet().isEmpty();final boolean consumerEmpty = heartbeatData.getConsumerDataSet().isEmpty();if (producerEmpty && consumerEmpty) {log.warn("sending heartbeat, but no consumer and no producer");return;}if (!this.brokerAddrTable.isEmpty()) {long times = this.sendHeartbeatTimesTotal.getAndIncrement();Iterator<Entry<String, HashMap<Long, String>>> it = this.brokerAddrTable.entrySet().iterator();while (it.hasNext()) {Entry<String, HashMap<Long, String>> entry = it.next();String brokerName = entry.getKey();HashMap<Long, String> oneTable = entry.getValue();if (oneTable != null) {for (Map.Entry<Long, String> entry1 : oneTable.entrySet()) {Long id = entry1.getKey();String addr = entry1.getValue();if (addr != null) {if (consumerEmpty) { // 如果没有conumser则剔除掉slave结点，因为producer只需要与master维持心跳即可if (id != MixAll.MASTER_ID)continue;}try {int version = this.mQClientAPIImpl.sendHearbeat(addr, heartbeatData, 3000);if (!this.brokerVersionTable.containsKey(brokerName)) {this.brokerVersionTable.put(brokerName, new HashMap<String, Integer>(4));}this.brokerVersionTable.get(brokerName).put(addr, version);if (times % 20 == 0) {// 减少日志频率log.info("send heart beat to broker[{} {} {}] success", brokerName, id, addr);log.info(heartbeatData.toString());}} catch (Exception e) {if (this.isBrokerInNameServer(addr)) {log.info("send heart beat to broker[{} {} {}] failed", brokerName, id, addr);} else {log.info("send heart beat to broker[{} {} {}] exception, because the broker not up, forget it", brokerName,id, addr);}}}}}}}}

这里主要做了两个工作：
1-预备好心跳信息
2-发送心跳
其中根据客户端的类型不同，发送的对象会又差别。如果是Producer启动，那么MQClientInstance里面的conumser是空的，那么会剔除掉Broker的slave结点，只向master发送心跳。如果是是Consumer启动，那么MQClientInstance里面的consumer不为空，就会向所有的broker结点发送心跳。
sendHearbeat()非常简单，包装RemotingCommand对象，然后就是RemotingClient的调用了，涉及到Netty通讯了。这个之前已经讨论过，具体可以参考：RocketMQ是如何通讯的？
发送心跳返回的是broker端MQ的版本号，拿到后会更新本地保存的broker版本控制信息。

心跳内容

心跳内容比较简单，包括客户端id，生产者信息和消费者信息，一般情况下生产者信息和消费者信息是互斥的，producerDataSet和consumerDataSet有一个为空。但也不排除有的应用既是生产者，也是消费者，这种情况下producerDataSet和consumerDataSet都不为空。

public class HeartbeatData extends RemotingSerializable {private String clientID;private Set<ProducerData> producerDataSet = new HashSet<ProducerData>();private Set<ConsumerData> consumerDataSet = new HashSet<ConsumerData>();}

接下来，我们一个个分析，先看看ProducerData是什么：

public class ProducerData {private String groupName;
}

这非常简单了。。。就一个groupName，不需要过多解释了。下面再看ConsumerData：

public class ConsumerData {private String groupName; // 分组private ConsumeType consumeType; // 消费类型，有推模式和拉模式两种private MessageModel messageModel;// 消息类型，广播和集群消费两种private ConsumeFromWhere consumeFromWhere; // 从何处开始消费，从一开始偏移量，从最后偏移量，或者按时间戳消费。private Set<SubscriptionData> subscriptionDataSet = new HashSet<SubscriptionData>();// 订阅信息private boolean unitMode; // 单元模式，默认是false。这个与topic有关，但是没看懂拿来干嘛的？
}

ConsumerData主要包括的消费者的一些配置信息，如果写过消费者代码，对这些还是很熟悉的。其中SubscriptData是订阅信息，结构如下：

public class SubscriptionData implements Comparable<SubscriptionData> {public final static String SUB_ALL = "*"; // 常量，默认订阅所有tag类型消息private boolean classFilterMode = false;private String topic;// 主题private String subString;// 订阅表达式，例如"taga || tagb"private Set<String> tagsSet = new HashSet<String>();// tag列表private Set<Integer> codeSet = new HashSet<Integer>();// tag的hashcode列表private long subVersion = System.currentTimeMillis();private String expressionType; //订阅表达式类型，有tag模式和sql模式@JSONField(serialize = false)private String filterClassSource;
}

订阅信息里面比较常用的就是topic和subString，我们消费订阅信息主要就是这俩。例如：

consumer.subscribe("topic", "TagA || TagC || TagD");

其中的"TagA || TagC || TagD"就是这里的subString，创建订阅信息的时候，subString会被分割成TagA、TagB、TagD，然后保存至tagsSet里面。他们的hashcode会保存到codeSet里面。
以上就是心跳的所有内容。

Broker处理心跳

Broker处理心跳是在ClientManageProcessor中处理的，对于ProducerData的内容处理很简单，直接注册producer，把producer的ClientChannelInfo保存下来，后面与producer通讯的时候会用到。对于Consumer的处理就稍微复杂一点，除了注册consumer之外，如果消费分组配置不为空的话，还会创建一个用于重试的topic，这个在消息重新消费时有用。这部分在后面介绍consumer消费消息时会再次提到。

  public RemotingCommand heartBeat(ChannelHandlerContext ctx, RemotingCommand request) {RemotingCommand response = RemotingCommand.createResponseCommand(null);HeartbeatData heartbeatData = HeartbeatData.decode(request.getBody(), HeartbeatData.class);ClientChannelInfo clientChannelInfo = new ClientChannelInfo(ctx.channel(),heartbeatData.getClientID(),request.getLanguage(),request.getVersion());for (ConsumerData data : heartbeatData.getConsumerDataSet()) {SubscriptionGroupConfig subscriptionGroupConfig =this.brokerController.getSubscriptionGroupManager().findSubscriptionGroupConfig(data.getGroupName());boolean isNotifyConsumerIdsChangedEnable = true;if (null != subscriptionGroupConfig) {isNotifyConsumerIdsChangedEnable = subscriptionGroupConfig.isNotifyConsumerIdsChangedEnable();int topicSysFlag = 0;if (data.isUnitMode()) {topicSysFlag = TopicSysFlag.buildSysFlag(false, true);}String newTopic = MixAll.getRetryTopic(data.getGroupName());this.brokerController.getTopicConfigManager().createTopicInSendMessageBackMethod(newTopic,subscriptionGroupConfig.getRetryQueueNums(),PermName.PERM_WRITE | PermName.PERM_READ, topicSysFlag);}boolean changed = this.brokerController.getConsumerManager().registerConsumer(data.getGroupName(),clientChannelInfo,data.getConsumeType(),data.getMessageModel(),data.getConsumeFromWhere(),data.getSubscriptionDataSet(),isNotifyConsumerIdsChangedEnable);if (changed) {log.info("registerConsumer info changed {} {}",data.toString(),RemotingHelper.parseChannelRemoteAddr(ctx.channel()));}}for (ProducerData data : heartbeatData.getProducerDataSet()) {this.brokerController.getProducerManager().registerProducer(data.getGroupName(),clientChannelInfo);}response.setCode(ResponseCode.SUCCESS);response.setRemark(null);return response;}

为什么Producer不与NameSrv维持心跳呢

这个问题的同类问题是，为什么Consumer不与NameSrv维持心跳？或者说，为什么Broker不与NameSrv维持心跳？其实Producer、Consumer、Broker都与NameSrv有“维持心跳”的动作，就是Producer、Consumer定期从NameSrv拉取Topic路由信息，Broker定期向NameSrv注册包装了Topic路由的broker信息，只是它们没有明显的使用HeartbeatData相关的写法。HeartbeatData相关的内容都在common工程下的protocol.heartbeat包下：

有个可能的原因是，客户端和broker的心跳维持信息比较复杂，不像客户端与NameSrv、Broker与NameSrv那样需求几乎稳定不变，所以作者单独写了心跳模块。至于真实的原因是什么，我目前也不知道。