消息队列流处理后端微服务消息路由【免费下载链接】pulsarApache Pulsar - distributed pub-sub messaging system项目地址https://gitcode.com/gh_mirrors/pu/pulsar点击查看免费下载PIP-300 为 Apache Pulsar 引入了一项面向插件开发者的关键能力第三方插件如认证提供者、授权提供者可以在BrokerService中注册自定义动态配置项随后通过pulsar-admin在线更新其值并通过配置监听器感知变更。本文以 PIP-300 为骨架结合当前仓库中BrokerService.java的实现与AdminApiDynamicConfigurationsTest.java的测试用例从动机、API 用法、底层机制到实操验证进行完整解读读者阅读后可直接在自己的插件中接入这一能力。一、背景Pulsar 的动态配置机制在生产环境中动态配置是一项非常重要的能力。Pulsar 的 Broker 配置类ServiceConfiguration中凡是带有FieldContext(dynamic true)注解的字段都会被标记为动态配置运维人员可以直接使用pulsar-admin在线修改这些配置而无需重启 Broker。例如在 ServiceConfiguration.java 中可以看到大量此类字段maxConcurrentLookupRequest、maxConcurrentTopicLoadRequest、managedLedgerCacheSizeMB等其定义形如FieldContext( category CATEGORY_SERVER, required false, doc ... ) dynamic true, // 表示该配置支持动态更新 private int maxConcurrentLookupRequest ...;pulsar-admin提供了与动态配置相关的管理命令# 查询所有动态配置 pulsar-admin brokers get-all-dynamic-config # 更新某个动态配置的值Broker 无需重启即生效 pulsar-admin brokers update-dynamic-config --config key --value value # 删除动态配置使其回退到默认值 pulsar-admin brokers delete-dynamic-config keyPulsar 本身拥有多个可插拔插件体系例如认证提供者Authentication Provider、授权提供者Authorization Provider等。在插件内部开发者可以读取 Pulsar 的配置以及自定义配置。但在 PIP-300 之前自定义配置一旦写入配置文件就无法通过pulsar-admin在线变更插件的相关行为只能通过重启 Broker 调整这在需要热切换策略例如动态切换认证 ID的场景中十分不便。二、PIP-300 的目标与设计思路PIP-300 的目标非常明确允许pulsar-admin更新自定义配置的值并允许插件通过org.apache.pulsar.broker.service.BrokerService.registerConfigurationListener监听自定义配置的变更。其核心设计思路极其简洁BrokerService.dynamicConfigurationMap保存着所有动态配置项只要把需要动态配置的自定义项加入这个 Mappulsar-admin就能像操作内置动态配置一样更新它。关键约束是用户只能注册自定义配置且每个 key 只能注册一次。如果对同一个 key 重复注册会抛出IllegalArgumentException。三、核心 APIregisterCustomDynamicConfigurationPIP-300 在 BrokerService.java 中新增了公开方法/** * Allows the third-party plugin to register a custom dynamic configuration. */ public void registerCustomDynamicConfiguration(String key, PredicateString validator) { if (dynamicConfigurationMap.containsKey(key)) { throw new IllegalArgumentException(key already exists in the dynamicConfigurationMap); } ConfigField configField ConfigField.newCustomConfigField(null); configField.validator validator; dynamicConfigurationMap.put(key, configField); }参数说明参数类型说明keyString自定义配置项的名称必须全局唯一不能与已有动态配置或已注册的自定义配置重复否则抛出IllegalArgumentExceptionvalidatorPredicateString可选的值校验器。pulsar-admin更新配置时新值会先经过该校验器校验校验不通过则更新被拒绝返回PreconditionFailedException传null表示不做校验从源码可以看到注册时通过ConfigField.newCustomConfigField(null)创建了一个自定义配置项其内部field为null区别于普通动态配置项持有的反射FielddefaultValue为null并把validator挂到配置项上。随后该 key 被放入dynamicConfigurationMap从此它就与内置动态配置享有完全相同的pulsar-admin管理通道。相关辅助方法围绕dynamicConfigurationMapBrokerService还提供了一系列配套方法见 BrokerService.javapublic ListString getDynamicConfiguration() // 返回全部动态配置 key public boolean isDynamicConfiguration(String key) // 判断某 key 是否为动态配置 public boolean validateDynamicConfiguration(String key, String value) // 用 validator 校验新值 public T void registerConfigurationListener(String configKey, ConsumerT listener) // 注册变更监听器注意registerConfigurationListener内部会调用validateConfigKey(configKey)也就是说监听器只能注册到已经存在于dynamicConfigurationMap中的 key 上如果 key 尚未注册无论是内置还是自定义会抛出IllegalArgumentException(key doesnt exits in the dynamicConfigurationMap)。四、底层机制dynamicConfigurationMap 与 ConfigField要理解自定义动态配置为何能无缝接入需要看清dynamicConfigurationMap的构成与更新链路。1. 内置动态配置的构建反射扫描在BrokerService启动时prepareDynamicConfigurationMap()BrokerService.java通过反射扫描ServiceConfiguration的所有字段for (Field field : ServiceConfiguration.class.getDeclaredFields()) { if (field ! null field.isAnnotationPresent(FieldContext.class)) { field.setAccessible(true); if (field.getAnnotation(FieldContext.class).dynamic()) { Object defaultValue field.get(pulsar.getConfiguration()); dynamicConfigurationMap.put(field.getName(), new ConfigField(field, defaultValue)); } } }凡是带有FieldContext(dynamic true)注解的字段其反射 Field 与默认值都会被包装成ConfigField放入dynamicConfigurationMap。这解释了为什么pulsar-admin只认识注解过的动态配置——因为这张 Map 就是动态配置的白名单。2. ConfigField 内部结构ConfigField是BrokerService的私有静态内部类BrokerService.java包含四个关键成员private static class ConfigField { final Field field; // 内置配置对应的反射 Field自定义配置为 null volatile String lastDynamicValue; // 最近一次设置的动态值未设置过为 null final Object defaultValue; // 配置文件中初始化的默认值自定义配置为 null PredicateString validator; // 值校验器 ... public static ConfigField newCustomConfigField(String customValue) { ConfigField configField new ConfigField(null, null); configField.lastDynamicValue customValue; return configField; } }field null正是区分自定义配置项与内置动态配置项的标志这一区分在配置变更处理逻辑中起到了关键作用。3. 配置变更的完整处理链路当pulsar-admin执行update-dynamic-config后配置值被写入元数据服务通过PulsarResources的动态配置资源BrokerService随后异步拉取并逐个调用configValueChanged(configKey, newValueStr)BrokerService.javaprivate void configValueChanged(String configKey, String newValueStr) { ConfigField configFieldWrapper dynamicConfigurationMap.get(configKey); ... Consumer listener configRegisteredListeners.get(configKey); try { final Object existingValue; final Object newValue; if (configFieldWrapper.field ! null) { // 内置动态配置解析字符串并写回 ServiceConfiguration 对应字段 if (StringUtils.isBlank(newValueStr)) { newValue configFieldWrapper.defaultValue; } else { newValue FieldParser.value(newValueStr, configFieldWrapper.field); } existingValue configFieldWrapper.field.get(pulsar.getConfiguration()); configFieldWrapper.field.set(pulsar.getConfiguration(), newValue); } else { // 自定义配置项field null // 仅触发事件通知监听器不写入任何内存字段 existingValue configFieldWrapper.lastDynamicValue; newValue newValueStr null ? configFieldWrapper.defaultValue : newValueStr; } // 记录最新动态值 configFieldWrapper.lastDynamicValue newValueStr; ... if (listener ! null !Objects.equals(existingValue, newValue)) { listener.accept(newValue); } } catch (Exception e) { ... } }这段源码揭示了自定义配置的两条重要行为内置动态配置新值通过FieldParser.value解析并直接写回ServiceConfiguration对应字段配置立即在 Broker 内存中生效自定义配置由于没有对应的反射字段Pulsar不会也不可能把值写进某个配置字段而是把newValue交给监听器由插件自己决定如何使用新值——这正是监听器模式的底层依据值变更判定只有!Objects.equals(existingValue, newValue)时监听器才会被调用避免重复通知删除配置newValueStr null时监听器会收到null表示已回退到默认值。五、插件接入实战PIP-300 给出了在插件中接入该特性的完整范式。以认证插件为例在initialize(PulsarService)阶段完成注册与监听Override public void initialize(PulsarService pulsarService) throws Exception { String myAuthIdKey my-auth-id; myAuthIdValue pulsarService.getConfiguration().getProperties().getProperty(myAuthIdKey); pulsarService.getBrokerService().registerCustomDynamicConfiguration(myAuthIdKey, null); pulsarService.getBrokerService().registerConfigurationListener(myAuthIdKey, (newValue) - { // The myAuthIdKey value has changed myAuthIdValue String.valueOf(newValue); }); }这段代码的核心要点先读取初始值getProperties().getProperty(myAuthIdKey)从配置文件如conf/broker.conf中的自定义属性读取初始值并缓存注册自定义动态配置registerCustomDynamicConfiguration(myAuthIdKey, null)把my-auth-id加入动态配置白名单第二个参数传null表示不做值校验注册变更监听器registerConfigurationListener在值变化时回调插件只需在回调中更新自己的缓存变量即可实现配置热更新。配置校验器的使用如果希望限制可接受的取值范围可以在注册时提供PredicateString。例如只允许特定的非空值pulsarService.getBrokerService().registerCustomDynamicConfiguration( my-auth-id, value - value ! null value.matches(^[a-zA-Z0-9-]$));此后通过pulsar-admin更新为非法值时更新请求会因校验失败而被拒绝PulsarAdminException.PreconditionFailedExceptionHTTP 412插件不会收到非法值。六、pulsar-admin 实际操作与验证注册成功后运维人员即可用标准的动态配置命令管理自定义配置# 更新自定义配置 pulsar-admin brokers update-dynamic-config --config my-auth-id --value new-auth-id # 查看所有动态配置含自定义项 pulsar-admin brokers get-all-dynamic-config # 删除自定义配置回退到默认值监听器收到 null pulsar-admin brokers delete-dynamic-config my-auth-id这一流程在仓库测试用例 AdminApiDynamicConfigurationsTest.java 中有完整的端到端验证其中testRegisterCustomDynamicConfigurationL82-L113覆盖了以下关键场景// 1. 注册自定义动态配置并附带校验器 pulsar.getBrokerService().registerCustomDynamicConfiguration(key, value - !value.equals(invalidValue)); // 2. 重复注册同一 key 必须抛 IllegalArgumentException assertThrows(IllegalArgumentException.class, () - pulsar.getBrokerService().registerCustomDynamicConfiguration(key, null)); // 3. 未赋值前getAllDynamicConfigurations 中不包含该 key assertThat(allDynamicConfigurations).doesNotContainKey(key); // 4. 注册监听器后通过 admin 更新监听器异步收到新值 pulsar.getBrokerService().registerConfigurationListener(key, changeValue::set); admin.brokers().updateDynamicConfiguration(key, newValue); Awaitility.await().untilAsserted(() - { assertThat(changeValue.get()).isEqualTo(newValue); }); // 5. 传入校验器不通过的值更新请求被拒绝 assertThrows(PulsarAdminException.PreconditionFailedException.class, () - admin.brokers().updateDynamicConfiguration(key, invalidValue)); // 6. 删除后配置项从动态配置列表中消失 admin.brokers().deleteDynamicConfiguration(key);另一个用例testDeleteCustomizedDynamicConfigL169-L195则验证了删除语义注册时未赋值监听器初始收到null默认值updateDynamicConfiguration(a123, xxx)后监听器收到xxxdeleteDynamicConfiguration(a123)后监听器再次收到null即回退到默认值。从这些测试还可以确认两个容易被忽略的细节其一未赋值的自定义配置不会出现在get-all-dynamic-config结果中只有被update-dynamic-config显式设置后才可见其二自定义配置的默认值就是null。七、兼容性、限制与注意事项兼容性PIP-300 是纯增量特性不改变任何既有动态配置的语义也不改动pulsar-admin的对外命令格式因此对现有部署完全向后兼容插件 API 的变更仅体现为BrokerService新增公开方法registerCustomDynamicConfiguration不会破坏已有的插件 SPI 签名。限制与注意事项注册唯一性同一 key 只能注册一次重复注册直接抛IllegalArgumentException因此插件应在initialize阶段集中注册并避免与其他插件或内置配置重名先注册后监听registerConfigurationListener依赖dynamicConfigurationMap中已存在该 key必须先注册自定义配置再注册监听器自定义配置不写入配置字段由于ConfigField.field nullPulsar 不会把新值回写到ServiceConfiguration或任何 Java 字段值只经由监听器回调交给插件插件必须自行保存最新值如 PIP-300 示例中的myAuthIdValue变量删除即回退删除自定义动态配置后监听器收到null插件应将该值视为恢复默认null并做相应处理避免残留旧值更新是异步生效的pulsar-admin命令只是将新值写入元数据存储Broker 通过异步拉取触发监听器因此从命令执行到监听器回调存在短暂延迟测试中使用Awaitility轮询等待正是为了处理这一异步性。演进方向从源码结构看从源码结构看该机制为插件生态打开了更灵活的热更新空间插件既可以将自定义项纳入统一的动态配置管理面也可以结合多个监听器实现联动更新例如认证插件切换 ID 的同时更新授权策略。只要插件遵循注册 → 监听 → 自行维护状态的范式就能与 Pulsar 的内置动态配置体系完全对齐。八、小结PIP-300 用最简练的方式打通了插件自定义配置与Pulsar 动态配置体系之间的壁垒新增一个公开方法registerCustomDynamicConfiguration把自定义 key 塞进dynamicConfigurationMap这张白名单再借助已有的registerConfigurationListener监听机制完成值变更通知。其核心价值在于——插件开发者不再需要重启 Broker 就能热更新自己的配置且管理入口、命令工具、校验链路与内置动态配置完全一致。如需深入理解实现细节建议阅读 BrokerService.java 中的configValueChanged、prepareDynamicConfigurationMap与ConfigField相关代码并运行 AdminApiDynamicConfigurationsTest.java 中的两个自定义配置测试用例作为行为参照。赞分享消息队列流处理后端微服务消息路由【免费下载链接】pulsarApache Pulsar - distributed pub-sub messaging system项目地址https://gitcode.com/gh_mirrors/pu/pulsar点击查看免费下载相关推荐Apache Pulsar 自定义配置动态化基于 PIP-300 与 registerCustomDynamicConfiguration 的插件扩展指南Apache Pulsar 自定义配置动态化基于 PIP 300 与 registerCustomDynamicConfiguration 的插件扩展指南 导消息队列后端Apache Pulsar Schema 管理完全指南AutoUpdate、手动注册与自定义存储实战Apache Pulsar Schema 管理完全指南AutoUpdate、手动注册与自定义存储实战 本指南以 Apache Pulsar 的 Schema消息队列后端流处理Apache Pulsar PIP-337 详解可插拔 SSL Factory 插件自定义 SSLContext/SSLEngine 生成机制Apache Pulsar PIP 337 详解可插拔 SSL Factory 插件自定义 SSLContext/SSLEngine 生成机制 导读 本文基消息队列后端上一篇如何解决《鸣潮》帧率限制与画质优化难题WaveTools工具箱完全指南下一篇WaveTools终极指南如何简单快速解锁《鸣潮》120帧性能飞跃创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
