您的位置:  首页 > 技术杂谈 > 正文

Apache Pulsar 技术系列 - GEO replication 中订阅状态的同步原理

2023-07-26 15:00 https://my.oschina.net/u/4587289/blog/10090802 腾讯云中间件 次阅读 条评论

导语

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,这样就可以实现切换集群订阅时数据不重复。

展开阅读全文
  • 0
    感动
  • 0
    路过
  • 0
    高兴
  • 0
    难过
  • 0
    搞笑
  • 0
    无聊
  • 0
    愤怒
  • 0
    同情
热度排行
友情链接