[[438652]]

本文轉(zhuǎn)載自微信公眾號「程序員jinjunzhu」，作者jinjunzhu。轉(zhuǎn)載本文請聯(lián)系程序員jinjunzhu公眾號。

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

1.多機房部署

Pulsar 多機房部署如下圖：

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

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

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

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

2.1 存儲模型回顧

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

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

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

Subscription 會持續(xù)從 Ledger 中獲取消息推給 Consumer，當(dāng)然前提是 Consumer 要有消息緩存空間。

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

2.2 跨地域復(fù)制過程

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

上海機房的 Pulsar 集群中有一個 Replicator，這個 Replicator 中有一個 Producer-R，綁定的了北京機房的 Topic1，把數(shù)據(jù)用生產(chǎn)者的方式發(fā)送到北京機房。

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

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

整個復(fù)制流程如下：

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

2.3 消息丟失和冪等

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

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

2.4 消息順序

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

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

2.5 低延遲

跨區(qū)域復(fù)制的低延遲從兩個方面來保證：

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

跨地域復(fù)制采用異步方式

2.6 ZooKeeper 集群

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

這樣每個集群就可以根據(jù) ZooKeeper 中保存的信息來創(chuàng)建本地的 Replicator。

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

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

比如現(xiàn)在有一個西安機房的 Pulsar 集群自己不生產(chǎn)消息，只接受從北京、上海、貴陽三個機房的復(fù)制數(shù)據(jù)，如下圖：

3.復(fù)制原理

Pulsar 中 Topic 的格式如下：

persistent://tenant/namespace/topic 

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

3.1 tenant 授權(quán)

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

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

3.2 namespace 級別啟動

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

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

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

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

List<String> restrictReplicationTo = Arrays.asList( 
        "pulsar-shanghai", 
        "pulsar-beijing" 
); 
Producer producer = client.newProducer() 
        .topic("my-topic") 
        .create(); 
producer.newMessage() 
        .value("my-payload".getBytes()) 
        .setReplicationClusters(restrictReplicationTo) 
        .send(); 

3.3 Topic 級別啟動

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

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

3.4 防止循環(huán)復(fù)制

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

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

總結(jié) 

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