跨地域復制是 Apache Pulsar 企業級特性的重要組成部分，它保證了系統的高可用，在操作和管理上也非常便捷，今天用 5 張圖來帶大家學習這個功能。

1.多機房部署

Pulsar 多機房部署如下圖：

上面的 Pulsar 架構中，Pulsar 3 個集群分別部署在北京、上海、貴陽 3 個機房，每個機房一套集群，每個集群中都有一個 Topic1，并且對應著訂閱是 Subscription1。但是 3 個集群之間并沒有數據同步。如果某一個機房發生故障，那這個機房的存量消息將不能被消費掉。

2.跨地域復制 (GEO-Replication)

Pulsar 最初是在 Yahoo 內部開發，在設計之初就加入了對 Yahoo 全球十多個機房的跨地域復制的需求。在上面的例子中，如果這 3 個機房可以相互同步數據，那即使某一個機房發生故障了，這個機房的存量數據因為已經被同步到其他機房，可以被其他機房的消費者消費掉。如下圖：

那 Pulsar 的跨地域復制到底是怎么做的呢?

2.1 存儲模型回顧

首先我們回顧一下 Pulsar 的存儲模型。我們知道，Pulsar 的消息持久化用到了存儲系統 BookKeeper，如下圖：

Producer 生產完消息后，會刷到底層的 BookKeeper 存儲引擎進行持久化。

Consumer 創建的時候要訂閱一個 Topic，Pulsar 就會給它分配一個 Subscription 進行綁定，如上圖 Consumer 綁定了 Subscription2。

Subscription 會持續從 Ledger 中獲取消息推給 Consumer，當然前提是 Consumer 要有消息緩存空間。

Consumer 消費完成一個消息后，回復給 Subscription 一個 ACK，Subscription 收到 ACK 后把游標向后推一位。這個游標也是保存在了 BookKeeper，BookKeeper 會專門為這個游標開一個 Ledger。

2.2 跨地域復制過程

Pulsar 的跨地域復制跟上面的存儲模型很類似，集群中多了一個 Replicator。以上海機房復制到北京機房為例，如下圖：

上海機房的 Pulsar 集群中有一個 Replicator，這個 Replicator 中有一個 Producer-R，綁定的了北京機房的 Topic1，把數據用生產者的方式發送到北京機房。

上海機房集群中生產的消息首先在本地集群中持久化，然后再被異步轉發到北京集群。

上海機房 Replicator 中的 Producer-R 跟集群中的 Producer1 沒有任何關系，它配置的集群地址是北京機房集群地址。

整個復制流程如下：

• Producer1 生產消息到上海機房 Topic1;
• 上海機房把消息持久化到 BookKeeper;
• BookKeeper 返回成功后把消息推給 Replicator 的 Cursor;
• Replicator 的 Cursor 通過 Producer-R 把消息發給北京機房 Topic1;
• 北京機房 Topic1 寫入 BookKeeper 成功后給上海機房 Replicator 的 Cursor 回復一個 ACK，上海機房 Cursor 收到 ACK 后通過 Producer-R 推送下一條消息。

2.3 消息丟失和冪等

因為在 Replicator 中維護了一個 Cursor，如果一條消息沒有收到北京機房的 ACK，Replicator 可以通過 Producer-R 再次把這條消息發送北京機房，這樣可以防止消息丟失。

如果因為網絡問題，Producer-R 給北京機房推送消息后，北京機房回復的 ACK 上海機房沒有收到，怎么處理呢?Producer-R 會再次給北京機房發送同一條消息，這種場景很容易導致消息重復。為了解決消息冪等的問題，Pulsar 提供了一個 Producer 冪等配置，北京機房開啟這個設置后，broker 中會緩存一個內部 Cursor，用于保存收到的上一條消息的 MessageId ，如果收到一條新消息的 MessageId 小于等于當前 Cursor 中緩存的 MessageId，這條消息就會被丟掉。

2.4 消息順序

上圖中，上海機房的 Producer-R 和 北京機房的 Producer2 都往北京機房的 Topic1 寫消息，消息的順序怎么保證呢?

因為跨機房復制是異步的過程，Pulsar 只能保證上海機房和北京機房各自寫入消息的順序性，比如上海機房Producer-R 寫入 msg1~msg5 這 5 條消息，北京機房 Producer2 寫入 msgA~msgE 這5條消息，最終消息順序可能如下：

2.5 低延遲

跨區域復制的低延遲從兩個方面來保證：

Replicator 和 broker 是在一個進程中，這樣減少了數據拷貝

跨地域復制采用異步方式

2.6 ZooKeeper 集群

跨機房復制可以采用全局 ZooKeeper 集群，把 Pulsar 集群信息注冊到 ZooKeeper 集群。如下圖：

這樣每個集群就可以根據 ZooKeeper 中保存的信息來創建本地的 Replicator。

但是如果沒有全局 ZooKeeper 集群，因為保存的數據是輕量級的，使用本地 ZooKeeper 集群也是可以的。如下圖：

這樣每個機房的 Pulsar 集群從本地 ZooKeeper 中獲取到需要復制的遠程集群信息，就可以創建 Replicator 了。這種情況反而更加靈活。因為下面這種方式的 Pulsar 集群，全局 ZooKeeper 是不能滿足要求的。

比如現在有一個西安機房的 Pulsar 集群自己不生產消息，只接受從北京、上海、貴陽三個機房的復制數據，如下圖：

3.復制原理

Pulsar 中 Topic 的格式如下：

1. persistent://tenant/namespace/topic

一個 Topic 的上級目錄有 namespace 和 tenant。要允許兩個集群間消息跨地域復制，首先要允許 tenant(租戶) 有權限訪問兩個集群。而跨地域復制是在 namespace 級別進行管理的，如果允許一個 namespace 跨地域復制，那發布到這個 namespace 上的任意一個 topic 的消息，都會被復制到指定集合的所有集群中。

3.1 tenant 授權

要使用跨地域復制，首先要給租戶設置訪問權限。下面命令給 my-tenant 這個租戶授予了 pulsar-shanghai、pulsar-beijing 和 pulsar-guiyang 的訪問權限。

1. bin/pulsar-admin tenants create my-tenant --admin-roles my-admin-role --allowed-clusters pulsar-shanghai,pulsar-beijing,pulsar-guiyang

3.2 namespace 級別啟動

跨地域復制是在 namespace 級別進行管理的，租戶擁有了權限后，把 namespace 指定給要復制的集群：

1. bin/pulsar-admin namespaces set-clusters my-tenant/my-namespace --clusters pulsar-shanghai,pulsar-beijing,pulsar-guiyang

namespace 級別的復制可以隨時改變，改變后立刻生效。

namespace 配置跨地域復制后，默認該 namespace 下創建的所有 Topic 都會復制到列表中其他集群。如果要選擇固定的集群進行復制，可以使用 Pulsar Client 來指定，比如 Java Client 下面的代碼只允許 my-topic 這個 topic 在pulsar-shanghai,pulsar-beijing 這兩個集群間復制。

1. List restrictReplicationTo = Arrays.asList(
2. "pulsar-shanghai",
3. "pulsar-beijing"
4. );
5. Producer producer = client.newProducer()
6. .topic("my-topic")
7. .create();
8. producer.newMessage()
9. .value("my-payload".getBytes())
10. .setReplicationClusters(restrictReplicationTo)
11. .send();

3.3 Topic 級別啟動

要讓一個 Topic 能夠跨地域復制，要在 Topic 級別啟動：

bin/pulsar-admin topics set-replication-clusters --clusters pulsar-shanghai,pulsar-beijing,pulsar-guiyang my-tenant/my-namespace/Topic1

3.4 防止循環復制

如果配置了上海機房和北京機房之間的跨地域復制，那從上海機房復制到北京機房后，消息有沒有可能從北京機房再復制到上海機房呢?

當然不會。上海機房發送消息到北京機房時，會給消息加一個 Property，用來表示是哪個機房生產的數據。北京機房收到這個數據后，就會知道是從別的機房復制來的，Replicator 中的 Cursor 在訂閱消息時就會把這部分消息過濾掉。

總結

一句話概括，Pulsar 的跨地域復制，其實就是在一個本地集群中創建一個 Producer，把異地的集群作為這個 Producer 的發送地址，將本地集群的消息發送過去，并且在本地維護一個 Cusor 來保證消息可靠性和冪等性。

原文鏈接：https://mp.weixin.qq.com/s/3V4EAMPGD-yklgcS95hPLw