Pulsar消息路由：分区选择、Bundle归属与Key_Shared分发

消息在 Pulsar 里从来不是"直接扔进一个队列"那么简单。我印象最深的一次排查是某个订单类 Topic 的消费延迟曲线：二十多个分区里，只有两个分区的 backlog 持续上涨，其余分区的消费进度一路平稳。一开始怀疑是消费者数量不够、线程池打满、broker 磁盘抖动，折腾了大半天，最后定位到问题出在路由上——生产端用了一个分布极不均匀的 key，那两个分区扛了将近六成的流量。这件事之后我把 Pulsar 的消息路由从头到尾梳理了一遍，才发现"路由"这个词在这套系统里其实指三条完全不同的链路，混在一起谈，排查方向必然跑偏。

这篇内容讲的是 Pulsar 消息路由的完整机制：一条消息从send()调用开始，怎么被分配到某个分区，怎么找到服务这个分区的 Broker，又怎么在消费端落到某个具体的消费者实例。适合已经在用 Pulsar、但对内部机制还停留在"配个 Topic 就能发"这个层次的开发者，也适合正在做分区数规划、顺序性保障、热点治理的团队参考。下面所有内容都围绕这三条链路展开，涉及参数的地方我会把取值理由和踩过的坑一并写进去。

1. 三条互相独立的路由链路，先分清才不会查错方向

很多人第一次听到"Pulsar 消息路由"，脑子里浮现的是一个函数：输入消息，输出目标地址。实际上在 Pulsar 里至少有三种不同粒度的路由在同时发生，它们由不同的组件执行、在不同的时机触发、出问题时表现出的症状也完全不一样。分不清这三层，排查起来就是在三个方向上瞎撞。

1.1 写入侧：Producer 选择分区的那个函数

第一层路由发生在客户端。当 Topic 是分区 Topic（Partitioned Topic）时，它在逻辑上是一个名字，物理上是 N 个真实的 Topic，形如persistent://tenant/ns/order-events-partition-3。生产者在创建时需要知道这个逻辑 Topic 有几个分区，这个信息来自一次元数据查询（CommandPartitionedTopicMetadata），拿到分区数后，客户端会为每个分区建立一个内部的子生产者，真正的消息发送走的是子生产者。

那么一条消息到底交给哪个子生产者？这就是MessageRouter接口干的活。它的方法签名很朴素，输入是消息和 Topic 元数据，输出是一个 int 分区下标：

public interface MessageRouter extends Serializable { int choosePartition(Message<?> msg, TopicMetadata metadata); }

这个设计的好处是把决策权完全下放到客户端，Broker 不需要关心消息为什么落到这个分区。代价是路由逻辑的正确性和均匀性完全由客户端保证——如果自定义路由写歪了，Broker 侧看不到任何异常，只会看到某个分区的写入速率异常高。这也是我开头那次排查花了很久的原因：服务端指标一切正常，问题在客户端的一个 lambda 里。

还有一点容易被忽略：MessageRouter是Serializable的。生产者实例在某些场景下需要被序列化传递（比如在函数计算环境里），如果自定义路由里持有不可序列化的成员（比如一个数据库连接或者 Spring 的 Bean），运行时才会炸，编译期毫无提示。

1.2 服务侧：Topic 归属哪台 Broker 由 Bundle 说了算

第二层路由发生在 Broker 集群内部，决定的是"这个分区的数据由哪台机器负责读写"。Pulsar 用的是分片归属模型：每个 namespace 被划分成若干个Bundle，每个 Bundle 是一个哈希区间，比如[0x00000000, 0x3FFFFFFF]。一个 Topic 名字经过哈希后落在哪个区间，就归属哪个 Bundle，而每个 Bundle 在任意时刻只有一个 owner Broker。

这套机制的直接后果是：Pulsar 里的 Topic 不像某些系统那样需要人工分配，它是自动归属的；但反过来，归属关系会因为负载均衡而漂移。当某个 Broker 上线、下线，或者负载均衡器判定需要迁移时，Bundle 会被 unload 到另一台机器，客户端会收到关闭信号并重新做一次 lookup。对于生产者来说，表现为一次短暂的重连；对于消费者来说，表现为一次 rebalance。理解这一点，才能在看到"producer 频繁重连"的告警时判断是网络问题还是归属漂移。

Bundle 的数量和切分策略直接影响归属的粒度。默认一个 namespace 的 Bundle 数量有限（常见配置是 4），意味着如果这个 namespace 下有几百个分区 Topic，多个分区的归属会被绑在一起——一个 Bundle 迁移，会连带它下面的所有 Topic 一起搬家。这是个很隐蔽的耦合，后面第 4 节会专门展开。

1.3 消费侧：Key_Shared 订阅把 Key 映射到消费者

第三层路由发生在消费端，只有在使用Key_Shared订阅类型时才存在。此时 Broker 需要决定一条消息交给订阅组里的哪个消费者实例，依据是消息的 key。这个决策和写入侧的分区选择是两套独立的哈希逻辑，不能混为一谈。

很多团队一开始会有一个幻觉："我给消息带了 key，生产端按 key 落到固定分区，消费端也按 key 分给固定消费者，所以端到端都是有序且均衡的。"前半句成立，后半句要看订阅类型。如果是Shared订阅，Broker 是轮询分发的，同一个 key 的消息完全可能被两个消费者并行处理；只有Key_Shared（或Failover）才会按 key 绑定。而Key_Shared内部用的哈希槽模型和分区路由的 Murmur3 哈希并不是同一个映射空间，所以"同 key 同分区"和"同 key 同消费者"是两条独立的保证，需要分别验证。

把这三层拆开之后，排查思路就清晰了：单分区流量高，查第一层；producer 抖动、lookup 失败，查第二层；消费端负载不均但生产端均匀，查第三层。

2. MessageRouter 的实现细节与自定义接入

第一层路由是大多数业务能直接控制的唯一一层，也是最容易改出问题的一层。Pulsar 内置了三套实现，加上自定义路由一共四条路，每条路的行为差异比名字听起来大得多。

2.1 三种内置路由模式的真实行为差异

配置入口是messageRoutingMode，取值有RoundRobinPartition、SinglePartition、CustomPartition。光看名字容易误解，我把实际行为整理成一张表：

这里有个反直觉的点：RoundRobinPartition在消息带 key 的时候，根本不轮询。它内部会先判断msg.hasKey()，有 key 就直接算哈希。所以"我用了轮询模式所以消息是分散的"这个假设，在业务给消息设了 key 之后就不成立了。我见过一个案例，业务方为了做幂等给每条消息都塞了一个业务流水号作为 key，结果所有消息都被这个几乎唯一的 key 打散到了所有分区——看似均匀，但同一个业务实体的消息彻底失去了顺序保证。

反过来，SinglePartition也不是"所有消息进一个分区"。不带 key 时它确实只挑一个分区（每个生产者实例随机选一个，不同实例选的可能不同），带 key 时依然按哈希走。所以在多生产者实例的场景下，指望SinglePartition实现全局有序是要打问号的：三个生产者实例可能选了三个不同分区。

还有一个细节值得说：RoundRobinPartition在开启 batching 时会受一个"分区切换频率"参数影响。默认情况下，客户端会在一个 batching 窗口内把消息尽量分散到不同分区，提升并行度，但这意味着同一个窗口内的多条消息可能进不同分区。如果你的业务依赖"连续两条消息的顺序"，batching 和轮询组合起来是会破坏这个假设的。要么关 batching，要么用 key，要么显式指定单分区。

2.2 自定义 MessageRouter 的接入姿势与序列化陷阱

当内置模式满足不了需求时，就走CustomPartition。典型的自定义场景有两类：一是业务上有明确的"租户隔离"需求，希望某几个大客户的流量固定落在专属分区上，避免和其他租户抢资源；二是做灰度，把特定标记的消息引导到新扩容的分区上观察。

接入代码大致长这样：

public class TenantAwareRouter implements MessageRouter { private static final long serialVersionUID = 1L; private final int dedicatedPartitions; public TenantAwareRouter(int dedicatedPartitions) { this.dedicatedPartitions = dedicatedPartitions; } @Override public int choosePartition(Message<?> msg, TopicMetadata metadata) { int total = metadata.numPartitions(); byte[] key = msg.getKeyBytes(); if (key == null || key.length == 0) { return ThreadLocalRandom.current().nextInt(total); } int raw = Math.floorMod( Murmur3_32Hash.getInstance().makeHash(key), total); // 大租户固定映射到前 N 个分区 if (isVipTenant(key) && raw >= dedicatedPartitions) { return raw % dedicatedPartitions; } return raw; } }

要注意的坑有几个。第一，MessageRouter必须可序列化，上面这个类只持有基本类型字段所以没问题，但如果你在里面注入了配置中心客户端或者缓存对象，务必用transient修饰或者干脆改成静态持有。第二，choosePartition的返回值必须落在[0, numPartitions)区间内，返回越界值会直接抛异常，而Math.abs(hash % n)在哈希为Integer.MIN_VALUE时会返回负数——这个经典 bug 我至少在两份生产代码里见过，改用Math.floorMod就稳了。第三，自定义路由一旦上线就很难改：改路由函数等于改变了历史消息的归属规则，已经在某个分区里的数据不会跟着搬。所以设计时最好预留一个"版本位"，比如在 key 前缀里带上路由版本号，将来切换时可以双写过渡。

2.3 路由函数的幂等性与并发安全

choosePartition会被多个发送线程并发调用，这一点从它的签名看不出任何提示。内置的轮询实现用的是AtomicInteger自增，天然线程安全；自定义实现如果用了普通成员变量做计数，就会出现计数错乱，表现为分区分布莫名其妙地偏斜，而且复现概率随并发度升高而升高，本地压测很难触发。

我的建议是：路由函数里只做纯计算，不维护任何可变状态。确实需要计数的场景（比如加权轮询），用AtomicLong或者ThreadLocal加时间片，别用int。另外，哈希计算本身要注意稳定性——不要用key.hashCode()之后直接取模，因为 Java 的String.hashCode()在不同 JDK 版本上虽然规则固定，但你拿到的key是字节数组，从byte[]到String的编码转换一旦有字符集差异（比如默认编码在容器环境里是 POSIX 而不是 UTF-8），哈希结果就会变，导致路由漂移。

最后补一句：路由函数的"幂等"指的是同一个 key 必须永远返回同一个分区。这在单分区数固定时是成立的，但分区数一变，所有 key 的归属都会重排。这是分区扩容最大的隐性成本，下一节接着聊。

3. 分区数、Key 设计与顺序性的三角关系

分区数怎么定，是我被问得最多的问题，也是返工成本最高的决策。定少了后面扩不动，定多了管理开销和顺序性都会受影响。这里面没有标准答案，但有几个必须避开的错误算法。

3.1 分区数估算的常见错误算法

最流行也最坑的算法是"目标吞吐 ÷ 单分区吞吐"。比如目标 50 万条每秒，测出来单分区能到 2 万条每秒，于是拍板 25 个分区。这个算法的问题在于它假设了流量在分区之间完全均匀，而现实中不均匀才是常态——只要业务消息带了 key，分布就取决于 key 的基数与哈希质量。

更靠谱的做法是做一次真实 key 分布的采样：拿生产环境一天的消息样本，按你打算用的 key 做哈希，统计落到每个分区的消息量，看最大值与平均值的比值。如果这个比值超过 1.5，说明你的 key 设计有问题，先把 key 改好再谈分区数；如果比值在 1.1 到 1.3 之间，那就在理论值上再留 30% 的余量。

第二个常见错误是忽略分区数对订阅端的影响。分区数等于这个 Topic 并行消费的上限——一个分区在同一时刻只能被同一个订阅组里的一个消费者消费。如果分区数是 8，你就算起 32 个消费者实例，也有 24 个在空转。这在扩容时会造成很尴尬的局面：消费跟不上，加机器没用，只能扩分区，而扩分区又会触发路由重排。

我的经验公式大致是：分区数 = max(峰值写入吞吐 ÷ 单分区实测吞吐 × 1.3, 峰值消费吞吐 ÷ 单分区消费吞吐 × 1.3)，然后再向上取整到一个便于管理的数字。这个数字一旦确定，短期内不要动。

3.2 带 Key 与不带 Key 是两套完全不同的路由

这一点我在第 2 节提过，但值得单独展开，因为它决定了很多业务的行为边界。

不带 key 的消息，在RoundRobinPartition下是真正轮询的，分布天然均匀，但代价是没有任何顺序保证——同一个业务实体的两条消息可能进不同分区，被不同消费者并行处理，处理顺序完全不可控。这在日志采集、指标上报这类场景完全没问题，但在订单状态变更、账户流水这类场景是灾难。

带 key 的消息，路由由哈希决定，同一个 key 永远落同一个分区，顺序性有保证。但均匀性就得看 key 的基数。这里有个反直觉的结论：key 基数越高不代表越均匀。如果 key 是自增 ID 且低位被业务截断（比如订单号末两位是渠道编码，只有几种取值），哈希之后照样会挤在少数几个分区上。我见过一个业务用"用户 ID 后四位"做 key，恰好这批用户 ID 是批量注册的连号，哈希结果高度聚集，两个分区扛了七成流量。

判断 key 质量最简单的办法是拿真实数据跑一遍哈希，画出分布直方图。别靠直觉，直觉在这个问题上几乎总是错的。

3.3 分区扩容后路由漂移的真实影响

分区扩容不是"加几个分区然后流量自动均衡"这么简单。从 N 个分区扩到 M 个，所有 key 的哈希取模结果都变了，绝大部分 key 的归属分区会发生变化。这意味着：

历史消息还留在老分区里，新消息按新规则进新分区，同一条业务线的消息在时间轴上被人为切开。如果消费者是幂等处理、只看最终状态，问题不大；如果消费者依赖"同 key 消息的严格时序"，扩容期间必然出现乱序。

扩容还会触发一次元数据变更，所有在线生产者重新拉取分区数、重建子生产者，表现为一批短连接和瞬时错误率上升。稳妥的做法是在低峰期操作，并且提前把生产者配置里的重试策略调好，让它能扛住这个窗口。

如果业务绝对不能接受乱序，有两套方案：一是给消息加版本号或序列号，消费端做缓冲重排；二是引入"双写过渡期"，先按新老两套路由同时投递一段时间，消费端去重后再切换。第一种成本低但有延迟代价，第二种成本高但业务无感，看你的容忍度。

4. Broker 侧路由：Lookup、Bundle 与所有权漂移

前面讲的是"消息去哪个分区"，这一节讲"哪个 Broker 服务这个分区"。这层路由业务方一般不直接接触，但它决定了系统的稳定性和可观测性。

4.1 一次 send 之前的 Lookup 往返

生产者第一次往某个分区发消息之前，必须先做一次 lookup。客户端向任意一台 Broker 发CommandLookupTopic，Broker 计算这个 Topic 的哈希落在哪个 Bundle，查本地缓存（或元数据存储）找到该 Bundle 的 owner。如果 owner 就是自己，直接返回服务地址；如果不是，返回一个重定向指令，带上目标 Broker 的地址，客户端再去目标 Broker 完成握手。

这个流程的结果会被客户端缓存，后续发送不再重复 lookup。缓存失效的触发点有几个：连接断开、收到CommandCloseProducer、元数据版本变更（比如分区数调整）、以及客户端的定期刷新。任何一个触发点都会导致一次新的 lookup 往返，表现为短暂的发送失败或延迟尖峰。

排查这类问题最直接的命令是看 owner 归属：

bin/pulsar-admin topics lookup \ persistent://my-tenant/my-ns/order-events-partition-3

输出会告诉你当前是哪台 Broker 在服务这个分区。如果这个结果和你从监控上看到的连接来源不一致，说明客户端缓存的归属信息过期了，通常是上一次迁移后客户端没有及时刷新。

4.2 Bundle 切分与负载均衡器的连带效应

Bundle 是归属的最小单位。一个 namespace 默认只有少量 Bundle，所有 Topic 按名字哈希落到这些区间里。如果某个 namespace 下有几百个分区 Topic，而 Bundle 只有 4 个，那平均每个 Bundle 要承载上百个分区。这时候负载均衡器看到的粒度就是 Bundle——它觉得某个 Bundle 所在的 Broker 负载高，就整个 Bundle 搬走，连带里面上百个分区同时重连。

处理办法是提前做 Bundle 切分，让归属粒度更细：

bin/pulsar-admin namespaces split-bundle \ --bundle 0x00000000_0x3FFFFFFF \ my-tenant/my-ns

切分的时机要提前，别等出问题再切——切分本身也是一次元数据变更，会引起归属重算和客户端重连。我的做法是在 namespace 创建时就按预计的 Topic 数量一次性把 Bundle 切到位，后续只做微调。

另外要注意 Bundle 和分区数的关系不会自动优化。扩了分区但没扩 Bundle，等于把更多的东西塞进了同样的几个盒子里，负载均衡的精度反而更差了。

4.3 所有权变更时客户端看到的现象

Bundle 迁移对客户端来说是个"半透明"事件。生产者侧会收到关闭指令，SDK 内部自动重连，如果重连失败会按退避策略重试，重试期间发送的消息要么阻塞在队列里，要么直接抛异常，取决于blockIfQueueFull和发送超时的配置。消费者侧则是一次 rebalance，在Key_Shared订阅下还会触发 key 的重新分配。

这里有个很容易踩的坑：迁移期间如果生产者的发送队列设得太小（maxPendingMessages默认 1000），又开了blockIfQueueFull=true，那么应用线程会被直接阻塞，表现出来是接口 RT 飙升，但日志里除了重连警告什么都没有。我第一次遇到时排查了很久，最后才发现是队列满了在阻塞。

调优方向是：把maxPendingMessages按峰值 QPS 乘以预计的重连窗口来估，比如峰值 5000 QPS、重连窗口 2 秒，那至少要 10000 的缓冲。同时把sendTimeout设成一个业务能接受的值，让实在发不出去的消息快速失败而不是无限堆积。

5. 消费端的 Key 分发与再平衡

第三层路由只在使用Key_Shared订阅时存在，但它对业务的影响往往比前两层更直接——因为它决定了消息在消费者实例之间怎么分。

5.1 Key_Shared 的 hash 槽分配模型

Key_Shared的默认策略是自动切分哈希范围（autoSplitHashRange）。整个 key 空间被视为[0, 65535]的 65536 个槽位，Broker 用 Murmur3 对消息 key 做哈希后映射到某个槽，每个消费者实例负责一段连续的槽。收到消息时，Broker 只需要判断这个槽在谁的区间内，就投给谁。

这个模型的优点是稳定：只要消费者列表不变，同一个 key 永远去同一个实例，顺序性得到保证。缺点是粒度受消费者数量限制——16 个消费者分 65536 个槽，每个拿 4096 个槽，看起来还行；但如果某个 key 对应的槽恰好落在一个消费能力弱的实例上，这个实例就会成为瓶颈，而且你没法通过加消费者来缓解（加消费者会触发重新分槽，反而引发全局抖动）。

还有一种少见但确实存在的策略是指定哈希范围（hashRange），适用于你明确知道某些 key 应该由特定实例处理的场景，比如按地域分流或者按租户隔离。配置时要保证各实例的区间不重叠且覆盖完整，否则会出现"某些 key 无人认领"导致消息堆积。

5.2 消费者扩缩容时的 Key 迁移

消费者实例数量一变，哈希槽的分配就得重来，这个过程叫 rebalance。Pulsar 的Key_Shared在 rebalance 时会尽量缩短不可用窗口：新实例加入后先接管一部分槽，老实例把对应位置的未确认消息处理完再释放。但"尽量缩短"不等于"零影响"，在槽迁移的瞬间，某些 key 的消息会被短暂挂起。

影响的程度取决于两个因素：消费者的确认模式，以及未确认消息的数量。如果用的是单条确认（AcknowledgementMode.Individual），迁移时需要等待的在途消息少，窗口短；如果用累积确认或者批量确认，在途消息多，窗口就长。

我的经验是给Key_Shared的消费者配KeySharedPolicy时把allowOutOfOrderDelivery显式设为false（默认值），并且把receiverQueueSize调小一些，比如从默认的 1000 降到 200 左右。队列小意味着 rebalance 时要重投的消息少，恢复更快。代价是吞吐会略微下降，但对顺序敏感的业务这个交换是值得的。

5.3 消息积压在单个消费者的定位手法

症状是：整体积压不高，但订阅组里总有一个实例的处理延迟明显高于其他。这时候按下面的顺序查：

先确认是不是 key 倾斜。用topics stats看这个订阅的分区级消费速率，如果所有分区的消费速率差不多，但消费者实例之间差异大，基本可以确定是 key 分布问题。再看是不是分区级倾斜——如果只有一两个分区的消费慢，那就是写入侧路由的问题，回到第 2 节去查 key 设计。

然后是槽位分配的偶然性。消费者数量少的时候（比如只有 3 个），槽位分配是粗粒度的，本来就可能出现明显的量差。这时候临时加一两个消费者观察一下分布是否改善，能快速区分"结构性倾斜"和"分配偶然性"。

最后别忘了检查单个实例的处理能力差异。同一个订阅组里如果有实例跑在配置不同的机器上，或者某个实例内部有线程池竞争，也会表现出类似症状。这时候看的是 CPU 和线程栈，不是消息指标。

6. 几类线上高频故障的排查路径

把前面五节的内容串起来，这里给几条可以直接照着走的排查链路。这些都是我在实际环境里反复用过的顺序，不是理论推演。

6.1 单分区积压但整体流量正常的排查顺序

这是最典型的写入侧路由问题，按这个顺序走：

第一步，确认积压是集中在单个分区还是分散。用topics stats-internal能看到每个分区的 backlog 明细。如果只有一个分区涨，先把它的 owner 和写入速率拉出来。

第二步，看这个分区的消息 key 分布。如果日志里能采到样本，抽样 1 万条做哈希统计，看是不是某个高频 key 砸在了这里。这一步能解决绝大部分案例。

第三步，排除人为因素。检查是不是有生产者实例配置了SinglePartition模式，或者自定义路由里对某类消息做了固定映射。我遇到过一版代码把"补偿消息"全部路由到分区 0 的处理，结果补偿任务一跑，分区 0 立刻堵死。

第四步，才考虑资源问题。单个分区的写入受限于它所在的 Broker 的磁盘和网络，如果这个 Broker 上还扛着别的重负载 Topic，也会表现出单分区慢。这时候用topics lookup确认归属，再看那台机器的整体指标。

6.2 消息顺序错乱的几种真凶

顺序问题一旦出现，往往已经造成了业务损失，所以定位要快。我把见过的情况按出现频率排了个序：

排在第一位的是订阅类型用错。业务以为用了Key_Shared就是有序的，实际配的是Shared。这个错误极其常见，尤其在一些封装过的客户端里，订阅类型是框架层的默认值，业务代码根本没显式设置。

第二位是 key 不一致。生产者发消息时，有的路径带了 key，有的没带。不带 key 的消息走轮询，自然就乱了。这种情况在重构后特别容易发生——新增了一个发送入口，忘了设置 key。

第三位是 batching 与重试的组合。开启 batching 后，一批消息在客户端被打包，如果发送失败重试，整批重发，可能导致同一批内的消息相对顺序在消费端看起来变了。更隐蔽的是消息在 Broker 侧的重投，未确认消息在连接断开后会重新投递，如果前后两批消息落在不同的消费实例上，顺序就彻底乱了。

第四位才是扩容导致的路由漂移，前面 3.3 节已经说过，这种情况属于预期内的行为，关键在于提前规划而不是事后补救。

排查的时候，最快的办法是给每条消息打上生产时间戳和序列号，消费端记录下来，然后找出错序的那一对消息，看它们的 key 是否相同、分区是否相同、消费者实例是否相同。三个都相同还乱序，那才需要往 Broker 内部深挖。

6.3 一份可以直接抄的参数清单

最后把关键参数整理成一张表，都是我在生产环境验证过的取值方向，具体数字要根据自己的压测结果调整。

有一点必须强调：这些参数之间存在耦合，单独调一个往往会引出新问题。比如把maxPendingMessages调大解决了迁移时的阻塞，但同时会增加内存占用和重连后的重发量；把receiverQueueSize调小加快了 rebalance，但会降低吞吐。每次只改一个，压测一轮再改下一个，这是我唯一推荐的做法。

我个人的体会是，Pulsar 的路由机制设计得相当克制——客户端负责分区选择，元数据层负责归属，Broker 负责分发，三层各管一段，互不越界。这种清晰的分工带来的好处是每一层都可以独立优化，坏处是出了问题需要你自己去判断到底是哪一层。前面那次分区积压的排查，最后改动的其实只是业务代码里一个 key 的生成规则，一行改动解决了问题，但找到这一行花掉的时间，全部花在排除另外两层上。把三层链路在脑子里画清楚，这种排查就能从半天缩短到十几分钟。