我第一次认真研究 Pulsar 的跨地域复制是因为一个很现实的局面消息集群要跨两个机房既要容灾又要让两边的应用都能就近生产和消费。最早的想法很简单——应用层双写结果数据竞争、顺序错乱、补偿逻辑全部压到业务代码上维护成本直线上升。后来切到 Pulsar 的跨地域复制geo-replication我才真正意识到一份消息由消息系统自己在后台异步复制到另一个集群和业务自己想办法写两份是两种完全不同的架构思路。这篇文章我不打算从官方文档抄概念而是用 8 张图把 Pulsar 跨地域复制的核心机制和落地经验过一遍适合正在做消息中间件选型、异地多活或容灾方案的同学。1. 先讲清楚跨地域复制解决什么问题以及“双写”为什么靠不住1.1 三种典型场景你对号入座看看是哪种跨地域复制不是为某个单一需求设计的我接触下来最常见的落地场景有三类。第一类是最经典的容灾备份。主集群正常承接写入后台把消息异步复制到备集群。平时备集群不承担生产流量只作为只读副本或者承担少量离线分析、报表消费。主集群一旦出问题直接切到备集群继续消费业务能快速恢复。第二类是就近读写。用户或业务流量分散在不同地域如果都打到同一个中心机房网络延迟很难看。于是让各地域的应用连本地的 Pulsar 集群消息在后台互相复制最终每个集群都持有全量数据。这样读写都在本地完成用户体验好也不丢数据。第三类是异地多活。两个机房都可以生产、都可以消费数据双向复制。任何一个机房宕机另一个机房能完整接管流量。这种对复制机制的要求最高也是这篇文章后面重点讨论的场景。不管哪一种核心需求都一样数据要在多个物理集群之间保持一致而且是系统级的一致性不是业务拼出来的。这就是跨地域复制存在的意义。1.2 双写的问题不在“写两次”而在“顺序和失败”很多人第一反应是不用搞那么复杂业务层两条连接两边各写一份不就行了我最早也这么想过直到我梳理清楚双写要面对的三个问题。第一个是顺序问题。单机房里消息严格有序双写时两个机房同时写入网络延迟不同、落盘时间不同两边看到的顺序可能完全相反。消费者在 A 机房读到 1、2、3在 B 机房却可能读到 3、1、2。对订单状态、账户流水这类数据来说顺序错了就是事故。第二个是部分失败。写 A 成功、写 B 失败这笔消息算生产成功还是失败如果算成功B 缺一条如果算失败A 已经多了一条。要补偿就得自己记日志、自己对账又得引入一整套分布式事务逻辑。第三个是业务代码被污染。双写逻辑一旦写进业务代码消息中间件换掉、复制拓扑调整、新增集群业务代码都要跟着改。数据同步这个本应是基础设施的事牢牢耦合在了业务里。Pulsar 的做法是把这个问题吸到 broker 内部客户端只往本地集群写一份本地确认即返回后台的 Replicator 组件负责把这条消息搬送到远端集群。业务看到的还是一份数据、一份代码复制成了基础设施能力这就是最核心的差别。异步复制不代表会丢消息。Pulsar 的复制是“先落盘、后复制”本地确认的消息一定已经写进本地 BookKeeperReplicator 只是把这部分已确认的消息搬运过去。这一点后面会详细展开。2. 图1—图4先把几件基础事实摆清楚2.1 图1两个集群的复制拓扑该共享的只有数据先看最宏观的拓扑------------------------ ------------------------ | Cluster A | | Cluster B | | | | | | ProducerA ── BrokerA | | BrokerB ── ProducerB | | │ | | │ | | ▼ | | ▼ | | BookKeeperA (ledger) | | BookKeeperB (ledger) | | topic: orders | | topic: orders | ------------------------ ------------------------ │ │ └────────── 复制链路(Replicator) ──────────┘图 1 说清楚了一个最容易混淆的事实复制链路发生在 broker 层不是 bookie 层更不是元数据层。A 集群的 BookKeeper 只存 A 的消息B 集群的 BookKeeper 只存 B 的消息两边各是完整独立的 Pulsar 集群拥有各自的 broker、bookie、元数据服务。我见过不少刚开始接触 Pulsar 的同学以为跨地域复制是底层存储连起来、bookie 互相备份这种理解是错的。跨地域复制做的是逻辑层面的数据同步B 集群以一个消费者的视角从 A 集群接收消息再以一个生产者的身份把消息写进自己的 BookKeeper。整个过程中A 不知道 B 的存储布局B 也不知道 A 的 ledger 结构。还有一个重点这个拓扑里没有第三个“中心协调者”。两个集群是对等的不需要一个额外的组件去管“谁复制给谁”。复制方向完全由命名空间namespace的集群列表配置决定这也是 Pulsar 跨地域复制部署起来比很多自研方案省事的原因之一。2.2 图2一条消息从生产到跨集群消费的完整路径宏观看完看微观上一条消息到底走了哪几步Producer1 → BrokerA.prodHandler → BookKeeperA 落盘 │ (复制游标推进) ▼ ReplicatorA │ 按原顺序读取消息 │ 打包批量发送给B ▼ BrokerB 接收 │ ▼ BookKeeperB 落盘 │ ▼ Consumer2 从B订阅消费这条链路里有三个关键点。第一消息先落 A 的盘再谈复制。生产者调用send()得到确认意味着消息已经写进 A 的 BookKeeper。Replicator 只读已确认的消息不会出现“消息还没确认就被复制过去”的情况。第二对 B 来说它看到的就是一股普通的消息流。Replicator 在 B 侧是以一个持久化生产者的身份出现的B 收到消息后正常落盘、正常触达消费订阅。这意味着消费者在 B 上订阅 topic 时完全不需要知道消息是本地生产的还是跨机房复制过来的接口和本地消费没有差别。第三整个过程是异步的生产者不需要等待跨机房网络往返。这是跨地域复制和同步双写最本质的区别生产者在 A 写入后立刻返回复制到 B 需要多少时间完全不影响生产端的 TPS。代价是 RPO 不为零极端场景会丢最后一段未复制的消息这一点后面专门说。2.3 图3Replicator 不是外部组件它住在 broker 里很多人第一次接触跨地域复制会以为需要单独部署一个类似同步工具的进程。实际上完全不需要BrokerA 内部视图 ------------------------------------------ | topic: persistent://app/orders | | partition-0 ── cursor(本地消费游标) | | partition-0 ── cursor(复制专用游标) | | │ | | ▼ | | Replicator (每个远端集群独立一个) | | [ Reader ] ──→ [ 持久化Producer ] | | │ | ------------┼----------------------------- ▼ 网络 → Cluster BReplicator 是 broker 进程内的一个逻辑组件它由两部分拼成一个Reader负责按复制游标从本集群读消息一个持久化 Producer负责以生产者的身份把消息发给远端 broker。把“复制”拆成读和写两段是我觉得 Pulsar 设计里非常优雅的地方——它复用了消息系统自身的生产消费能力来搬消息而不是单独发明一套新的传输协议。所以排障的时候你完全可以把 Replicator 理解成一个“长驻在 broker 里的客户端程序”。它也是用 Pulsar 的二进制协议跟远端 broker 通信的也会有连接断开、重连、积压这些普通客户端会遇到的问题。复制游标和本地消费游标是分开维护的。本地消费者读到哪里不影响复制进度复制读到哪里也不影响本地消费。这样做的好处是即使本地没有任何消费者复制照样能跑。2.4 图4消息 ID 为什么必须在目标集群重新映射这是很多刚开始用跨地域复制的人会忽略的一个点。Pulsar 的消息 IDMessageId是一个物理指针长这样阶段消息身份本质A 集群本地(ledgerId100, entryId35)指向 A 集群 BookKeeper 的物理存储位置复制到 B 之后(ledgerId5000, entryId12)指向 B 集群 BookKeeper 的物理存储位置业务可见消息内容、publishTime、eventTime 不变消息语义没变物理位置变了原因很简单ledgerId 和 entryId 是 BookKeeper 里的存储地址A 集群的 ledger 100 对 B 集群没有任何意义。B 不知道、也不应该知道 A 的存储布局。消息复制到 B 后B 会把它当作一条普通的新消息来落盘分配属于 B 自己的 ledger 和 entry。由此引出一个实操建议不要把 MessageId 当作业务消息的唯一标识。双活场景下同一条业务消息在生产集群和备份集群里拿到的 MessageId 大概率不一样拿它去数据库做唯一键、去对账很容易踩重复消费的坑。需要唯一标识时优先用消息内容中的业务字段或者发消息时自己生成一个 messageKey。顺序方面可以放心。Replicator 使用单发送链路按游标顺序读取、顺序发送B 端顺次落盘所以单个分区的跨地域复制消息顺序和源集群保持一致。这在很多自研跨集群同步方案里是最难保证的一点Pulsar 天然就做到了。3. 图5—图6双向复制里最容易被绕晕的两个问题3.1 图5“乒乓效应”是怎么被消息血缘标记化解的双向复制A ⇄ B刚接触时所有人都会问同一个问题A 复制给 BB 又复制给 AA 再复制给 B……消息会不会永远在两个集群之间循环也就是“乒乓效应”看这张图消息在 A 被本地生产 │ 消息附带原生产集群标记 originClusterA ▼ Replicator 把消息发送到 B ▼ B 收到消息B 的 Replicator 检查 originCluster ├── originClusterA不是 B │ └── 这条消息不会被 B 复制回 A只提供给 B 本地消费 │ └── 如果 B 本地新生产一条消息originClusterB └── 会被复制到 A但 A 收到后同样不回传Pulsar 给每条消息都记录了一个“血缘标记”即它最初是由哪个集群的原生生产者写入的。这个标记会跟着消息一路复制。当 B 集群的 Replicator 发现消息的原始集群不是 B 自己时就只消费、不转发从根上断掉了循环。这个设计也带来一个隐藏约束消息只能从它的“出生集群”向其他集群传播一层。如果拓扑是 A → B → CA 生产的一条消息复制到 BB 不会继续把它复制到 C要让 C 也拿到这条消息只能把 C 也加入 A 的复制目标列表或者让 B 本地重新生产。所以设计复制拓扑时尽量用星型或对称结构避免链式依赖带来的“复制断层”。Pulsar 客户端 API 里提供了getReplicatedFrom()方法返回 null 表示这条消息是本地生产的返回集群名则表示它来自哪个集群。排查重复消费和环路问题的时候这个字段是第一个要看的信息。3.2 图6消息复制 ≠ 消费进度复制游标到底随不随这是跨地域复制里误解最深的一点。很多人以为A 上消费者读到第 100 条了切到 B 之后 B 上的消费者会自动从第 101 条开始。实际上默认情况下完全不随Cluster A Cluster B topic: orders topic: orders subscription: my-sub subscription: my-sub cursor: 已推进到 msg 50 cursor: 独立维护可能还在 0 消费者 A 读到 50 条 消费者 B 从 0 开始消费为什么因为 subscription cursor 是每个集群自己在 BookKeeper 里维护的本地状态它不参与跨地域复制。消息复制走的是 Replicator 的复制游标和消费者的订阅游标各管各的。这带来的后果很实际A 机房宕机后切到 B 机房如果你直接在 B 上创建一个同名的 subscription消费者默认会从这个 topic 在 B 的第一条消息开始消费——通常不是你期望的“从 A 的进度继续”。想解决这个问题有两个方案。方案一是手动对齐。切换时在 B 上创建订阅后根据上次 A 的消费位置调用seek()指定一个时间点或某个特定 MessageId。优点是实现简单、可控缺点是精度不高时间点不一定对应准确的消息边界。方案二是开启replicated subscription复制订阅。Pulsar 会把订阅游标的变更同步到其他集群原理是 broker 内部借助一个__change_events主题把游标变化以消息形式广播出去。这样 A、B 的订阅名相同、游标不断同步A 宕机后 B 的消费者能从上一次同步的位置继续消费。要注意replicated subscription 不是让你在 A、B 两边同时消费同一个订阅做并行负载均衡的。它的设计初衷是“单个消费组可以在任意一个集群接管”不是“两边同时消费且互不干扰”。如果你真的要在两个机房并行读写、各自消费正确的做法是给两边用不同的订阅名各消费各的。两边消费进度是否需要对齐那是业务层自己决定的事。4. 图7用时间线推演一次真实的机房故障切换4.1 A 集群宕机前后的完整时间线图 7 我画成了一条时间线这样比静态架构图更能说明问题T0: A 正常生产流量打在 A 集群 Replicator 持续把消息复制到 B T1: A 机房发生故障断网/断电/集群崩溃 生产请求开始超时写入不可用 T2: 路由层DNS / LB / 客户端寻址策略 把生产流量切换到 B 集群 T3: 消费者在 B 上创建/复用同一个订阅 若开启 replicated subscription游标自动接续 T4: B 消费者开始消费 最后消费到的消息 ≈ 复制链路在 T1 前搬过来的一批 T5: A 恢复选择把 A 重新纳回拓扑 或让 B 继续作为主集群再同步回去每一步之间都有很多实际细节。T1 的故障不是“啪一下全没了”网络抖动、脑裂、负载飙升都算故障T2 的路由切换也不只是改个 IP生产端要处理大量在途请求客户端重连要设置合理的退避策略T3 的订阅接续是整个切换过程中最容易出错的环节我在实际项目里见过切换后消费延迟积压了几千万条就是因为游标没有同步过来消费者从最早的 offset 开始重放。还要注意 T5A 恢复之后不要急着把流量切回去。A 在宕机时间段内没有收到任何新消息B 期间积累了新消息。如果 A 恢复后自动开启向 B 的复制B 不会把宕机期间的数据反向传给 A除非你手动触发一次双向复制或重新生产。恢复流程需要单独设计不是集群起来了就一切自动回归。4.2 RPO 和 RTO量化的账要算清楚聊容灾方案躲不开 RPO 和 RTO 这两个指标。跨地域复制在这两项上的表现取决于复制链路的状态。RPO 约等于复制链路的延迟。因为 Pulsar 是异步复制A 集群瞬间全挂时最后一段已经生产成功但还没来得及复制到 B 的消息就丢了。这个窗口可能短到几十毫秒也可能长到几秒取决于网络带宽、复制积压程度和消息体大小。想缩小 RPO可以调整复制频次和队列参数但不要幻想 RTO 能做到严格为零——Pulsar 原生跨地域复制没有等待远端确认的机制。RTO 取决于切换链路。路由层切换耗时 客户端重连耗时 消费者在 B 上建立订阅的耗时。开启了 replicated subscription 的情况下消费者连接 B 后可以直接从同步游标继续整个过程通常在秒级到几十秒。相比重新拉起一套服务、回放消息日志的方案这个 RTO 已经非常低了。我见过一些团队为了追求“零 RPO”在业务层把双写逻辑又写了回去。折腾到最后发现真正能兜底的不是零 RPO而是消费端幂等。只要消费者做到“同一条消息重复处理也不产生副作用”那极端情况下丢失或重复几条消息的影响范围就很可控。跨地域复制保证的是系统级的基础能力业务级的一致性最终还是要靠幂等去兜。5. 图8把图画成命令——最小可运行的配置清单5.1 配置项映射图前面所有机制最后都要落到具体的配置上。图 8 把它们串起来Pulsar 集群: cluster-a, cluster-b namespace: my-tenant/my-ns └── set-clusters --clusters cluster-a,cluster-b 作用声明该 namespace 的数据允许复制到哪些集群 topic: persistent://my-tenant/my-ns/orders └── 是否复制由 namespace 配置继承不用逐 topic 配置 └── 若要 replicate subscription: broker.conf → replicatedSubscriptionsEnabledtrue 消费者创建订阅时使用完全相同的订阅名这里有个很容易忽略的点复制的决定单位是 namespace不是单个 topic。只要把 namespace 的集群列表设为 A 和 B那么在这个 namespace 下新建的 topic 会自动进入复制流程不需要对每个 topic 手动开关。这样设计的出发点是让复制策略在业务单元级别统一管理而不是靠运维逐 topic 配置。5.2 最小配置五步走假设你已经有了两个独立运行的 Pulsar 集群输入以下命令前确保两个集群之间网络互通默认端口 6650 和 8080 要放通。第一步注册两个集群pulsar-admin clusters create cluster-a \ --url http://broker-a:8080 \ --broker-url pulsar://broker-a:6650 pulsar-admin clusters create cluster-b \ --url http://broker-b:8080 \ --broker-url pulsar://broker-b:6650第二步创建租户和命名空间并把集群列表绑定好pulsar-admin tenants create my-tenant --admin-roles my-admin pulsar-admin namespaces create my-tenant/my-ns pulsar-admin namespaces set-clusters my-tenant/my-ns \ --clusters cluster-a,cluster-b第三步在 A 集群创建 topic 并生产消息pulsar-admin topics create persistent://my-tenant/my-ns/orders第四步在 B 集群上验证消息是否到达。可以用命令行消费也可以在客户端里订阅同名 topicpulsar-client consume persistent://my-tenant/my-ns/orders \ --subscription-name check-sub --num-messages 10第五步开启 replicated subscription如需要在集群两端 broker.conf 里设置replicatedSubscriptionsEnabledtrue重启 broker 后客户端创建订阅时使用相同订阅名即可实现游标同步。这五步跑通之后整个跨地域复制的骨架就搭起来了。注意--admin-roles、--subscription-name这些参数按你自己环境调整具体字段在不同版本里可能有细微差异以官方文档为准。5.3 参数调优与复制延迟控制跑通之后接下来就是调延迟了。跨地域复制的几个常用参数参数作用调整思路replicationPolicyCheckDurationSeconds隔多久检查一次 namespace 复制策略调小可让新 topic 更快进入复制但会增加元数据检查频率replicationProducerQueueSize复制生产者发送队列大小调大有利于批量发送、提高吞吐但会略微抬高单条延迟replicationConsumerQueueSize复制消费者读取队列大小影响从本地 BookKeeper 读消息的批大小通常跟着上一条调replicationStartInBackOff复制启动阶段的重试退避网络不稳定时避免频繁重连但调太大延迟会更明显网络带宽永远是跨地域复制的第一瓶颈。带宽不足时复制积压会持续累积RPO 随之恶化。监控时除了看消息 TPS一定要看复制积压量topic stats 里的replicationBacklog。积压持续上涨说明复制速度跟不上生产速度要么加带宽要么需要检查消息体大小是否合理。另外强烈建议两端集群的消息保留策略和 TTL 保持一致。我踩过的一个坑是 A 集群消息保留 7 天、B 集群只保留 1 天结果业务切到 B 后发现 1 天前的消息全没了。跨地域复制只保证“复制那一刻数据一致”不保证“后续生命周期一致”保留策略这类配置必须人工对齐。6. 我踩过的几个跨地域复制坑经验台账6.1 命名空间集群列表没配全复制静默失败最早一次配置时A 和 B 的 topic 都建了但 B 上怎么等都等不到消息。排查了一圈网络、权限都没问题最后用一行命令发现问题pulsar-admin namespaces get-clusters my-tenant/my-ns返回只有cluster-a。也就是说 namespace 的集群列表压根没把 B 加进去topic 自以为在复制实际 Replicator 根本找不到远端目标。这类问题在界面上很难看出来因为 topic 在两端都存在只是互不相通。所以每配置好一个 namespace第一件事就是get-clusters确认集群列表。还有一个隐蔽的变体两个集群名称大小写不一致或者 broker URL 填了内网地址而另一端解析不到。复制链路是长连接并不会像普通请求那样立刻报错而是持续重试表现为“复制延迟越来越大、消息一直过不去”。遇到这种问题看 broker 日志里关于 replication 的告警是最快的定位手段。6.2 双向复制下消费端重复消费双向复制本身没有问题问题出在消费者配置上。我在双活项目里遇到过一件事A 和 B 两端各有一组消费者为了“逻辑统一”用了相同的订阅名。A 端消费者读到 100 条B 端消费者也读到 100 条两边各自推进游标。由于普通订阅的游标不复制两边很快就各走各的。某次 A 发布变更B 端的自动重平衡把消费位置跳了回去于是 B 端开始大量重放 A 端已经消费过的消息。解决办法有两个分支。第一如果想真正并行消费两边数据给两个机房分别用不同的订阅名各自管理各自游标。第二如果希望在故障切换时由一个订阅组无缝接管开启 replicated subscription但保证同一时刻只有一个集群在消费这个订阅。两个目标不能混在一起这是我给出的最核心的实践结论。6.3 版本差异和网络分区带来的“伪脑裂”Pulsar 客户端和服务端跨版本兼容性做得不错但 broker 与 broker 之间的复制链路对版本差异会更敏感一些。一次测试环境中两个集群版本差了三个大版本复制生产者握手时行为不一致频繁断连重连复制吞吐一直上不去。升级到相同版本后问题消失。所以生产环境强烈建议所有参与跨地域复制的集群保持同一版本至少也不要差出大版本。网络分区时Replicator 会暂停发送但不会丢消息。网络恢复后积压的消息会迅速补发这个瞬间对 B 集群的写入压力会有一个明显峰值。要注意监控 BookKeeper 的写入延迟提前预留一定余量不然恢复期可能把 B 集群写入拖垮。6.4 最后一点体会跨地域复制不是银弹。它把“一份消息从 A 搬到 B”这件事变成了 Pulsar 的基础能力让应用层少写很多分布式逻辑但它不解决应用层的幂等、不解决业务的双活语义、也不保证绝对的零丢失。如果你想在生产环境上双活我的建议是排个优先级先把消息幂等消费做好再启用 replicated subscription 做消费切换最后才考虑自动路由。先让数据能复制再让进度能接管最后才是优雅切换。这个顺序我走了不少弯路才验证出来希望你能少踩一轮。