在 Pulsar 里，可以基于 geo-replication 将数据在多个机房进行复制，这个特性主要是复制的消息数据，而今天我们分享的主要是订阅状态的跨机房同步。
订阅状态主要指订阅的消费进度，开启订阅状态跨机房复制可以在其中某个机房出现故障时，consumer 切换其他机房可以根据消费进度继续进行消费，这样可以减少消息的重复消费以及consumer丢失消息的问题。
在正式开始之前，先来回顾一下近期的 Pulsar 动态。
StreamNative Cloud 功能已发布，你可以在 StreamNative 官网进入 cloud 页面相关，进行下载和创建集群等。更多详情参考文章➡️开源流数据公司 StreamNative 推出 Pulsar 云服务，推进企业“流优先”进程
在本周三（即今晚 8-9 点），将会举行一场 Pulsar Community Meeting，有 Apache Pulsar 的 PMC 和众多 contributor 共同参与，主要是分享一些近期 Apache Pulsar 的安排和计划，共同探讨关于 Apache Pulsar 的使用与实践等等。
如果你是 Pulsar 的 contributor，也欢迎参与到此次交流会中，分享你对 Pulsar 的使用和建议等等，更多参会详情和讨论可以参考下方链接。
https://github.com/streamnative/pulsar-community-loc-cn/issues/1
在之后 2.7.0 版本中，正在测试加入这些功能：
Support MaxUnackedMessagesOnConsumer on topic level
Support Persistence Policies on topic level
Support deduplication on topic level
在 2.4.0 版本之后，Pulsar 加入了 geo replication 特性。如下图所示，三个 cluster 都有 Topic T1 存在，他们的数据会互相之间进行复制。
也就是说 producer 可以连接到任意一个 Topic T1 去写数据，比如数据在集群 A 写入数据，那么也会往 B 和 C 集群同步，其他同理。假设三个集群不在同一个地方，通过 geo-replication 特性，就可以完成跨地域复制同时提高消息容灾能力。关于如何配置 geo cluster，可以参考下方链接。
http://pulsar.apache.org/docs/en/administration-geo/在上图中，consumer 也可以连接到任意集群中，同时共用 Subscription S1。如果说没有进行订阅状态复制的话，上图中两个集群里的 Subscription S1 只是名字相同，而并不是同一个 Subscription。这样就会导致切换集群时消费进度无法同步，导致consumer消费多余的数据。
Replicated subscription
订阅状态复制
订阅状态复制解决的问题主要是将两个集群中 topic 里的订阅进行状态同步，相当于你在下载电脑游戏时，中间休眠/断网后重新下载时，是接着之前的进度进行的，而不是重新又从头下载。在真正讲解「订阅模式复制状态」前，我们需要了解几个概念。首先是 Message ID，这个在我们第一次 TGIP-CN 的分享里有提到过。Message ID 字段一般由这几个：ledger-id（在哪个 segment）、entry-id（entry 在这个 segment 的位置）、batch-id（消息被匹配后的位置）、partition-index（消息在 topic 的哪个 partition）构成。其次是 Message Acknowledgment（消息签收），Pulsar 目前支持两种签收方式：单条逐一签收和累积签收。累积签收的操作比较简单，但是单条签收的话需要对 broker 的状态进行保存。
比如连续签收了 0、1、2、3、4，那么就会在 4 上标记“Mark delete position”，6 和 9 单独签收了，也会进行标记。这种状态都可以被记录到 ledger 中，方便后续数据的恢复。订阅复制其实就是对 message 签收状态的一些复制，也就是关系到 「Mark delete position」和单独签收的复制。
上图是在同一个 topic 里，集群 A 和集群 B 的 Message ID 列举。比如在集群 A 里写入的 Message ID 为（1，0），复制到集群 B 时当前 topic 所用的 Message ID 就变成了（3，0）。那么如何解决不同集群中 Message ID 对应关系呢？目前我们采用的做法是定期做 cursor snapshot，即定期查看一下多个集群中 Message ID 的对应关系。>>> Cursor snapshot 流程
Cursor snapshot 会从一个集群中，定期发起`ReplicatedSubscriptionsSnapshotRequest`的指令，从一个时间点发起发给不同的集群。这个请求一般会包含这个 `snapshot_id` 和 `source_cluster` 两个信息。
当集群 B 和 C 收到请求后，会给集群 A 返回 `ReplicatedSubscriptionsSnapshotResponse`指令，这时会回给集群 A，告诉它当前集群的 ledger ID 和 entry ID（Message ID 包含的信息）。集群 A 就会根据 B 和 C 返回的消息来确定他们相对应的 Message ID 关系。
当知道了集群间对应 Message ID 后，集群 A 也需要把 snapchat 对应结果的状态传递给其他集群。不管是 request 还是 response 以及最后 snapshot 结果，都是与 topic 数据存储在一起的。
0、1、2 等数字这些是 topic 里的消息，标记 S 的位置并不是消息，而是一种标记。它可以是 request 的标记，也可以是 response 的标记等。这样的好处是在跨地域复制中，数据和标记都可以复制同步到其他集群中。相当于利用消息复制的机制，达到传递消费进度状态的目的。记录完成以后，后续的更新状态如何记录呢？总不能一直不变吧。当其中某一个集群 ack 速度超过了当前 snapshot 进度，对于正在连接的集群，就可以发起`ReplicatedSubscriptionUpdate`指令，通知其他集群「xxx 更新了，准备作战」的状态。
这时其他集群就可以接收到此次更新他们各自需要更新的信息，会进行相应地修改，达到同步状态的特性。
在 Pulsar 中，当创建 consumer 时，默认是不会开启「订阅状态复制」这个特性的，你需要在 broker 中利用 `enableReplicatedSubscriptions=true` 指令来开启这个特性。同时在创建 consumer 时将 `replicateSubscriptionState( )`改为 ture。
开启后也可以通过设置一些指令来控制「订阅状态复制」时的一些行为，如下图所示。
🧐 由于 cursor snapshot 是定期进行的，在时间上精确度不会太高，多少有些偏差。
🧐 目前只会同步“Mark delete position”的位置，对于单独签收的消息暂时无法同步。
🧐 只有在所有相关集群都处于「可用」状态时，才可以进行 cursor snapshot。
🧐 当使用 cursor snapshot 后，会产生一些缓存，影响到后续涉及 backlog 的计算结果。
A：目前我们的云服务是基于 Google Cloud 的，下一步目标计划就是会适应国内云厂商的一些平台，比如阿里云、腾讯云等。Q：有 Pulsar transaction 相关的介绍么？2.7.0 预计什么时候正式发布？A：我们将在 10 月份会发布 Pulsar transaction 的预览版。2.7.0 版本的话预计也在 10 月份左右发布吧。敬请期待！本次分享主要针对 Pulsar 中「订阅状态复制」进行了一些操作和概念上的分享，希望通过本次分享，大家可以对相关知识有了更清晰的理解。更多细节回放可查看下方视频。
本周末 TGIP-CN，我们将邀请到 BIGO 大数据平台的陈航，为我们分享关于 Apache Pulsar 在 BIGO 的一些实践。详情可扫描下方二维码进行报名～
👍 相关阅读