消息队列流处理后端微服务消息路由【免费下载链接】pulsarApache Pulsar - distributed pub-sub messaging system项目地址https://gitcode.com/gh_mirrors/pu/pulsar点击查看免费下载导读本文基于 Apache Pulsar 仓库中的设计文档 PIP-392系统讲解如何在 Failover 订阅模式下为**分区主题Partitioned Topic**启用一致哈希算法来选择 Active Consumer。读完本文你将理解传统partitionedIndex % consumerSize取模算法在少量分区场景下的负载不均问题、一致哈希的完整实现步骤哈希环构建 消费者选择、新增配置项activeConsumerFailoverConsistentHashing的启用方式与兼容性影响并掌握对应的 broker 配置 与 底层分发器源码 的对应关系。背景知识从非分区主题的一致哈希说起Pulsar 社区在 [PR #19502] 中为**非分区主题Non-Partitioned Topic**引入了基于一致哈希的 Active Consumer 选择机制即通过消费者名称在哈希环上的分布来决定由哪一个消费者作为当前活跃消费者。PIP-392 的目标是把这套已经验证过的算法能力延伸应用到分区主题上并提供一个显式的开关供用户控制。动机取模算法在分区主题上的负载不均问题在 PIP-392 之前分区主题的 Active Consumer 选择使用的是取模公式partitionedIndex % consumerSize即在 AbstractDispatcherSingleActiveConsumer.java 中用分区的序号对消费者数量取模得到消费者下标。这个方法的缺陷在于当分区数量很少尤其是单分区而消费者数量很多时大量分区会被均匀取模却集中落在同一个消费者身上造成严重的负载倾斜。PIP-392 文档给出了一个非常典型的问题场景假设有 100 个主题命名为public/default/topic-{0~100}每个主题只有 1 个分区one partition。用regex订阅 Failover模式创建 10 个消费者后由于每个主题都只有 1 个分区partitionIndex恒为 00 % 10 0因此所有主题的 Active Consumer 都是第一个连接的消费者其余 9 个消费者完全空转。这在单分区主题 正则订阅 Failover的组合下是常见现象分区索引无法参与差异化选择取模算法退化为固定选择 0 号消费者。目标与边界In Scope本次设计解决的范围解决Failover 订阅类型在单分区或少数分区主题上的消费者分配不均问题。通过一致哈希让消费者名称和主题名称共同决定分区归属使大量主题在多个消费者之间更均匀地分布。Out of Scope本次设计明确排除的范围Exclusive独占订阅类型不在本次改动范围内保持一致行为。已知的副作用消息重复投递需要特别说明无论是取模算法还是一致哈希算法在消费者集合发生变化消费者加入、退出时都可能触发 Active Consumer 转移从而导致消息被重复投递给消费者。这是 Pulsar Failover 订阅模式的已知特性官方文档Failover 订阅说明已明确提示Failover 模式下消息可能被投递多次消费者需具备幂等处理能力。高层设计一致哈希算法的两个核心步骤PIP-392 复用了 [PR #19502] 中已实现的一致哈希算法。算法整体分两步步骤一构建哈希环Hash Ring Creation遍历所有消费者以消费者名称 虚拟节点序号作为 key计算哈希值并放入一个有序的TreeMap中为每个消费者生成100 个虚拟节点CONSUMER_CONSISTENT_HASH_REPLICAS 100虚拟节点的作用是让哈希环上的分布更加均匀避免真实节点过少导致的聚集效应。对应仓库源码位于 AbstractDispatcherSingleActiveConsumer.javaprivate NavigableMapInteger, Integer makeHashRing(int consumerSize) { NavigableMapInteger, Integer hashRing new TreeMap(); for (int i 0; i consumerSize; i) { for (int j 0; j CONSUMER_CONSISTENT_HASH_REPLICAS; j) { String key consumers.get(i).consumerName() j; int hash Murmur3_32Hash.getInstance().makeHash(key.getBytes()); hashRing.put(hash, i); } } return Collections.unmodifiableNavigableMap(hashRing); }要点说明每个消费者的虚拟节点 key 形如consumerName0、consumerName1、……、consumerName99哈希算法使用Murmur3_32HashMurmurHash3 的 32 位变体key 为该字符串的字节序列哈希环用TreeMapInteger, Integer承载key 为哈希值value 为消费者在consumers列表中的下标天然有序支持后续的取上界查询返回的哈希环被包装为不可修改视图Collections.unmodifiableNavigableMap防止外部篡改。步骤二用主题名哈希选择消费者Consumer Selection用**主题名topicName**计算哈希值在哈希环上找到第一个大于等于该哈希值的节点ceilingEntry该节点对应的消费者即被选中若主题名哈希值大于环上所有节点则回退到环上第一个节点firstEntry。对应源码位于 AbstractDispatcherSingleActiveConsumer.javaprivate int peekConsumerIndexFromHashRing(NavigableMapInteger, Integer hashRing) { int hash Murmur3Hash32.getInstance().makeHash(topicName); Map.EntryInteger, Integer ceilingEntry hashRing.ceilingEntry(hash); return ceilingEntry ! null ? ceilingEntry.getValue() : hashRing.firstEntry().getValue(); }这种主题名 → 哈希 → 环上最近节点的映射保证了相同主题始终被映射到相同消费者确定性同时不同主题的哈希值在环上随机分布从而在大量主题的场景下实现统计意义上的均匀分配改善负载均衡与资源利用率。详细设计与实现配置项与选择逻辑实现思路实现本身非常简洁只要activeConsumerFailoverConsistentHashing开关被启用无论主题是否分区一律使用一致哈希算法选择消费者开关未启用时分区主题沿用取模算法非分区主题沿用原有逻辑。在 AbstractDispatcherSingleActiveConsumer.java 的pickAndScheduleActiveConsumer()方法中选择逻辑为int consumersSize consumers.size(); // 若存在不同优先级的消费者只在高优先级消费者之间分配hasPriorityConsumer 相关逻辑 ... int index partitionIndex 0 !serviceConfig.isActiveConsumerFailoverConsistentHashing() ? partitionIndex % consumersSize : peekConsumerIndexFromHashRing(makeHashRing(consumersSize)); Consumer selectedConsumer consumers.get(index);从源码可以看出选择逻辑的三层结构先按**优先级priority level**排序仅在高优先级消费者之间做分配hasPriorityConsumer分支会把consumersSize截断为最高优先级消费者的数量若开关未开启且partitionIndex 0即分区主题走partitionIndex % consumersSize取模逻辑其余情况非分区主题或开关已开启的分区主题走peekConsumerIndexFromHashRing(makeHashRing(consumersSize))一致哈希逻辑。新增配置项配置字段定义位于 ServiceConfiguration.javaFieldContext( category CATEGORY_POLICIES, doc Enable consistent hashing for selecting the active consumer in partitioned topics with Failover subscription type. For non-partitioned topics, consistent hashing is used by default. ) private boolean activeConsumerFailoverConsistentHashing false;参数说明属性值配置名activeConsumerFailoverConsistentHashing类型boolean默认值false保持原有取模行为分类CATEGORY_POLICIES策略类配置生效方式通过serviceConfig.isActiveConsumerFailoverConsistentHashing()在每次选择 Active Consumer 时读取配置文件中的对应项该配置项已同步出现在两处标准配置文件中可直接通过注释与示例值对照使用broker.conf# Enable consistent hashing for selecting the active consumer in partitioned topics with Failover subscription type. # For non-partitioned topics, consistent hashing is used by default. activeConsumerFailoverConsistentHashingfalsestandalone.conf单机模式# Enable consistent hashing for selecting the active consumer in partitioned topics with Failover subscription type. # For non-partitioned topics, consistent hashing is used by default. activeConsumerFailoverConsistentHashingfalse启用方式在 broker 配置文件或 standalone 配置文件中把该值改为true后重启 broker或 standalone 服务即可。注意它是 broker 级配置作用于该 broker 上所有采用 Failover 订阅的分区主题。底层依赖的哈希工具源码中的两处哈希调用分别来自不同的工具类Murmur3_32Hash.getInstance().makeHash(...)来自org.apache.pulsar.common.util用于构建哈希环时对消费者名 虚拟节点序号计算哈希Murmur3Hash32.getInstance().makeHash(...)来自org.apache.pulsar.client.impl用于对主题名计算哈希。两者均为 MurmurHash3 的 32 位实现保证了哈希的分布随机性与计算效率这也是一致哈希在大量主题场景下分布均匀的前提。对外行为变化消费者与分区的对应关系启用该配置后Failover 订阅模式下消费者与分区的对应关系将发生明显变化启用前按照文档描述第一个消费者必然消费 P1第二个消费者必然消费 P2……即按连接顺序依次分配分区启用后上述顺序对应关系不再成立由哈希算法决定哪个消费者消费哪个分区。第一个连接的消费者可能不再消费 P1具体归属由主题名哈希在哈希环上的落点决定。这一行为变化影响所有依赖消费者连接顺序 分区分配顺序这一隐含假设的客户端逻辑升级前需评估现有消费端是否依赖该顺序约定。向后与向前兼容性PIP-392 明确指出默认值为false未显式开启时保持原有的取模行为因此对存量集群与存量应用完全向后兼容该配置为纯新增字段不影响既有配置文件解析向前兼容老版本 broker 忽略未知字段也无障碍只有用户主动开启后分区主题的 Failover 分配行为才会改变属于可选开启的增强特性。总结与适用建议PIP-392 通过一个布尔开关把非分区主题上已经验证的一致哈希能力平滑引入分区主题的 Failover 订阅适用场景大量单分区或少数分区主题 正则订阅 Failover 模式的负载均衡优化典型如监控类、采集类主题群如topic-{0~100}这类批量命名主题核心收益避免所有主题的 Active Consumer 集中到首个连接的消费者让消费者集合在主题维度上分布更均匀提升整体吞吐与资源利用率注意事项消费者加入/退出导致的 Active Consumer 转移会带来消息重复投递取模算法同样存在消费端需保持幂等同时启用后分区分配不再遵循连接顺序依赖顺序语义的应用需谨慎升级路径先在测试集群将 broker.conf 或 standalone.conf 中的activeConsumerFailoverConsistentHashing置为true验证分配效果再逐步推广到生产集群。如需深入阅读可继续查看设计文档pip/pip-392.md核心实现AbstractDispatcherSingleActiveConsumer.java配置定义ServiceConfiguration.java配置文件broker.conf、standalone.conf赞分享消息队列流处理后端微服务消息路由【免费下载链接】pulsarApache Pulsar - distributed pub-sub messaging system项目地址https://gitcode.com/gh_mirrors/pu/pulsar点击查看免费下载相关推荐一致性哈希Consistent Hashing深度解析从取模哈希的痛点到大厂分布式系统实践一致性哈希Consistent Hashing深度解析从取模哈希的痛点到大厂分布式系统实践 一致性哈希是分布式系统中把数据分布到多台服务器上的经典技术被后端文档教程Apache Pulsar 主题级Topic-specific消费者 priorityLevel 配置PIP-184 实战与源码实现解析Apache Pulsar 主题级Topic specific消费者 priorityLevel 配置PIP 184 实战与源码实现解析 本指南围绕 Ap消息队列流处理后端微服务消息路由Grokking System Design 一致性哈希Consistent Hashing全解析哈希环、虚拟节点与分布式系统中的工程实践Grokking System Design 一致性哈希Consistent Hashing全解析哈希环、虚拟节点与分布式系统中的工程实践 一致性哈希C教程创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考