导语
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 到 备集群。文章来源:https://www.toymoban.com/news/detail-604142.html
备集群的订阅在消费数据时,根据主、备 集群的 MessageId 映射关系以及主集群复制过来的 IndiviindividuallyDeletedMessages,就可以判定这条消息是否已经被 Ack,如果 Ack 则不推送给 Consumer,这样就可以实现切换集群订阅时数据不重复。文章来源地址https://www.toymoban.com/news/detail-604142.html
到了这里,关于Apache Pulsar 技术系列 - GEO replication 中订阅状态的同步原理的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!