消息队列流处理后端微服务消息路由【免费下载链接】pulsarApache Pulsar - distributed pub-sub messaging system项目地址https://gitcode.com/gh_mirrors/pu/pulsar点击查看免费下载Apache Pulsar 的负载均衡器长期承担着 TopicBundle分配、Bundle 分裂与卸载三大职责但随着集群规模扩大基于 ZooKeeper 广播与 Leader RPC 驱动的旧架构在一致性、数据复制量和卸载可用性上逐渐暴露出瓶颈。本文以 PIP-192New Pulsar Broker Load Balancer为核心系统讲解 Pulsar 新一代负载均衡器的设计目标、四层负载数据模型、读写流程、Bundle 状态通道Service Unit State Channel的完整状态机与冲突解决算法并结合当前仓库源码pulsar-broker/.../loadbalance/extensions与 broker.conf 的扩展配置段给出可落地的配置与验证依据。读完本文你将理解新负载均衡器如何把分配与分裂决策下沉到每个 Broker、如何通过transfer预分配目标节点来缩短卸载停摆时间以及如何安全地在集群中启用与调优它。一、PIP-192 背景与项目目标PIP-192 源于社区对既有负载均衡器多个维度的改进观察。由于改动几乎触及数据模型、事件处理器、缓存/存储、日志/指标等所有位置PIP 决定新建一个负载均衡器并在新类中隔离新代码而不是在原实现上打补丁从而保证客户可以先通过配置安全地开启/关闭新负载均衡器再逐步淘汰旧实现。其目标分为用户面向目标与内部实现目标两个层面。1.1 用户面向目标逻辑以最小延迟将集群利用率平衡得尽可能均匀。日志 / 指标通过日志与指标透明地展示负载均衡决策过程。管理 API / 配置提供覆盖系统决策的方式如手动指定 Bundle 迁移目标减少需要手工调优的配置项数量提供带说明的、更合理的配置默认值第二阶段提供设置自定义负载均衡策略的能力。1.2 内部实现目标逻辑层面——保留三大核心负载均衡逻辑但使其更高效、更快具体算法改进另行讨论TopicBundle- Broker 分配改进随机化与分配分布Bundle 分裂重审当前基于阈值的策略Bundle 卸载重审卸载频率。实现层面将 bundle-broker 分配与 bundle 分裂决策分发到各本地 Broker 执行由 Leader Broker 同步协调 bundle 卸载决策减少 Broker 之间的负载数据复制量用 Topic TableView 取代负载数据的元数据存储ZK移除负载均衡逻辑中的客户端依赖移除分配逻辑中的客户端重定向在卸载逻辑中加入 bundle transfer 选项取代依赖客户端 broker discovery 调用的旧方式通过 bundle transfer 选项最小化 bundle 卸载导致的 topic 不可用时间引入 bundle 状态通道table-view使 bundle 负载均衡操作在 Broker 间保持一致且具备容错性将新负载均衡器代码隔离在新类中用 bundle 状态通道取代 bundle 所有权元数据存储ZK znodes。日志 / 指标为所有主要负载均衡事件增加有意义的日志与指标补充如何解读负载均衡指标与日志的文档。管理 API / 配置增加将 bundle 转移到指定 Broker 的 admin CLI增加必要的配置以覆盖负载均衡决策基于负载数据动态调整内部配置阈值让管理 API 具备容错性且易于监控。测试记录测试计划与覆盖状态为正常与异常场景添加单元测试增加全局负载均衡逻辑测试并对比当前 Load Manager 与新 Load Manager 的行为。二、API 变更新增目标 Broker 卸载选项PIP-192 提出为卸载操作新增目标参数将 topicbundle定向卸载transfer到指定 Brokerpulsar-admin topics unload persistent://tenant/namespace/topic --dest ${destination_broker}从当前仓库的 Admin CLI 实现看这一指定目标节点能力实际落地在namespace bundle 级卸载命令上CmdNamespaces.javapulsar-admin namespaces unload tenant/namespace --bundle start-boundary_end-boundary --destinationBroker brokerWebServiceAddress其中--destinationBroker别名-d指定目标 Broker 的 WebService 地址且必须与--bundle同时使用未指定 bundle 时给出ParameterException。与之配套ExtensibleLoadManagerImpl.unloadNamespaceBundleAsync会校验源 Broker 与目标 Broker 相同的非法场景ExtensibleLoadManagerImpl.java再通过publishUnloadEventAsync将 Unload 事件写入状态通道由目标 Broker 预接管所有权。这就是新方案与旧方案的本质区别先预分配新 owner再关闭旧连接客户端无需重新发起分配请求即可被重定向。三、高层组件设计新的 Load ManagerPIP-192 的落地载体是ExtensibleLoadManagerImpl源码它以更好的模块化重构既有负载均衡逻辑将新代码隔离在新类中不破坏既有逻辑新 Load Manager 在最初几个版本中默认禁用直到被验证稳定。从源码生命周期Constructor - initialize - start - close可以看到它聚合了以下子组件BrokerRegistryBroker 可用性注册表、LeaderElectionService基于 ZK 选举选举根路径为/loadbalance/extension/leader、ServiceUnitStateChannelBundle 状态通道、BrokerLoadDataReporter/TopBundleLoadDataReporter负载上报器、UnloadScheduler/SplitScheduler卸载与分裂调度器、UnloadManager/SplitManager状态机执行器以及一条BrokerFilter过滤器链含BrokerLoadManagerClassFilter、BrokerMaxTopicCountFilter、BrokerVersionFilter、BrokerIsolationPoliciesFilter、AntiAffinityGroupPolicyFilter最终通过LeastResourceUsageWithWeight等策略选出目标 Broker。在 ExtensibleLoadManagerImpl.java 中还能看到新方案的核心内部 topic 定义BROKER_LOAD_DATA_STORE_TOPIC non-persistent://pulsar/system/loadbalancer-broker-load-data TOP_BUNDLES_LOAD_DATA_STORE_TOPIC non-persistent://pulsar/system/loadbalancer-top-bundles-load-data三者连同状态通道 topic组成INTERNAL_TOPICS这些内部 topic 的分配会特殊处理为分配给通道 owner以避免循环依赖。四、负载数据模型Load Data Models新负载均衡器把负载数据拆分为四类模型各自承担不同职责、使用不同存储4.1 LocalBrokerDataBroker 的事实数据内容示例{webServiceUrl, pulsarServiceUrl, ...}存储持久化在 MetadataStoreZK中。这是每个 Broker 的身份 地址数据写入其临时 znodeephemeral znode用于监控存活 Broker与旧方案一致。4.2 BrokerLoadDataBroker 的负载数据内容示例{cpu, memory, io, msgIn/Out, ...}存储发布到BrokerLoadDataStoreTableView非持久化 topic。每个 Broker 周期性计算自身负载并发布到该 TableView由于非持久化 TableView 可能丢失数据需要为旧 KV 增加 TTL 墓碑tombstone策略。源码中对应BrokerLoadData类与TableViewLoadDataStoreImplstore 目录。4.3 BundlesLoadDataBundle 的负载数据内容示例{bundleName, msgIn/Out, ...}存储仅缓存在本地 BrokerIn-memory HashMap。每个 Broker 监控自己被分配的 bundle 的负载并存入本地缓存不复制到其他 Broker 的缓存。本地读取该数据并计算 top-n 高负载 bundle支撑本地化的分裂决策。4.4 TopBundlesLoadDataTop-n 高负载 Bundle 数据内容示例{brokerUrl, high_load_bundles: [{bundleName, ...}], ...}存储发布到TopBundlesLoadDataStoreTableView非持久化 topic同样需要 TTL 墓碑策略。每个 Broker 周期性计算自己的 top-n 高负载 bundle 并发布仅 Leader 消费该数据作为卸载transfer决策的输入。五、负载数据读写流程数据模型Write写入端Read读取端LocalBrokerData启动时各 Broker 将自身数据写入 MetadataStoreZK的临时 znode用于监控存活与旧方案一致Broker 级负载数据已移入 BrokerLoadData所有 Broker 读取以确认存活 Broker 列表BrokerLoadData各 Broker 周期性计算本地负载并发布到 BrokerLoadDataStoreTableView非持久化配合 TTL 清理旧 KV所有 Broker 消费该 Store基于聚合结果所有 Broker 无需经过 Leader 即可执行 bundle 分配BundlesLoadData各 Broker 监控所分配 bundle 的负载存入本地缓存不跨 Broker 复制各 Broker 本地读取并计算 top-n 高负载 bundleTopBundlesLoadData支撑无需 Leader 的本地分裂决策TopBundlesLoadData各 Broker 周期性计算并发布到 TopBundlesLoadDataStoreTableView非持久化配合 TTL 清理仅 Leader消费Leader 结合聚合后的 TopBundlesLoadData 与 BrokerLoadData 发起 bundle 卸载transfer用一句话概括数据流向分配看全局 Broker 负载、分裂看本地 Bundle 负载、卸载看全局 top-n Bundle 负载——每类决策使用恰好够用的最小数据集这正是减少负载数据复制目标的具体体现。六、Bundle Split / Unload / Assignment 三大流程的修改新方案对三大核心流程的决策权做了如下再分配分裂Split凭借本地 BundlesLoadData所有 Broker 无需经过 Leader 即可执行 bundle 分裂默认情况下新分裂出的子 bundle 将成为卸载transfer的目标。分配Assignment凭借聚合的 BrokerLoadData所有 Broker 无需经过 Leader 即可执行 bundle 分配。卸载Unload/Transfer凭借聚合的 TopBundlesLoadData 与 BrokerLoadData由 Leader 决策卸载transfer哪些 bundle。新增transfer卸载选项把 bundle 从一台 Broker 转移到另一台。引入全局通道Bundle State Channel向所有 Broker 共享一致的、线性化的 bundle 状态变更。七、Bundle State Channel服务单元状态通道Bundle 状态通道是一个持久化 topic 的 table-view充当 WALWrite-Ahead Log用于广播集群中所有 bundle 状态变更的全序。所有 Broker 异步地按相同顺序消费该通道中的消息并响应状态变更顺序一致性。借助 table-view 的 compaction通道最终会物化出当前的 bundle-broker 所有权。该通道上的读操作例如客户端的 topic lookup 请求可以延迟数秒具体取决于 bundle 的当前状态。在仓库中这一设计对应 ServiceUnitStateChannel 接口及其实现ServiceUnitStateChannelImpl/ServiceUnitStateTableViewImpl接口提供getOwnerAsync、getAssigned、publishAssignEventAsync、publishUnloadEventAsync、publishSplitEventAsync等核心操作。7.1 Bundle 状态生命周期PIP-192 定义了以下状态与动作并将 bundle 状态变更线性化注意这是解释概念的高层设计最终版本可能略有差异Bundle 动作ActionsOwn拥有 bundle 所有权——owner Broker 由本地 Load Manager 选出。Transfer将 bundle 所有权转移给目标 Broker——源 Broker 内部禁用该 bundle 所有权目标 Broker 接管。Return以目标 Broker URL 返回被延迟的客户端连接——若连接已在服务则直接关闭。Split将目标父bundle 分裂为子 bundle。Create在通道中创建子 bundle 条目初始分配给本地 Broker。Discard丢弃通道中的 bundle 条目tombstone 操作。Unload从 owner Broker 卸载 bundle 所有权——禁用所有权、关闭该 bundle 下的客户端连接、执行 Discard 动作。Bundle 状态StatesAssigned已分配给某 BrokerAssigning正在分配所有权的过程中Splitting正在分裂 bundle 范围的过程中Unassigned未分配给任何 Broker已从通道中移除。在 Assigning 状态下针对该 bundle 的新客户端连接会被延迟带超时直到收到 Return 动作。7.2 Bundle 状态变更示例Bundle Transfer转移示例状态/动作序列(Assigned, Transfer) (Assigning, Return) (Assigned,)Leader 从 TopBundlesLoadData 中找到目标 bundle通过向状态通道广播卸载转移状态变更来发起卸载以 bundleName 为 key。例如{key: bundleName, value: {flow: transfer, action: transfer, state: assigning, from: A, to: B}}所有 Broker 消费通道中的状态变更消息消费到消息后若状态变更涉及本地 Broker该 Broker 执行自己的角色并把新状态更新回通道以推进状态变更若与进行中的状态变更冲突则忽略与此同时若其他 Broker如 Broker C收到该 bundle 的 lookup 请求客户端连接会被延迟带超时直到收到 Return 动作当 Return 动作广播后所有 Broker 以 owner Broker 的 URL 返回挂起的连接源 Broker 上的既有连接被关闭。Bundle Split分裂示例状态/动作序列(Assigned, Split) (Splitting, Unload | Create) {(Unassigned, ) | (Assigned, ), (Assigned, )}每个 owner Broker 监控本地 BundlesLoadData通过广播转移状态变更来发起分裂以 bundleName 为 key。例如{key: bundleName, value: {flow: split, action: split, state: splitting, from: A, to: B, transfer: true}} 2-3. 与 Transfer 示例的步骤 2、3 相同。特别地分裂完成后owner 广播子 bundle 的所有权创建stateassigned与父 bundle 的所有权卸载空消息默认情况下owner 还会向 TopBundlesLoadData store 发布消息请求 Leader 卸载或转移子 bundle。Bundle Assignment分配示例状态/动作序列(Unassigned, Own) (Assigning, Return) (Assigned,)客户端请求到来时首先连接的 Broker 检查状态通道中是否有任何 Broker 拥有该 bundle有则直接返回 owner Broker 的 URL没有则广播分配状态变更发起分配。例如{key: bundleName, value: {flow: assignment, action: own, state: assigning, to: B}} 2-4. 与 Transfer 示例的步骤 2、3、4 相同。7.3 Bundle-Broker 所有权状态由于状态通道本身展示了当前的 bundle-broker 所有权就可以移除冗余的 bundle 所有权存储ZK znodes。每个 Broker 通过查询 bundle 所有权通道判断哪个 Broker 拥有请求的 bundle或该 bundle 正处于所有权分配/卸载transfer过程中。此外在 Return 之前还可以检查 Broker 可用性元数据存储LocalBrokerData znode 是否存在进一步确认 owner Broker 的可用性。7.4 通道 Owner 的选择与发现Bundle State ChannelBSC本身也是一个 topic因此存在循环依赖集群启动时每个 Broker 需要发起 BSC TopicLookUp找到 owner Broker才能消费 BSC 中的消息但初始状态下没人知道谁拥有 BSC。PIP-192 给出的解法是Channel Owner 选择利用ZK Leader 选举选出 owner Brokerowner 不可用时某个 follower 会成为新 owner可以为每个 bundle 状态分区分别选举 owner。Channel Owner 发现在 Broker 的 TopicLookUp 逻辑中增加特例对 BSC topic 直接返回当前 Leader被选出的 BSC owner。在源码中ExtensibleLoadManagerImpl.start()正是通过LeaderElectionService选举根路径/loadbalance/extension/leader选主并通过getChannelOwnerAsync/isChannelOwnerAsyncServiceUnitStateChannel.java完成通道 owner 的查询与判断assign()中对内部 topic 也特殊处理为分配给通道 owner从而打破循环引用ExtensibleLoadManagerImpl.java。7.5 冲突状态解决竞态条件不借助分布式锁PIP-192 通过乐观的、最终一致的冲突解决算法处理冲突Brokers 在线性化视图中取第一个有效的状态变更作为胜者忽略后续的冲突变更。需要注意一个关键实现约束由于当前 table-view 的 compaction 只保留最后一个值需要为该通道引入内部 compaction 算法使其遵循第一个有效状态变更作为结果值的冲突解决算法。伪代码如下Bundle State Conflict Resolution Algorithm Example For each bundle: // A[i] is a linearized bundle state change action at i, and // S is the current bundle state after A[i-1], // where the sequence number i monotonically increases. for each A[i] and S: // no arrows in the state diagram If A[i] is invalid from S: Reject A[i] Else: Accept A[i]示例一假设 bundle x 同时发起了两个冲突的分配线性化的状态变更消息如下(own, to:B), (own, to:A)按冲突解决算法第二个状态变更(own, to:A)会被所有 Broker以及 compaction 算法忽略。最终会广播 return 消息声明 owner 是 B(own, to:B), (own, to:A), (return, to:B)示例二假设 bundle x 已分配给 Broker B但另一个 Broker 在消费 return 动作之前又发起了 own 动作。由于在状态图中从 assigned 状态出发没有 own 动作箭头这个 own 状态变更会被忽略(own, to:B), (return, to:B), (own, to:A)仓库中对应的实现是ServiceUnitStateDataConflictResolverchannel 目录。7.6 故障恢复当 Broker 宕机时当状态变更的参与者Broker突然不可用时状态变更可能成为孤儿orphan因为参与者没有执行自己的角色。针对这些孤儿状态变更Leader Broker 运行孤儿状态清理逻辑在 Broker 不可用通知处理器znode watcher中加入 bundle 状态清理逻辑清理来自不可用 Broker 的挂起状态变更与所有权为提升容错性Leader 在初始化时也会运行清理函数此外可以让 Leader 在独立的监控线程中周期性调用清理但不应过于频繁地冗余调用。从源码看ExtensibleLoadManagerImpl内置MONITOR_INTERVAL_IN_MILLIS 120_000的周期监控任务并提供了loadBalancerServiceUnitStateMaxConcurrentOverrides孤儿 bundle 所有权覆盖并发上限默认 64与loadBalancerInFlightServiceUnitStateWaitingTimeInMillis等待修复 in-flight 状态的时长等配套配置正是这条清理路径的工程化落地。当整个 ZK 宕机并恢复时ZK 会话出现连接问题时每个 Broker 都会收到通知随后进入safe 模式继续按现状服务既有 topic但不允许执行 ZK 相关操作Leader 在知晓 ZK 宕机时不会运行 bundle 清理、transfer 或 unload 逻辑。ZK 恢复后各 Broker 会感知到 ZK 会话重新建立等待 2-3 分钟让所有 Broker 完成 ZK hand-shaking然后恢复 bundle 状态 table-view 并返回正常模式。7.7 Bundle 状态与负载数据 TableView 的可扩展性预期的读写流量写相对较少bundle 状态变更频率较低偶有尖峰读在超大集群中fan-out 广播可能成为瓶颈。由于 bundle 状态变更相对不频繁bundle 状态通道在生产端的压力较轻但当集群非常大时向消费者分发消息的开销可能较重。同样的问题也适用于本提案引入的其他 table-view如 BrokerLoadDataStore。PIP-192 给出了三种扩展思路将 Broker 集群拆分为多个集群最直接的做法——用不同端点把大规模 Broker 集群拆成多个集群BookKeeper 与配置层可以在多个 Broker 集群间共享。分区 TableView短期基于分区 topic 构建 table-view将消息负载分散到多个分区 owner Broker。分片长期按传统扩展思路把集群分成多个 Broker 组为每个分片创建独立通道。这需要额外的发现层将 topic 映射到 Broker 分片并需与 Namespace Isolation Policies 对齐。需要说明的是元数据同步的可扩展性问题在 Pulsar 中并非新问题——当前 Pulsar 使用 n-replication所有 Broker 与所有 bundle 的负载元数据都通过 ZK watcher 复制到所有 Broker而本提案中多个 table-view owner Broker配合分区 table-view可以向参与者Broker分发元数据变更消息。PIP-192 认为该元数据同步可扩展性属于较低优先级因为只有少量客户运行如此大规模的 Pulsar 集群。建议先让客户拆分集群、再启用分区 table-view单集群上千 Broker 并不现实。但设计上仍要保证无缝可扩展two-way-door 决策。八、为什么拒绝增强现有负载均衡器PIP-192 专门说明了为何不直接在旧负载均衡器上增量改造由于 PIP 几乎改动所有位置数据模型、事件处理器、缓存/存储、日志/指标新建并隔离新代码更安全、更干净。客户可在弃用旧实现前通过配置安全地启用/禁用新负载均衡器。它带来从零开始、不受既有包袱约束的灵活性可以尝试显著不同的方案。现有的ModularLoadManagerImpl不会消失新 Load Manager 稳定到一定程度后社区可以另行讨论是否更换默认实现即便如此用户仍可选择旧 Load Manager。在 broker.conf 中可以看到旧实现仍是默认值loadManagerClassNameorg.apache.pulsar.broker.loadbalance.impl.ModularLoadManagerImpl这也印证了 PIP-192新负载均衡器默认禁用、待验证稳定后再切换的规划。九、修改摘要Before / After 对照PIP-192 的修改摘要不含逻辑与算法改进因为本 PIP 不聚焦逻辑与算法优化如下目标Before改动前After改动后使负载均衡操作在 Broker 间容错且一致Leader Broker 通过带重试的 RPC 向 owner Broker 发送负载均衡命令引入全局 bundle 状态通道持久化 topic 的 table-viewbundle 命令的全序被可靠持久化并由所有 Broker 广播分散负载均衡操作Leader 决策 bundle 分配、卸载与分裂owner Broker 经 RPC 通知执行卸载与分裂每个 Broker 决策并执行 bundle 分配与分裂Leader 决策 bundle 卸载transferowner Broker 经状态通道通知执行卸载减少 Broker 间负载数据复制所有 Broker 与所有 bundle 的负载数据存于 ZK 并经 ZK watcher 复制到所有 Broker所有 Broker 的负载数据经非持久化 topictable-view复制到所有 Broker仅每台 Broker 的 top-n bundle 负载数据经非持久化 topictable-view复制到 Leader最小化卸载导致的 topic 不可用topic 连接关闭后客户端重连到新 Broker新 Broker 发起新的 topic 分配Leader 分配新 owner客户端最终被重定向到新 owner新增 transfer 卸载选项在关闭 topic 连接前预分配新 owner客户端立即重定向到新 owner无需客户端发起的 topic 分配在 Broker 间共享 bundle-broker 所有权元数据以支持 owner 发现bundle-broker 所有权数据存于 ZK所有 Broker 在 TopicLookUp 请求时读取 bundle 所有权信息并缓存本地 bundle 所有权信息全局所有权数据存于 bundle 状态通道持久化 topic 的 table-view借助 compaction所有 Broker 在 TopicLookUp 时读取其最新的全局所有权 table-view内存缓存用日志与指标透明展示负载均衡决策尽力而为地输出日志将日志/指标设计为独立逻辑组件为所有重要负载均衡事件记录并共享主要日志消息与指标十、落地实现配置项全解与源码对照PIP-192 的 Post Update 为ServiceConfiguration新增了一整套loadBalancer*扩展配置。这些配置已全部落地到 broker.conf 的### --- Load balancer extension --- ###段并通过ExtensibleLoadManagerImpl、TransferShedder等类读取生效。下表逐项说明默认值以当前仓库 broker.conf 为准配置项默认值说明loadBalancerDebugModeEnabledfalse开启负载均衡逻辑调试模式打印更多日志如负载均衡状态与决策仅用于扩展逻辑loadBalancerBrokerLoadTargetStd0.25Broker 资源使用的目标标准差100% 资源使用 1.0 load。shedder 试图跨 Broker 分配 bundle 负载以达到该目标 std值越小负载均衡越频繁。仅 TransferShedder 使用loadBalancerSheddingConditionHitCountThreshold3连续满足卸载shedding条件的计数阈值超过该阈值调度器才真正卸载 bundle。值越大卸载/转移越少。仅 TransferShedder 使用loadBalancerTransferEnabledtrue是否启用 bundle transfer 模式分发负载On 卸载时预分配目标 BrokertransferOff 卸载后在 lookup 时后置分配目标 Broker。仅 TransferShedder 使用loadBalancerMaxNumberOfBrokerSheddingPerCycle3每个卸载周期最多卸载多少台 Broker 的 bundle 负载值越大每周期卸载/转移越多。仅 TransferShedder 使用loadBalanceSheddingDelayInSeconds180卸载后距下一个卸载周期的延迟秒给 Broker 足够时间重算负载值越大延迟越久。仅 TransferShedder 使用loadBalancerBrokerLoadDataTTLInSeconds1800Broker 负载数据 TTL秒避免使用可能不可用的过期负载数据的 Broker。调优时需参考loadBalancerReportUpdateMaxIntervalMinutes当前默认值为其 2 倍。仅 TransferShedder 使用loadBalancerMaxNumberOfBundlesInBundleLoadReport10每台 Broker 在 bundle 负载报告中上报的 bundle 数量上限。基于 topK bundle 负载与其他 Broker 负载做分配值越大上报开销越高。ExtensibleLoadManagerImpl与ModularLoadManagerImpl均使用Modular 置 -1 可关闭过滤loadBalancerSplitIntervalMinutes1服务单元bundle分裂检查间隔Broker 周期性检查热点 bundle 是否需要分裂loadBalancerMaxNumberOfBundlesToSplitPerCycle10每个周期最多分裂的 bundle 数量loadBalancerNamespaceBundleSplitConditionHitCountThreshold3连续满足分裂条件的计数阈值超过后触发分裂若 bundle 数小于loadBalancerNamespaceMaximumBundlesloadBalancerServiceUnitStateTombstoneDelayTimeInSeconds3600状态通道对半终止状态semi-terminalservice unit 执行 tombstone 的延迟。例如分裂后父 bundle 先deleted再经此延迟后tombstonedPulsar 不立即移除半终止状态以避免混淆tombstoned 状态可能暂时看起来可重新分配。极少情况下可调低以激进清理通道最小 30 秒loadBalancerSheddingBundlesWithPoliciesEnabledfalse是否自动卸载带亲和isolation或反亲和组策略的 namespace bundle。此类 bundle 因目标 Broker 受限不适合作为自动卸载目标loadBalancerInFlightServiceUnitStateWaitingTimeInMillis30000Leader 监控器修复卡住的 in-flight service unit 状态的等待时长超过该时长则重新分配所有权loadBalancerServiceUnitStateMonitorIntervalInSeconds60Leader Broker 周期性监控状态通道的间隔用于修复孤儿 bundle 所有权、卡住的 in-flight 状态及其他清理任务。注意loadBalancerServiceUnitStateTombstoneDelayTimeInSeconds * 1000必须大于loadBalancerInFlightServiceUnitStateWaitingTimeInMillisloadManagerServiceUnitStateTableViewClassNameorg.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitStateTableViewImplServiceUnitStateTableView 实现类名broker.conf 中额外提供loadBalancerServiceUnitTableViewSyncerNone迁移期间在 metadata store 与 system topic table-view 之间同步 bundle 状态。可设MetadataStoreToSystemTopicSyncer或SystemTopicToMetadataStoreSyncer启用None禁用broker.confloadBalancerServiceUnitStateMaxConcurrentOverrides64Leader 检测到孤儿 bundle 所有权后触发并发覆盖的最大数量broker.conf10.1 TransferShedder新方案的卸载策略实现在仓库中TransferShedder源码正是 PIP-192 中 TransferShedder 概念的落地实现其类注释明确列出了设计目标TransferShedder.java与本文前述配置一一对应以loadBalancerBrokerLoadTargetStd为目标降低各 Brokerstd(exp-moving-avg(max(cpu, memory, network, throughput)))以loadBalancerTransferEnabledtrue启用 transfer 协议把负载从最高负载 Broker 转移到最低负载 Broker通过卸载后重算历史资源使用率、跳过近期已卸载 bundle避免重复卸载当目标 Broker 消息吞吐量为零新 Broker时优先向其卸载不使用过期 Broker 负载数据loadBalancerBrokerLoadDataTTLInSeconds卸载后给足重算负载的时间loadBalanceSheddingDelayInSeconds不转移带 namespace isolation policies 或 anti-affinity group policies 的 bundle限制每个周期转移负载的 Broker 数上限loadBalancerMaxNumberOfBrokerSheddingPerCycle支持loadBalancerDebugModeEnabledtrue打印更多日志。在 broker.conf 中也明确提示使用可扩展负载管理器ExtensibleLoadManagerImpl时应将卸载策略配置为loadBalancerLoadSheddingStrategyorg.apache.pulsar.broker.loadbalance.extensions.scheduler.TransferShedder10.2 相关测试与验证仓库提供了针对新负载均衡器的完整测试覆盖可供读者深入学习行为细节TransferShedderTest.java卸载策略单元测试ExtensibleLoadManagerImplTest.java 与 ExtensibleLoadManagerImplBaseTest.java新 Load Manager 整体行为测试CustomBrokerSelectionStrategyTest.java自定义 Broker 选择策略对应 PIP 第二阶段的自定义策略目标集成测试ExtensibleLoadManagerTest.java 与 ProxyWithExtensibleLoadManagerTest.java。十一、总结与启用建议PIP-192 描绘并落地了 Pulsar 新一代负载均衡器的完整蓝图用持久化 topic table-viewBundle State Channel替代 ZK 所有权存储与 RPC 命令下发用非持久化 table-view 承载全局 Broker 负载与 top-n Bundle 负载把分配与分裂决策下沉到每个 Broker而 Leader 只保留卸载决策与孤儿清理职责同时通过 transfer 预分配机制显著缩短卸载停摆时间。给实际使用者的启用建议可归纳为在 broker.conf 中把loadManagerClassName切换为org.apache.pulsar.broker.loadbalance.extensions.ExtensibleLoadManagerWrapper或按发行版文档启用 extension Load Manager并确认内部 topic 的 compaction threshold 已正确配置源码中由configureSystemTopics自动设置参考 ExtensibleLoadManagerImpl.java将loadBalancerLoadSheddingStrategy设为TransferShedder按集群规模与负载波动调优loadBalancerBrokerLoadTargetStd、loadBalancerMaxNumberOfBrokerSheddingPerCycle、loadBalancerBrokerLoadDataTTLInSeconds等扩展配置开启loadBalancerDebugModeEnabledtrue观察决策日志逐步验证后再正式切换默认实现超大集群优先采用拆分为多个 Broker 集群 共享 BookKeeper/配置层的方案再评估分区 table-view。新负载均衡器在最初版本中保持默认禁用正是为了让用户可以在生产环境旁路验证、灰度切换——理解其数据模型、状态机与冲突解决机制是安全使用这一新架构的前提。赞分享消息队列流处理后端微服务消息路由【免费下载链接】pulsarApache Pulsar - distributed pub-sub messaging system项目地址https://gitcode.com/gh_mirrors/pu/pulsar点击查看免费下载相关推荐Apache Pulsar 负载均衡架构与配置实战Bundle 分配、Unload 与自动负载均衡Apache Pulsar 负载均衡架构与配置实战Bundle 分配、Unload 与自动负载均衡 本文以 Apache Pulsar 的负载均衡机制为主题消息队列后端流处理Apache Pulsar 模块化负载均衡器Modular Load Manager深度解析启用、验证与实现原理Apache Pulsar 模块化负载均衡器Modular Load Manager深度解析启用、验证与实现原理 Apache Pulsar 的 modu消息队列后端流处理Apache Pulsar 负载分布机制详解Broker 间流量均衡、Bundle 拆分与自动负载均衡实践Apache Pulsar 负载分布机制详解Broker 间流量均衡、Bundle 拆分与自动负载均衡实践 本文基于 Apache Pulsar 官方文档v消息队列后端流处理上一篇3个关键优化让ReactFlow拖拽如丝滑从卡顿到毫秒级响应的实战指南下一篇ssd_keras终极Keras目标检测框架快速实现高精度物体识别创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考