RocketMQ源码 — 六、 RocketMQ高可用(1)

高可用究竟指的是什么?请参考:关于高可用的系统

RocketMQ做了以下的事情来保证系统的高可用

  • 多master部署,防止单点故障
  • 消息冗余(主从结构),防止消息丢失
  • 故障恢复(本篇暂不讨论)

那么问题来了:

  • 怎么支持多broker的写?
  • 怎么实现消息冗余?

下面分开说明这两个问题

多master集群

这里强调出master集群,是因为需要多个broker set,而一个broker set只有一个master(见下文的“注意”),所以是master集群

broker有三种角色:ASYNC_MASTER、SYNC_MASTER和SLAVE,这些角色常用的搭配为:

  1. ASYNC_MASTER、SLAVE:容许丢消息,但是要broker一直可用,master异步传输CommitLog到slave
  2. SYNC_MASTER、SLAVE:不允许丢消息,master同步传输CommitLog到slave
  3. ASYNC_MASTER:如果只是想简单部署则使用这种方式

master:负责消息的读写

slave:只负责读消息

SYNC_MASTER与ASYNC_MASTER的区别是sync会等待消息传输到slave才算消息写完成,而async不会同步等待,而是异步复制到slave

RocketMQ的架构图(原图地址

RocketMQ源码 — 六、 RocketMQ高可用(1)

注意:在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注册自己的信息

// org.apache.rocketmq.broker.BrokerController#registerBrokerAll
public synchronized void registerBrokerAll(final boolean checkOrderConfig, boolean oneway) {
TopicConfigSerializeWrapper topicConfigWrapper = this.getTopicConfigManager().buildTopicConfigSerializeWrapper(); // 省略中间代码...
RegisterBrokerResult registerBrokerResult = this.brokerOuterAPI.registerBrokerAll(
this.brokerConfig.getBrokerClusterName(),
this.getBrokerAddr(),
this.brokerConfig.getBrokerName(),
this.brokerConfig.getBrokerId(),
this.getHAServerAddr(),
topicConfigWrapper,
this.filterServerManager.buildNewFilterServerList(),
oneway,
this.brokerConfig.getRegisterBrokerTimeoutMills());
// 省略中间代码...
}

信息中包括:

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信息,如:

{
"dataVersion": {
"counter": 2,
"timestamp": 1514252649572
},
"topicConfigTable": {
"TopicTest": {
"order": false,
"perm": 6,
"readQueueNums": 4,
"topicFilterType": "SINGLE_TAG",
"topicName": "TopicTest",
"topicSysFlag": 0,
"writeQueueNums": 4
},
"%RETRY%please_rename_unique_group_name_4": {
"order": false,
"perm": 6,
"readQueueNums": 1,
"topicFilterType": "SINGLE_TAG",
"topicName": "%RETRY%please_rename_unique_group_name_4",
"topicSysFlag": 0,
"writeQueueNums": 1
}
}
}

上面包含了两个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注册的方法是:

// org.apache.rocketmq.namesrv.routeinfo.RouteInfoManager#registerBroker
public RegisterBrokerResult registerBroker(
final String clusterName,
final String brokerAddr,
final String brokerName,
final long brokerId,
final String haServerAddr,
final TopicConfigSerializeWrapper topicConfigWrapper,
final List<String> filterServerList,
final Channel channel) {
RegisterBrokerResult result = new RegisterBrokerResult();
try {
try {
// 省略中间代码...
// 这里会判断只有master才会创建QueueData,因为只有master才包含了读写队列的信息
// slave没有自己独立的读写队列信息(salve不会创建自己的queue信息),只是和master的的读写队列信息一致
if (null != topicConfigWrapper
&& MixAll.MASTER_ID == brokerId) {
if (this.isBrokerTopicConfigChanged(brokerAddr, topicConfigWrapper.getDataVersion())
|| registerFirst) {
ConcurrentMap<String, TopicConfig> tcTable =
topicConfigWrapper.getTopicConfigTable();
if (tcTable != null) {
for (Map.Entry<String, TopicConfig> entry : tcTable.entrySet()) {
// 这个方法创建了QueueData,QueueData包含broker set下的读写队列的信息
this.createAndUpdateQueueData(brokerName, entry.getValue());
}
}
}
} // 省略中间代码...
} catch (Exception e) {
log.error("registerBroker Exception", e);
} return result;
}

上面涉及到的namesrv的几个重要数据结构

// 每个cluster下的broker set信息,一个brokerName对应的broker set
private final HashMap<String/* clusterName */, Set<String/* brokerName */>> clusterAddrTable;
// 每个broker set中的broker信息(set中有哪些broker,每个broker的brokerId和brokerAddr)
private final HashMap<String/* brokerName */, BrokerData> brokerAddrTable;
// 每个broker的存活情况
private final HashMap<String/* brokerAddr */, BrokerLiveInfo> brokerLiveTable;
// 每个topic下的queue信息,包括每个broker set中读写队列的个数,consumer消费消息和producer发送消息的路由信息都从这个数据结构中获取
private 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获取,获取的方法如下:

// org.apache.rocketmq.client.impl.factory.MQClientInstance#updateTopicRouteInfoFromNameServer(java.lang.String, boolean, org.apache.rocketmq.client.producer.DefaultMQProducer)
public boolean updateTopicRouteInfoFromNameServer(final String topic, boolean isDefault,
DefaultMQProducer defaultMQProducer) {
try {
if (this.lockNamesrv.tryLock(LOCK_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS)) {
try {
TopicRouteData topicRouteData;
if (isDefault && defaultMQProducer != null) {
// 省略中间代码...
} else {
// 从manesrv获取topic的路由信息,namesrv从topicQueueTable获取到该topic对应的所有的QueueData
// 然后将每个brokerName下的BrokerData返回
topicRouteData = this.mQClientAPIImpl.getTopicRouteInfoFromNameServer(topic, 1000 * 3);
}
// 省略中间代码...
for (BrokerData bd : topicRouteData.getBrokerDatas()) {
// 每个broker set下所有的broker地址(ip:port)
this.brokerAddrTable.put(bd.getBrokerName(), bd.getBrokerAddrs());
}
// Update Pub info
{
// 将从namesrv获取到的路由信息转换为TopicPublishInfo
// 期间会将没有master的broker set的queue信息去除
TopicPublishInfo publishInfo = topicRouteData2TopicPublishInfo(topic, topicRouteData);
// 省略中间代码...
} catch (InterruptedException e) {
log.warn("updateTopicRouteInfoFromNameServer Exception", e);
} return false;
}

到此,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

// org.apache.rocketmq.client.impl.factory.MQClientInstance#findBrokerAddressInPublish
public String findBrokerAddressInPublish(final String brokerName) {
// 上面updateTopicRouteInfoFromNameServer方法将broker set下的broker地址信息保存到brokerAddrTable
// 再次重申:一个broker set下的broker的brokerName相同
HashMap<Long/* brokerId */, String/* address */> map = this.brokerAddrTable.get(brokerName);
if (map != null && !map.isEmpty()) {
// 没有花样,就是直接返回brokerId时MixAll.MASTER_ID的broker的ip:port信息
// 前面说过master的brokerId就是MixAll.MASTER_ID,所以获取到的broker是broker set中的master
return map.get(MixAll.MASTER_ID);
} return null;
}

终于真相大白,producer只会向是master的broker发送消息,也就是一个broker set中brokerId是0的broker。

producer只能发送消息到master,而不能发送到slave,这也说明了master负责读“写”,而slave只负责读(当然,这里只说明了“写”的部分,关于master 和slave的“读”下一篇介绍)。

总结

本篇介绍了RocketMQ究竟做了什么来实现作为一个消息队列中间件的高可用,由于篇幅会偏长,所以分为两篇文章来说明,下一篇说明文中遗留下的另一个问题——RocketMQ源码 — 六、 RocketMQ高可用(2)


参考:

关于高可用的系统

RocketMQ Architecture

RocketMQ源码 — 三、 Producer消息发送过程

上一篇:JQuery实现方法,简单示例


下一篇:解决MySQL中【Cannot load from mysql.proc. The table is probably corrupted