从消息中间件看分布式系统的多种套路

消息中间件作为分布式系统的重要成员，各大公司及开源均有许多解决方案。目前主流的开源解决方案包括RabbitMQ、RocketMQ、Kafka、ActiveMQ等。消息这个东西说简单也简单，说难也难。简单之处在于好用方便，接入简单使用简单，异步操作能够解耦系统间的依赖，同时失败后也能够追溯重试。难的地方在于，设计一套可以支撑业务的消息机制，并提供高可用架构，解决消息存储、消息重试、消息队列的负载均衡等一系列问题。然而难也不代表没有方法或者“套路”，熟悉一下原理与实现，多看几个框架的源码后多总结势必能找出一些共性。

消息框架大同小异，熟练掌握其原理、工作机制是必要的。就拿用的比较多的RocketMQ为引，来说说消息引擎的设计与实现。阿里的消息引擎经过了从Notify到Napoli、再到MetaQ三代的发展，现在已经非常成熟，在不同部门的代码中现在没准都还可以从代码里看到这一系列演进过程。当前的Apache RocketMQ 就是阿里将MetaQ项目捐赠给了Apache基金会，而内部还是沿用MetaQ的名称。

首先诠释几个消息相关的基本概念。

• 每个消息队列都必须建立一个Topic。
• 消息可以分组，每个消息队列都至少需要一个生产者Producer和一个消费者Consumer。生产者生产发送消息，消费者接收消费消息。
• 每个消费者和生产者都会分批提个ID。

RocketMQ 系统架构

接下来再来看看RocketMQ的架构，如图所示，简要描述一下几种角色及作用。

• NameServer
  • NameServer是消息Topic的注册中心，用于发现和管理消息生产者、消费者、及路由关系。
• Broker
  • 消息存储与转发的中转站，使用队列机制管理数据存储。Broker中会存储多份消息数据进行容错，以Master/Slave的架构保证系统的高可用，Broker中可以部署单个或多个Master。单个Master的场景，Master挂掉后，Producer新产生的消息无法被消费，但已经发送到Broker的消息，由于Slave节点的存在，还能继续被Consumer所消费；如果部署多个Master则系统能能正常运转。
  • 另外，Broker中的Master和Slave不是像Zookeeper集群中用选举机制进行确定，而是固定的配置，这也是在高可用场景需要部署多个Master的原因。
  • 生产者将消息发送到Broker中后，Broker会将消息写到本地的CommitLog文件中，保存消息。
• Producer
  • 生产者会和NameServer集群中某一节点建立长链接，定时从NamerServeri获取Topic路由信息，并且和Broker建立心跳。
• Consumer
  
  • 消费者需要给生产者一个明确的消费成功的回应，MetaQ才会认为消费成功，否则失败。失败后，RocketMQ会将消息重新发回Broker，在指定的延迟时间内进行重试，当重试达到一定的次数后（默认16次），MetaQ则认为此消息不能被消费，消息会被投递到死信队列。

这个架构看其实是否很熟悉？好像接触过的一些分布式系统的架构和这个长的都比较像是吧，甚至只要里面框图的角色稍微换换就能变成另外一个框架的介绍，比如Dubbo/Redis...。

并且在RocketMQ架构设计中，要解决的问题与其他分布式框架也可以触类旁通。Master/Slave机制，天然的读写分离方式都是分布式高可用系统的典型解决方案。

负载均衡

负载均衡是消息框架需要解决的又一个重要问题。当系统中生产者生产了大量消息，而消费者有多个或多台机器时，就需要平衡负载，让消息均分地被消费者进行消费。目前RocketMQ中使用了多种负载均衡算法。主要有以下几种，静态配置由于过于简单，直接为消费者配置需要消费的队列，因此直接忽略。

1. 求平均数法
2. 环形队列法
3. 一致Hash算法
4. Machine Room算法
5. 静态配置

来看一下源码，RocketMQ内部对以上负载均衡算法均有实现，并定义了一个接口 AllocateMessageQueueStrategy，采用策略模式，每种负载均衡算法都依靠实现这个接口实现，在运行中，会获取这个接口的实例，从而动态判断到底采用的是哪种负载均衡算法。

 1 public interface AllocateMessageQueueStrategy {
 2 
 3     /**
 4      * Allocating by consumer id
 5      *
 6      * @param consumerGroup current consumer group
 7      * @param currentCID current consumer id
 8      * @param mqAll message queue set in current topic
 9      * @param cidAll consumer set in current consumer group
10      * @return The allocate result of given strategy
11      */
12     List<MessageQueue> allocate(
13         final String consumerGroup,
14         final String currentCID,
15         final List<MessageQueue> mqAll,
16         final List<String> cidAll
17     );
18 
19     /**
20      * Algorithm name
21      *
22      * @return The strategy name
23      */
24     String getName();
25 }

1. 求平均数法

顾名思义，就是根据消息队列的数量和消费者的数量，求出单个消费者上应该负担的平均消费队列数，然后根据消费者的ID，按照取模的方式将消息队列分配到指定的consumer上。具体代码可以去Github上找，截取核心算法代码如下， mqAll就是消息队列的结构，是一个MessageQueue的List，cidAll是消费者ID的列表，也是一个List。考虑mqAll和cidAll固定时以及变化时，当前消费者节点会从队列中获取到哪个队列中的消息，比如当 averageSize 大于1时，这时每个消费者上的消息队列就不止一个，而分配在每个消费者的上的队列的ID是连续的。

 1 　　　　int index = cidAll.indexOf(currentCID);
 2         int mod = mqAll.size() % cidAll.size();
 3         int averageSize =
 4             mqAll.size() <= cidAll.size() ? 1 : (mod > 0 && index < mod ? mqAll.size() / cidAll.size()
 5                 + 1 : mqAll.size() / cidAll.size());
 6         int startIndex = (mod > 0 && index < mod) ? index * averageSize : index * averageSize + mod;
 7         int range = Math.min(averageSize, mqAll.size() - startIndex);
 8         for (int i = 0; i < range; i++) {
 9             result.add(mqAll.get((startIndex + i) % mqAll.size()));
10         }
11         return result;

2. 环形平均法

这种算法更为简单。首先获取当前消费者在整个列表中的下标index，直接用求余方法得到当前消费者应该处理的消息队列。注意mqAll的size和cidAll的size可以是任意的。

• 当ciAll.size() == mqAll.size() 时，该算法就是类似hashtable的求余分桶。
• 当ciAll.size() > mqAll.size() 时，那么多出的消费者上并不能获取到消费的队列，只有部分消费者能够获取到消息队列并执行，相当于在消费者资源充足的情况下，由于队列数少，所以使用其中一部分消费者就能满足需求，不用额外的开销。
• 当ciAll.size() < mqAll.size() 时，这样每个消费者上需要负载的队列数就超过了1个，并且区别于直接求平均的方式，分配在每个消费者上的消费队列不是连续的，而是有一定步长的间隔。

1         int index = cidAll.indexOf(currentCID);
2         for (int i = index; i < mqAll.size(); i++) {
3             if (i % cidAll.size() == index) {
4                 result.add(mqAll.get(i));
5             }
6         }
7         return result;

3. 一致Hash算法

循环所有需要消费的队列，根据队列toString后的hash值计算出处理当前队列的最近节点并分配给该节点。routeNode 中方法稍微复杂一些，有时间建议细看，这里就只说功能。

 1  　　　　Collection<ClientNode> cidNodes = new ArrayList<ClientNode>();
 2         for (String cid : cidAll) {
 3             cidNodes.add(new ClientNode(cid));
 4         }
 5 
 6         final ConsistentHashRouter<ClientNode> router; //for building hash ring
 7         if (customHashFunction != null) {
 8             router = new ConsistentHashRouter<ClientNode>(cidNodes, virtualNodeCnt, customHashFunction);
 9         } else {
10             router = new ConsistentHashRouter<ClientNode>(cidNodes, virtualNodeCnt);
11         }
12 
13         List<MessageQueue> results = new ArrayList<MessageQueue>();
14         for (MessageQueue mq : mqAll) {
15             ClientNode clientNode = router.routeNode(mq.toString());
16             if (clientNode != null && currentCID.equals(clientNode.getKey())) {
17                 results.add(mq);
18             }
19         }
20 
21         return results;

4. Machine Room算法

基于机房的Hash算法。这个命名看起来很诈唬，其实和上面的普通求余算法是一样的，只不过多了个配置和过滤，为了把这个说清楚就把源码贴全一点。可以看到在这个算法的实现类中多了一个成员 consumeridcs，这个就是consumer id的一个集合，按照一定的约定，预先给broker命名，例如us@metaq4，然后给不同集群配置不同的consumeridcs，从而实现不同机房处理不同消息队列的能力。

 1 /*
 2  * Licensed to the Apache Software Foundation (ASF) under one or more
 3  * contributor license agreements.  See the NOTICE file distributed with
 4  * this work for additional information regarding copyright ownership.
 5  * The ASF licenses this file to You under the Apache License, Version 2.0
 6  * (the "License"); you may not use this file except in compliance with
 7  * the License.  You may obtain a copy of the License at
 8  *
 9  *     http://www.apache.org/licenses/LICENSE-2.0
10  *
11  * Unless required by applicable law or agreed to in writing, software
12  * distributed under the License is distributed on an "AS IS" BASIS,
13  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14  * See the License for the specific language governing permissions and
15  * limitations under the License.
16  */
17 package com.aliyun.openservices.shade.com.alibaba.rocketmq.client.consumer.rebalance;
18 
19 import java.util.ArrayList;
20 import java.util.List;
21 import java.util.Set;
22 import com.aliyun.openservices.shade.com.alibaba.rocketmq.client.consumer.AllocateMessageQueueStrategy;
23 import com.aliyun.openservices.shade.com.alibaba.rocketmq.common.message.MessageQueue;
24 
25 /**
26  * Computer room Hashing queue algorithm, such as Alipay logic room
27  */
28 public class AllocateMessageQueueByMachineRoom implements AllocateMessageQueueStrategy {
29     private Set<String> consumeridcs;
30 
31     @Override
32     public List<MessageQueue> allocate(String consumerGroup, String currentCID, List<MessageQueue> mqAll,
33         List<String> cidAll) {
34         List<MessageQueue> result = new ArrayList<MessageQueue>();
35         int currentIndex = cidAll.indexOf(currentCID);
36         if (currentIndex < 0) {
37             return result;
38         }
39         List<MessageQueue> premqAll = new ArrayList<MessageQueue>();
40         for (MessageQueue mq : mqAll) {
41             String[] temp = mq.getBrokerName().split("@");
42             if (temp.length == 2 && consumeridcs.contains(temp[0])) {
43                 premqAll.add(mq);
44             }
45         }
46 
47         int mod = premqAll.size() / cidAll.size();
48         int rem = premqAll.size() % cidAll.size();
49         int startIndex = mod * currentIndex;
50         int endIndex = startIndex + mod;
51         for (int i = startIndex; i < endIndex; i++) {
52             result.add(mqAll.get(i));
53         }
54         if (rem > currentIndex) {
55             result.add(premqAll.get(currentIndex + mod * cidAll.size()));
56         }
57         return result;
58     }
59 
60     @Override
61     public String getName() {
62         return "MACHINE_ROOM";
63     }
64 
65     public Set<String> getConsumeridcs() {
66         return consumeridcs;
67     }
68 
69     public void setConsumeridcs(Set<String> consumeridcs) {
70         this.consumeridcs = consumeridcs;
71     }
72 }

由于近些年阿里海外业务的扩展和投入，RocketMQ 等中间件对常见的海外业务场景的支持也更加健全。典型的场景包括跨单元消费以及消息路由。跨单元消费是比较好实现的，就是在consumer中增加一个配置，指定接收消息的来源单元，RocketMQ内部会完成客户端从指定单元拉取消息的工作。而全球消息路由则是需要一些公共资源，消息的发送方只能将消息发送到一个指定单元/机房，然后将消息路由到另外指定的单元，consumer部署在指定单元。区别在于一个配置在客户端，一个配置在服务端。

总结