导语

Apache Pulsar 是一个多租户、高性能的服务间消息传输解决方案，支持多租户、低延时、读写分离、跨地域复制（GEO Replication）、快速扩容、灵活容错等特性，GEO Replication 可以原生支持数据和订阅状态在多个集群之间进行复制，GEO 目前在 Apache InLong 内部已经有长期稳定的实践，本文主要讲述 GEO 中的订阅状态的同步。

GEO 简介

GEO Replication 提供了数据在多个集群之间进行复制的能力。

上图描述了三个集群，并且集群之间配置了不同的 GEO Replication 策略，其中

• Cluster-A 和 Cluster-B 是双向复制，两个集群中的 Topic 数据都会复制到对端集群，即集群 A 的数据会被复制到集群 B，集群 B 的数据也会被复制到集群 A，A、B 两个集群都有对方的全部数据；

• Cluster-A 和 Cluster-C 是单向复制：A 集群的数据会被复制到 C 集群，C 集群的数据不会被复制到 A 集群；

• Cluster-B 和 Cluster-C 没有复制关系：集群 B 和 C 之间不会产生任何数据复制。

上述描述数据同步/复制的一个典型的场景，GEO Replication 中的另外一个场景就是订阅状态同步。

订阅状态同步的场景

订阅同步的一个典型的应用常见是集群容灾，正常情况下只有主集群提供写入和消费服务，主集群故障之后，生产和消费会切换到备集群。

生产的切换是无缝的，切换集群之后可以继续写入；消费比生产会复杂一些，如果只同步数据，在集群切换之后，备集群的订阅会重复消费历史数据，为了解决这个问题，就需要在两个集群之间同步订阅的状态，目前订阅同步的主要信息就是订阅的 MarkDeletePosition（MDP） 信息。

如上图：在主、备两个集群之间，每个 Topic（分区）的 Ledger 并不是一一对应的，比如在主集群中，订阅 sub-00 消费到了一条消息，这个消息所在的 Ledger 是 Ledger-x；经过复制之后，在备集群中这条消息对应的 Ledger 是 Ledger-y，这里 Ledger-x 和 Ledger-y 没有直接关系，所以订阅状态（MDP）不能简单的直接映射。

GEO 订阅状态同步原理

订阅状态的同步，大体上可以分为两个主要的步骤：

• 第一步是实现两个集群之间 MessageId（可以理解为 Offset 信息）的映射，即在主集群的一条消息的 MessageId 复制到备集群之后的 MessageId；

• 第二步是在主集群中一个订阅 ack 数据时，如果有 （MDP） 的变动，根据第一步中的主、备集群 MessageId 的映射关系，将主集群的 MDP 信息映射到备集群订阅的 MDP 中。

下面我们来详细看下整个流程。

MessageId 映射

MessageId 映射最直观的方法，就是维护主、备集群中每个 Message 的映射关系，但是这种方案的需要维护的映射关系太多，代价太大。

Pulsar 采用的方式是一个定时任务的方式，每隔一段时间同步一次主、备集群 LAC 信息之间的关系。假设集群 A 向集群 B 复制数据和订阅状态信息。

首先，集群 A 会定时生产一个 SnapshotRequest 信息，写入到本地 Topic（分区）中，这个信息会随着数据复制写入到集群 B 的 Topic 中。

B 集群会处理 SnapshotRequest 信息，然后将本地 Topic（分区）的 LAC（LAC-B） 信息封装在 SnapshotRespnse 中，写入到本地 Topic 中，通过 GEO Replciation 复制到 A 集群。

集群 A 在处理 SnapshotRespnse 时，记录 SnapshotRespnse 在本地的 MessageId（LocalMessageId） 和 LAC-B 的映射关系，由于 A -> SnapshotRequest -> B -> SnapshotRespnse -> A 的操作顺序，可以保证集群 A 订阅的 MDP 大于 LocalMessageId 时，LAC-B 对应的数据一定是被消费过的，通过这种方式实现了两个集群之间 MessageId 的映射关系。

订阅信息同步

集群 A 中的订阅会不断消费、Ack，当 Ack 触发了 MDP 的移动时，集群 A 会检查 LocalMessageId 是否小于 MDP，如果发现小于，说明需要更新集群 B 订阅的 MDP 信息，此时集群 A 会根据映射关系，找到 LAC-B 信息，然后构造一个 ReplicatedSubscriptionsUpdate 消息，写入到本地 Topic，这个 ReplicatedSubscriptionsUpdate 消息会通过 GEO 复制到集群 B。

集群 B 接收到 ReplicatedSubscriptionsUpdate 消息之后，会解析出 LAC 和订阅信息，然后更新订阅的 MDP。

至此，就完成了订阅状态的一次复制流程。

总结与思考

Pulsar 的订阅状态复制，依赖于原生的 GEO Replication 机制，并且需要主备集群之间双向的交互，所以对于单向复制的 GEO 集群，订阅状态是不能实现订阅状态同步的。

另外，当前的订阅状态同步，只考虑了 MDP 信息，实际上对于一个订阅（尤其是 Shared 和 Key-Shared 类型的订阅），订阅的 IndividuallyDeletedMessages 信息也是很重要的，尤其是在有大量 Consumer 都使用 Individual Ack 的场景，如果不同步 IndividuallyDeletedMessages 信息，就会导致数据的重复。

由于 IndiviindividuallyDeletedMessages 记录的是每个 message 的 Ack 情况，所以要解决这个问题就需要：

• 记录主、备集群每个 MessageId 的映射关系，比如在复制消息属性中记录原始消息的 MessageId 信息。

• 复制 IndiviindividuallyDeletedMessages 到 备集群。

备集群的订阅在消费数据时，根据主、备 集群的 MessageId 映射关系以及主集群复制过来的 IndiviindividuallyDeletedMessages，就可以判定这条消息是否已经被 Ack，如果 Ack 则不推送给 Consumer，这样就可以实现切换集群订阅时数据不重复。