高可用究竟指的是什么？请参考：关于高可用的系统
RocketMQ做了以下的事情来保证系统的高可用
- 多master部署，防止单点故障
- 消息冗余（主从结构），防止消息丢失
- 故障恢复（本篇暂不讨论）
那么问题来了：
- 怎么支持多broker的写？
- 怎么实现消息冗余？
下面分开说明这两个问题
多master集群
这里强调出master集群，是因为需要多个broker set，而一个broker set只有一个master（见下文的“注意”），所以是master集群
broker有三种角色：ASYNC_MASTER、SYNC_MASTER和SLAVE，这些角色常用的搭配为：
- ASYNC_MASTER、SLAVE：容许丢消息，但是要broker一直可用，master异步传输CommitLog到slave
- SYNC_MASTER、SLAVE：不允许丢消息，master同步传输CommitLog到slave
- ASYNC_MASTER：如果只是想简单部署则使用这种方式
master：负责消息的读写
slave：只负责读消息
SYNC_MASTER与ASYNC_MASTER的区别是sync会等待消息传输到slave才算消息写完成，而async不会同步等待，而是异步复制到slave
RocketMQ的架构图（原图地址）
注意：在RocketMQ里面有一个概念broker set，一个broker set由一个master和多个slave组成，一个broker set内的每个broker的brokerName相同。
在broker集群中每个master相互之间是独立，master之间不会有交互，每个master维护自己的CommitLog、自己的ConsumeQueue，但是每一个master都有可能收到同一个topic下的producer发来的消息
为了支持多master集群，需要解决几个问题：
- namesrv怎么管理broker
- producer发送消息的时候知道发送到哪一个broker（为什么是master）
1. namesrv怎么管理broker
broker启动的时候会向namesrv注册自己的信息
1// org.apache.rocketmq.broker.BrokerController#registerBrokerAll 2public synchronized void registerBrokerAll(final boolean checkOrderConfig, boolean oneway) { 3 TopicConfigSerializeWrapper topicConfigWrapper = this.getTopicConfigManager().buildTopicConfigSerializeWrapper(); 4 5 // 省略中间代码... 6 RegisterBrokerResult registerBrokerResult = this.brokerOuterAPI.registerBrokerAll( 7 this.brokerConfig.getBrokerClusterName(), 8 this.getBrokerAddr(), 9 this.brokerConfig.getBrokerName(), 10 this.brokerConfig.getBrokerId(), 11 this.getHAServerAddr(), 12 topicConfigWrapper, 13 this.filterServerManager.buildNewFilterServerList(), 14 oneway, 15 this.brokerConfig.getRegisterBrokerTimeoutMills()); 16 // 省略中间代码... 17}
信息中包括：
clusterName：broker 集群的名字，如：DefaultCluster
brokerAddr：broker的ip:port，如：192.168.0.102:10911
brokerName：注意这个字段，上面介绍过了，一个broker set中的brokerName是相同的，需要在部署的时候配置
brokerId：用来唯一标示一个broker set中的broker，master是0（org.apache.rocketmq.common.MixAll#MASTER_ID），slave是正整数
haServerAddr：haServer的ip:port，如：192.168.0.102:10912
topicConfigWrapper：是比较复杂的数据结构，主要包含了broker上所有的topic信息，如：
1{ 2 "dataVersion": { 3 "counter": 2, 4 "timestamp": 1514252649572 5 }, 6 "topicConfigTable": { 7 "TopicTest": { 8 "order": false, 9 "perm": 6, 10 "readQueueNums": 4, 11 "topicFilterType": "SINGLE_TAG", 12 "topicName": "TopicTest", 13 "topicSysFlag": 0, 14 "writeQueueNums": 4 15 }, 16 "%RETRY%please_rename_unique_group_name_4": { 17 "order": false, 18 "perm": 6, 19 "readQueueNums": 1, 20 "topicFilterType": "SINGLE_TAG", 21 "topicName": "%RETRY%please_rename_unique_group_name_4", 22 "topicSysFlag": 0, 23 "writeQueueNums": 1 24 } 25 } 26}
上面包含了两个topic：TopicTest和%RETRY%please_rename_unique_group_name_4，相关字段的含义：
order：是否是顺序消息
perm：表明该topic的权限，可读（4）、可写（2）、可继承（1），通过位运算组合
readQueueNums：决定了consume消费的MessageQueue共有几个
writeQueueNums：决定了producer发送消息的MessageQueue共有几个
这些信息发送给namesrv之后，namesrv转化为自己的数据结构，namesrv处理broker注册的方法是：
1// org.apache.rocketmq.namesrv.routeinfo.RouteInfoManager#registerBroker 2public RegisterBrokerResult registerBroker( 3 final String clusterName, 4 final String brokerAddr, 5 final String brokerName, 6 final long brokerId, 7 final String haServerAddr, 8 final TopicConfigSerializeWrapper topicConfigWrapper, 9 final List<String> filterServerList, 10 final Channel channel) { 11 RegisterBrokerResult result = new RegisterBrokerResult(); 12 try { 13 try { 14 // 省略中间代码... 15 // 这里会判断只有master才会创建QueueData，因为只有master才包含了读写队列的信息 16 // slave没有自己独立的读写队列信息（salve不会创建自己的queue信息），只是和master的的读写队列信息一致 17 if (null != topicConfigWrapper 18 && MixAll.MASTER_ID == brokerId) { 19 if (this.isBrokerTopicConfigChanged(brokerAddr, topicConfigWrapper.getDataVersion()) 20 || registerFirst) { 21 ConcurrentMap<String, TopicConfig> tcTable = 22 topicConfigWrapper.getTopicConfigTable(); 23 if (tcTable != null) { 24 for (Map.Entry<String, TopicConfig> entry : tcTable.entrySet()) { 25 // 这个方法创建了QueueData，QueueData包含broker set下的读写队列的信息 26 this.createAndUpdateQueueData(brokerName, entry.getValue()); 27 } 28 } 29 } 30 } 31 32 // 省略中间代码... 33 } catch (Exception e) { 34 log.error("registerBroker Exception", e); 35 } 36 37 return result; 38}
上面涉及到的namesrv的几个重要数据结构
1// 每个cluster下的broker set信息，一个brokerName对应的broker set 2private final HashMap<String/* clusterName */, Set<String/* brokerName */>> clusterAddrTable; 3// 每个broker set中的broker信息(set中有哪些broker，每个broker的brokerId和brokerAddr) 4private final HashMap<String/* brokerName */, BrokerData> brokerAddrTable; 5// 每个broker的存活情况 6private final HashMap<String/* brokerAddr */, BrokerLiveInfo> brokerLiveTable; 7// 每个topic下的queue信息，包括每个broker set中读写队列的个数，consumer消费消息和producer发送消息的路由信息都从这个数据结构中获取 8private final HashMap<String/* topic */, List<QueueData>> topicQueueTable;
所以，namesrv通过将broker注册来的信息构造成自己的数据结构：
- 每个cluster有哪些broker set
- 每个broker set包括哪些broker，brokerId和broker的ip:port
- 每个broker的存活情况，根据每次broker上报来的信息，清除可能下线的broker
- 每个topic的消息队列信息，几个读队列，几个写队列
namesrv汇总所有的broker的这些信息，然后供consumer和producer拉取
2. producer发送消息的时候知道发送到哪一个master
之前我们知道producer发送消息的时候发往哪一个broker是由MessageQueue决定的，所以我们先要搞清楚producer发送消息时候的MessageQueue怎么来的。producer维护了一个topicPublishInfoTable，里面包含了每个topic对应的MessageQueue，所以问题就变成了topicPublishInfoTable怎么构造的。
producer发送消息之前都会获取topic对应的队列信息，当topicPublishInfoTable中没有的时候会从namesrv获取，获取的方法如下：
1// org.apache.rocketmq.client.impl.factory.MQClientInstance#updateTopicRouteInfoFromNameServer(java.lang.String, boolean, org.apache.rocketmq.client.producer.DefaultMQProducer) 2public boolean updateTopicRouteInfoFromNameServer(final String topic, boolean isDefault, 3 DefaultMQProducer defaultMQProducer) { 4 try { 5 if (this.lockNamesrv.tryLock(LOCK_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS)) { 6 try { 7 TopicRouteData topicRouteData; 8 if (isDefault && defaultMQProducer != null) { 9 // 省略中间代码... 10 } else { 11 // 从manesrv获取topic的路由信息，namesrv从topicQueueTable获取到该topic对应的所有的QueueData 12 // 然后将每个brokerName下的BrokerData返回 13 topicRouteData = this.mQClientAPIImpl.getTopicRouteInfoFromNameServer(topic, 1000 * 3); 14 } 15 // 省略中间代码... 16 for (BrokerData bd : topicRouteData.getBrokerDatas()) { 17 // 每个broker set下所有的broker地址(ip:port) 18 this.brokerAddrTable.put(bd.getBrokerName(), bd.getBrokerAddrs()); 19 } 20 // Update Pub info 21 { 22 // 将从namesrv获取到的路由信息转换为TopicPublishInfo 23 // 期间会将没有master的broker set的queue信息去除 24 TopicPublishInfo publishInfo = topicRouteData2TopicPublishInfo(topic, topicRouteData); 25 // 省略中间代码... 26 } catch (InterruptedException e) { 27 log.warn("updateTopicRouteInfoFromNameServer Exception", e); 28 } 29 30 return false; 31}
到此，producer也知道自己可以向哪些MessageQueue发送消息了，接下来就是producer的负载均衡算法选出其中一个MessageQueue发送消息（org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl#selectOneMessageQueue，这个暂时不详表），MessageQueue包含的信息有topic、brokerName、queueId，但是producer发送的时候得知道broker的ip:port信息，而且一个brokerName对应的是一个broker set，并不能确定具体的broker，所以接下来应该找到具体的broker
1// org.apache.rocketmq.client.impl.factory.MQClientInstance#findBrokerAddressInPublish 2public String findBrokerAddressInPublish(final String brokerName) { 3 // 上面updateTopicRouteInfoFromNameServer方法将broker set下的broker地址信息保存到brokerAddrTable 4 // 再次重申：一个broker set下的broker的brokerName相同 5 HashMap<Long/* brokerId */, String/* address */> map = this.brokerAddrTable.get(brokerName); 6 if (map != null && !map.isEmpty()) { 7 // 没有花样，就是直接返回brokerId时MixAll.MASTER_ID的broker的ip:port信息 8 // 前面说过master的brokerId就是MixAll.MASTER_ID，所以获取到的broker是broker set中的master 9 return map.get(MixAll.MASTER_ID); 10 } 11 12 return null; 13}
终于真相大白，producer只会向是master的broker发送消息，也就是一个broker set中brokerId是0的broker。
producer只能发送消息到master，而不能发送到slave，这也说明了master负责读“写”，而slave只负责读（当然，这里只说明了“写”的部分，关于master 和slave的“读”下一篇介绍）。
总结
本篇介绍了RocketMQ究竟做了什么来实现作为一个消息队列中间件的高可用，由于篇幅会偏长，所以分为两篇文章来说明，下一篇说明文中遗留下的另一个问题——RocketMQ源码 — 六、 RocketMQ高可用（2）
参考：