企业官网建设流程全解析

• 人工智能
• AI Agent
• Agent 框架
• 工具调用
• Agent 工作流
• RAG
• MCP Clients
• 强化学习

分布式 Runner（Distributed Runner）是 openJiuwen agent-core-java 在多实例、跨进程场景下执行 Agent、Workflow 的基础设施。DistributedConfig正是这套分布式执行器的核心配置类：它封装了请求超时、最大并发、消息队列选型与主题（Topic）命名模板等关键参数，直接决定分布式运行时如何通信、如何限流、如何路由消息。读完本文，你将掌握DistributedConfig每个字段的默认值与底层语义，理解 Topic 模板与envPrefix的拼接规则，并能在自己的 Runner 启动流程中正确配置与验证分布式运行参数。

1. DistributedConfig 在 Runner 架构中的位置

在com.openjiuwen.core.runner包中，分布式运行配置的装配关系清晰分层：

• RunnerConfig是 Runner 全局配置入口，持有distributedMode（是否开启分布式模式，默认true）、envPrefix、instanceId等字段，并通过distributedConfig字段内嵌一个DistributedConfig实例；
• DistributedConfig进一步持有消息队列配置MessageQueueConfig，后者再引用PulsarConfig；
• RunnerImpl在start()阶段读取RunnerConfig.getRunnerConfig()，当distributedMode=true时，通过MessageQueueFactory.create(...)创建分布式消息队列并启动ReplyTopicSubscription，本地与分布式两级消息通路随即就位。

也就是说，DistributedConfig位于“Runner 全局配置 → 分布式子配置 → 消息队列配置”这条装配链的中间层，负责承载分布式运行特有的参数，而不是本地内存模式（distributedMode=false，使用FakeMessageQueue）关心的内容。

2. 字段总览：五个配置项与默认值

DistributedConfig以 Lombok@Data+@Builder实现，全部字段都带有@Builder.Default默认值与 Jackson@JsonProperty序列化名。字段与默认值如下（与官方 API 文档一致）：

类上标注的@JsonInclude(JsonInclude.Include.NON_NULL)意味着序列化时值为null的字段会被忽略，便于生成精简的配置 JSON。由于采用 Builder 模式，你可以只覆盖关心的字段：

DistributedConfig config = DistributedConfig.builder() .requestTimeout(60.0) .maxRequestConcurrency(5000) .build();

2.1 requestTimeout：远端调用的请求超时

requestTimeout（默认30.0秒）用于约束分布式模式下“发起请求 → 等待远端 Agent 应答”的等待时间。从源码调用链看，它主要在两处生效：

• MqRemoteClient的effectiveTimeout(timeoutSeconds)方法：当调用方未显式传入超时（参数为null）时，回退到RunnerConfigAccess.requestTimeout()，即DistributedConfig.requestTimeout的值；当显式传入0.0时表示不设超时；
• 请求与 STOP 消息均通过setExpireAt(nowSeconds() + timeoutSeconds)写入过期时间戳（MqRemoteClient.java），使消息在消息队列侧具备过期语义。

同时，ResponseCollector创建时也会携带该超时值（默认 30.0 秒的兜底逻辑见ReplyTopicSubscription），远端无响应时收集器按 TTL 过期清理。调大该值适用于远端 Agent 推理耗时较长的场景，调小则可加快失败感知。

2.2 maxRequestConcurrency：在途请求并发上限

maxRequestConcurrency（默认10000）控制同一 Runner 实例上同时登记在途响应的最大数量。它的落地位置在ReplyTopicSubscription.registerCollector()：每次发起分布式调用前都要先注册一个ResponseCollector，注册时检查collectors.size() >= maxConcurrency，超过则抛出"Too many collectors (...)"运行时异常，防止无界并发压垮订阅方。

该值属于容量规划参数：对高吞吐、多 Agent 并行编排的部署建议保持默认或调大；对资源受限的实例则适当调小，让超限错误在入口处快速暴露。

2.3 messageQueueConfig：消息队列选型

messageQueueConfig（默认new MessageQueueConfig()）决定分布式 Runner 实际使用哪一类消息队列。其内部结构为：

• type：消息队列类型，默认MessageQueueType.PULSAR.getValue()，即"pulsar"；可选项由MessageQueueType枚举定义，包含PULSAR("pulsar")与FAKE("fake")两种，大小写不敏感（fromValue内部统一转为小写匹配）；
• pulsarConfig：Pulsar 专属配置，见下。

MessageQueueFactory.create()（MessageQueueFactory.java）根据type分发：fake返回内存版FakeMessageQueue（本地调试/单元测试用）；pulsar则通过反射加载扩展类com.openjiuwen.extensions.message_queue.MessageQueuePulsar（该实现位于 extensions 扩展模块，仓库默认 core 中未内置），加载失败会抛出StatusCode.MESSAGE_QUEUE_INITIATION_ERROR对应错误。注意：选择pulsar时必须确保 classpath 中存在对应的 Pulsar 消息队列扩展实现，否则 Runner 启动会失败。

2.4 agentTopicTemplate 与 replyTopicTemplate：消息路由的命名骨架

两条 Topic 模板是分布式通信的路由基石：

• agentTopicTemplate：默认"openjiuwen.single_agent.{agent_id}.{version}"，用于生成“某个 Agent 的某个版本”所监听的请求主题。占位符{agent_id}替换为 Agent 标识、{version}替换为版本号，实际拼接逻辑见DistributedRunner.agentTopic()与RunnerConfigAccess.agentTopic()（后者额外兼容{instance_id}占位符）；
• replyTopicTemplate：默认"openjiuwen.reply.runner.{instance_id}"，用于生成“某个 Runner 实例”专属的应答主题。{instance_id}由RunnerConfig.instanceId提供（默认UUID.randomUUID().toString()），保证每个 Runner 实例的应答通道彼此隔离，见DistributedRunner.replyTopic()。

3. getAgentTopicTemplate / getReplyTopicTemplate：环境前缀拼接

DistributedConfig暴露的两个方法本质是“带环境前缀的模板解析器”：

public String getAgentTopicTemplate(String envPrefix) { if (envPrefix != null && !envPrefix.isEmpty()) { return envPrefix + "." + agentTopicTemplate; } return agentTopicTemplate; } public String getReplyTopicTemplate(String envPrefix) { if (envPrefix != null && !envPrefix.isEmpty()) { return envPrefix + "." + replyTopicTemplate; } return replyTopicTemplate; }

语义规则非常直观：

• 当envPrefix为空或null时，返回原始模板字符串；
• 否则在模板最前面拼接envPrefix + "."。

envPrefix由RunnerConfig.envPrefix（默认""）提供，典型用途是在同一套消息队列中隔离不同环境（如dev、staging、prod）或不同租户的 Topic。例如配置envPrefix="prod"后，RunnerConfig.agentTopicTemplate()/replyTopicTemplate()返回的最终模板为：

• Agent 主题：prod.openjiuwen.single_agent.{agent_id}.{version}
• 应答主题：prod.openjiuwen.reply.runner.{instance_id}

RunnerConfig侧对应方法（RunnerConfig.java）只是对distributedConfig.getAgentTopicTemplate(envPrefix)/getReplyTopicTemplate(envPrefix)的薄封装，因此所有环境前缀逻辑都收敛在DistributedConfig内。

3.1 完整示例：配置环境前缀并读取最终模板

RunnerConfig config = RunnerConfig.builder() .distributedMode(true) .envPrefix("prod") .distributedConfig(DistributedConfig.builder() .agentTopicTemplate("openjiuwen.single_agent.{agent_id}.{version}") .replyTopicTemplate("openjiuwen.reply.runner.{instance_id}") .build()) .build(); Runner.setConfig(config); // 带环境前缀的模板 System.out.println(config.agentTopicTemplate()); // prod.openjiuwen.single_agent.{agent_id}.{version} System.out.println(config.replyTopicTemplate()); // prod.openjiuwen.reply.runner.{instance_id} // 生成具体主题：替换占位符 System.out.println(DistributedRunner.agentTopic("order_agent", "v1")); // prod.openjiuwen.single_agent.order_agent.v1 System.out.println(DistributedRunner.replyTopic()); // prod.openjiuwen.reply.runner.<instanceId>

4. 深拷贝：copy() 的防御性设计

DistributedConfig.copy()通过 Builder 重建一份独立配置，并递归深拷贝内部的MessageQueueConfig（messageQueueConfig.copy()），后者又会深拷贝PulsarConfig（pulsarConfig.copy()）。这一设计的实际用途是：RunnerConfig.getRunnerConfig()在全局配置未设置时，会compareAndSet(null, DEFAULT_RUNNER_CONFIG.copy())（RunnerConfig.java），并以RunnerTest.resetConfig()中RunnerConfig.DEFAULT_RUNNER_CONFIG.copy()的用法（RunnerTest.java）保障测试隔离——每次复制都产生完全独立的对象图，避免共享可变状态串扰。

值得注意的是，默认配置DEFAULT_RUNNER_CONFIG（RunnerConfig.java）使用distributedMode=false且消息队列类型为FAKE，即开箱即用的是本地内存模式；要真正启用分布式，需要显式配置distributedMode(true)并设置messageQueueConfig的type为pulsar（且 classpath 含扩展实现）。

5. 在 Runner 启动流程中的实际生效路径

把上述字段串联起来，分布式 Runner 的启动时序如下（对应RunnerImpl.start()）：

1. 读取RunnerConfig.getRunnerConfig()，初始化 Checkpointer；
2. 若config.isDistributedMode()为真，调用MessageQueueFactory.create(distributedConfig.getMessageQueueConfig())创建分布式消息队列并start()；
3. 创建ReplyTopicSubscription（默认订阅replyTopicTemplate解析出的主题，其中{instance_id}替换为config.getInstanceId()）并激活；
4. 随后本地消息队列messageQueue启动，整个 Runner 进入可服务状态；
5. stop()时按逆序关闭：先deactivate()应答订阅，再stop()分布式消息队列，最后停止本地队列并释放资源（Runner.java）。

DistributedRunner（DistributedRunner.java）提供了ensureStarted()惰性启动的并行入口，保证无论从哪个 API 触发分布式调用，消息队列与应答订阅都只会初始化一次。

6. 测试验证：RunnerTest 覆盖的配置行为

与DistributedConfig直接相关的验证位于RunnerTest（即文档“相关测试”一节所列）：

• configSetAndGetUseGlobalRunnerConfig：验证Runner.setConfig(config)后Runner.getConfig()返回同一实例，且envPrefix正确回读（RunnerTest.java）；
• startAndStopDefaultLocalModeReturnSuccess：验证默认配置下distributedMode=false、Runner.getDistPubsub()为null，即本地模式不会初始化分布式消息队列（RunnerTest.java）；
• startFailureClosesCheckpointerCreatedFromConfigAndFallsBackToInMemory：将messageQueueConfig.type设为不存在的"missing-mq"并开启分布式模式，验证Runner.start()抛出CompletionException——这从测试侧印证了MessageQueueFactory对未知队列类型“快速失败”的行为（RunnerTest.java）。

7. 完整配置示例（YAML 形态）

DistributedConfig的 Jackson 序列化名使其可直接嵌入 Runner 配置 YAML/JSON。以下为一份开启分布式模式、使用 Pulsar 消息队列的完整配置示意（Pulsar 扩展实现需自行引入）：

distributed_mode: true env_prefix: "prod" instance_id: "runner-a-01" # 不配置则自动生成 UUID distributed_config: request_timeout: 60.0 # 远端调用超时（秒） max_request_concurrency: 5000 # 在途并发上限 agent_topic_template: "openjiuwen.single_agent.{agent_id}.{version}" reply_topic_template: "openjiuwen.reply.runner.{instance_id}" message_queue_config: type: "pulsar" # pulsar | fake pulsar_config: url: "pulsar://localhost:6650" max_workers: 8 # Pulsar 消费者工作线程数（默认 8）

其中PulsarConfig.maxWorkers（默认8）控制 Pulsar 消费侧工作线程数；PulsarConfig的toString()/repr()会通过UrlUtils.redactUrlPassword(url)对 URL 中的密码做脱敏处理，避免敏感信息进入日志（PulsarConfig.java）。

8. 常见问题与配置建议

• 本地调试不想依赖 Pulsar：将messageQueueConfig.type设为fake（或直接使用RunnerConfig.DEFAULT_RUNNER_CONFIG），Runner 走本地内存队列，DistributedConfig的分布式参数不生效；
• 开启分布式后启动失败：优先检查 Pulsar 扩展类com.openjiuwen.extensions.message_queue.MessageQueuePulsar是否在 classpath，以及pulsar_config.url是否可达——MessageQueueFactory对这两类问题都会抛出MESSAGE_QUEUE_INITIATION_ERROR；
• 多环境共用消息队列出现消息串扰：为每个环境设置不同的envPrefix，Topic 会被自动隔离为<prefix>.<template>；
• 高并发下出现Too many collectors异常：说明在途请求超过了maxRequestConcurrency，按实例规格调大该值，或从业务侧降低并发；
• 远端 Agent 响应慢导致频繁超时：调大request_timeout，并确认ResponseCollector的 TTL 与之一致（默认同为 30.0 秒语义）。

总结

DistributedConfig是 openJiuwen agent-core-java 分布式 Runner 的“五脏六腑”：requestTimeout与maxRequestConcurrency分别从时间维度和并发维度约束分布式调用；messageQueueConfig决定底层通信载体（Pulsar 或 fake）；两条 Topic 模板配合envPrefix、instanceId组成环境隔离、实例唯一、Agent 可寻址的消息路由体系。结合RunnerConfig、Runner、DistributedRunner与RunnerTest一起阅读，即可完整掌握从配置到运行的整条链路。

• 人工智能
• AI Agent
• Agent 框架
• 工具调用
• Agent 工作流
• RAG
• MCP Clients
• 强化学习