Apache Pulsar——跨机房复制

前言

在大型的分布式系统中，都会涉及到跨多个数据中心的需求，通常会使用跨地域复制机制提供额外的冗余防止服务无法正常运作。Apache Pulsar 的跨地域多机房互备特性，是 Pulsar 企业级特性的重要组成部分，它在保证数据稳定可靠的同时，为用户提供了便捷的操作和管理。

 

Pulsar 自带的跨地域复制机制（Geo-Replication）可以提供一种全连接的异步复制.

在上图系统中，有三个数据中心：Cluster-A、 Cluster-B、 Cluster-C。用户创建的一个 Topic 主题 T1 设置了跨越三个数据中心做互备。在三个数据中心中，分别有三个生产者：P1、P2、P3，它们往主题 T1 中发布消息；有两个消费者：C1、C2，订阅了这个主题，接收主题中的消息。

 

当消息由本数据中心的生产者发布成功后，会立即复制到其他两个数据中心。消息复制完成后，消费者不仅可以收到本数据中心产生的消息，也可以收到从其他数据中心复制过来的消息。

 

它的工作机制是在 Broker 内部，为跨地域的数据复制启动了一组内嵌的额外生产者和消费者。当外部消息产生后，内嵌的消费者会读取消息；读取完成后，调用内嵌的生产者将消息立即发送到远端的数据中心。

 

跨地域复制需要设置“租户”在数据中心之间的访问权限。

 

在配置了跨地域复制后，每个发送进来的消息，首先被保存在本地集群中；然后异步地推送到远端的集群。如果本地集群和远端集群之间没有网络问题，消息会被立即推送给远端集群。这样端到端的发送延迟主要由集群之间网络的决定。

 

在图示中，无论生产者（Producer）P1、P2 和 P3 在什么时候分别将消息发布给 Cluster A、 Cluster B 和 Cluster C 中的主题T1，这些消息均会立刻复制到整个集群。一旦完成复制，消费者（Consumer）C1 和 C2 即可从自己所在的集群消耗这些消息，并且保持消息在每个Producer 内部的发送顺序。

 

因为消息已经从其他远端集群发送到本地集群的 Topic，所以每个集群内部都会保留这个 Topic 中产生的所有消息。对于每个Consumer 来说，Consumer 的订阅（subscription，维护 Consumer 对 Topic 的消费和 ack 的位置）是针对本地集群的 Topic，相当于 Consumer 消费本地集群的消息。

一、多机房部署

Pulsar 多机房部署如下图：

上面的 Pulsar 架构中，Pulsar 3 个集群分别部署在北京、上海、贵阳 3 个机房，每个机房一套集群，每个集群中都有一个 Topic1，并且对应着订阅是 Subscription1。但是 3 个集群之间并没有数据同步。如果某一个机房发生故障，那这个机房的存量消息将不能被消费掉。

二、跨地域复制 (GEO-Replication)

Pulsar 最初是在 Yahoo 内部开发，在设计之初就加入了对 Yahoo 全球十多个机房的跨地域复制的需求。在上面的例子中，如果这 3 个机房可以相互同步数据，那即使某一个机房发生故障了，这个机房的存量数据因为已经被同步到其他机房，可以被其他机房的消费者消费掉。如下图：

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 集群自己不生产消息，只接受从北京、上海、贵阳三个机房的复制数据，如下图：

三、复制配置

Pulsar 中 Topic 的格式如下：

persistent://tenant/namespace/topic

一个 Topic 的上级目录有 namespace 和 tenant。要允许两个集群间消息跨地域复制，首先要允许 tenant(租户) 有权限访问两个集群。而跨地域复制是在 namespace 级别进行管理的，如果允许一个 namespace 跨地域复制，那发布到这个 namespace 上的任意一个 topic 的消息，都会被复制到指定集合的所有集群中。

3.1 tenant 授权

要使用跨地域复制，首先要给租户设置访问权限。下面命令给 my-tenant 这个租户授予了 pulsar-shanghai、pulsar-beijing 和 pulsar-guiyang 的访问权限。

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

3.2 创建namespace

bin/pulsar-admin namespaces create my-tenant/my-namespace

3.3 namespace 级别启动

跨地域复制是在 namespace 级别进行管理的，租户拥有了权限后，把 namespace 指定给要复制的集群：

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 这两个集群间复制。

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.4 Topic 级别启动

要让一个 Topic 能够跨地域复制，要在 Topic 级别启动：

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

3.5 防止循环复制

如果配置了上海机房和北京机房之间的跨地域复制，那从上海机房复制到北京机房后，消息有没有可能从北京机房再复制到上海机房呢？

 

当然不会。上海机房发送消息到北京机房时，会给消息加一个 Property，用来表示是哪个机房生产的数据。北京机房收到这个数据后，就会知道是从别的机房复制来的，Replicator 中的 Cursor 在订阅消息时就会把这部分消息过滤掉。

 

说明：在未来，如果新增了数据中心，或者关闭数据中心，可以随时进行配置调整操作，而且pulsar表示这样的操作并不会对流量有任何影响。

# 以下模拟了一个新增数据中心的操作: bin/pulsar-admin namespaces set-clusters my-tenant/my-namespace \ --clusters cluster1, cluster2, cluster3

四、目前 pulsar Geo-Replication 存在的问题:

• Pulsar 只能保证单机房生产的消息顺序，在多机房的场景下没办法保证多个机房的消息全局有序。
• 由于 cursor snapshot 是定期进行的，在时间上精确度不会太高，多少有些偏差。
• 目前只会同步“Mark delete position”的位置，对于单独签收的消息暂时无法同步。
• 只有在所有相关集群都处于「可用」状态时，才可以进行 cursor snapshot。
• 当使用 cursor snapshot 后，会产生一些缓存，影响到后续涉及 backlog 的计算结果。

五、总结

一句话概括，Pulsar 的跨地域复制，其实就是在一个本地集群中创建一个 Producer，把异地的集群作为这个 Producer 的发送地址，将本地集群的消息发送过去，并且在本地维护一个 Cusor 来保证消息可靠性和幂等性。

 

参考： https://blog.csdn.net/zjj2006/article/details/121757473

https://www.jianshu.com/p/fbfbeea1982c