從消息中間件看分佈式系統的多種套路

時間 2020-06-11

原文原文鏈接

　　消息中間件做爲分佈式系統的重要成員，各大公司及開源均有許多解決方案。目前主流的開源解決方案包括RabbitMQ、RocketMQ、Kafka、ActiveMQ等。消息這個東西說簡單也簡單，說難也難。簡單之處在於好用方便，接入簡單使用簡單，異步操做可以解耦系統間的依賴，同時失敗後也可以追溯重試。難的地方在於，設計一套能夠支撐業務的消息機制，並提供高可用架構，解決消息存儲、消息重試、消息隊列的負載均衡等一系列問題。然而難也不表明沒有方法或者「套路」，熟悉一下原理與實現，多看幾個框架的源碼後多總結勢必能找出一些共性。java

　　消息框架大同小異，熟練掌握其原理、工做機制是必要的。就拿用的比較多的RocketMQ爲引，來講說消息引擎的設計與實現。阿里的消息引擎通過了從Notify到Napoli、再到MetaQ三代的發展，如今已經很是成熟，在不一樣部門的代碼中如今沒準都還能夠從代碼裏看到這一系列演進過程。當前的Apache RocketMQ 就是阿里將MetaQ項目捐贈給了Apache基金會，而內部仍是沿用MetaQ的名稱。算法

首先詮釋幾個消息相關的基本概念。express

每一個消息隊列都必須創建一個Topic。
消息能夠分組，每一個消息隊列都至少須要一個生產者Producer和一個消費者Consumer。生產者生產發送消息，消費者接收消費消息。
每一個消費者和生產者都會分批提個ID。

RocketMQ 系統架構

　　接下來再來看看RocketMQ的架構，如圖所示，簡要描述一下幾種角色及做用。 apache

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中使用了多種負載均衡算法。主要有如下幾種，靜態配置因爲過於簡單，直接爲消費者配置須要消費的隊列，所以直接忽略。app

求平均數法
環形隊列法
一致Hash算法
Machine Room算法
靜態配置

　　來看一下源碼，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能夠是任意的。less

當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部署在指定單元。區別在於一個配置在客戶端，一個配置在服務端。