openJiuwen agent-core-java 本地消息队列解析：LocalMessageQueue 占位实现的设计与定位

LocalMessageQueue 是 openJiuwen agent-core-java 中 runner 运行时提供的"本地消息队列占位实现"：它不承载任何消息收发能力，start() 与 stop() 一律直接返回 true，在进程内以最小成本充当全局发布/订阅（pub/sub）入口的默认形态。阅读本文后，你将掌握该占位类在 com.openjiuwen.core.runner.mq 消息队列体系中的位置、它与内存队列、分布式队列的关系，以及如何在实际项目中按需替换为真正的消息队列实现。

LocalMessageQueue 是什么

LocalMessageQueue 是 com.openjiuwen.core.runner.mq 包中定义的一个公开类，按照 API 文档 的说明，它是"本地消息队列占位实现"，核心语义如下：

public class LocalMessageQueue

• start() 与 stop() 均直接返回 true；
• 不提供主题订阅、取消订阅、消息投递等额外队列能力。

从命名与行为可以推断，它对应 Python 版本 openjiuwen/core/runner/message_queue_base.py 中的同名 LocalMessageQueue，Java 侧作为 No-op stub 落地，见 LocalMessageQueue.java 的类注释。

方法签名一览

签名	说明
public boolean start()	占位启动：不创建任何线程或资源，直接返回 true
public boolean stop()	占位停止：不释放任何资源，直接返回 true

注意一个细节：占位类的 start()/stop() 返回 boolean，而 MessageQueueBase 抽象契约中的 start()/stop() 返回 void。这说明 LocalMessageQueue 并不继承 MessageQueueBase，它是独立存在的、面向"本地单进程"场景的最小占位形态，刻意规避了完整的 pub/sub 契约负担。

源码实现剖析：一个真正的 No-op 类

仓库中的真实实现非常简洁，完整代码如下（LocalMessageQueue.java）：

public class LocalMessageQueue {

    public boolean start() {
        return true;
    }

    public boolean stop() {
        return true;
    }
}

几点值得注意的实现事实：

• 类上没有 extends MessageQueueBase，也没有实现任何接口，因此它不暴露 subscribe、unsubscribe、produceMessage 等消息能力；
• 没有内部状态字段，start()/stop() 是无副作用的纯操作，调用多少次结果都恒定；
• 由源码结构可以推断，它的定位是"占位/桩"，让 Runner 在未配置分布式消息队列时仍能提供统一的 pubsub() 访问入口，而不必抛出空指针或引入额外依赖。

在 Runner 全局运行时中的角色

LocalMessageQueue 最典型的使用位置是 Runner 全局单例。在 Runner.java 中，它通过静态门面暴露给上层：

public static final LocalMessageQueue pubsub = GLOBAL_RUNNER.pubsub();
// ...
public static LocalMessageQueue getPubsub() {
    return GLOBAL_RUNNER.pubsub();
}
public static LocalMessageQueue pubsub() {
    return getPubsub();
}

其底层实现位于内部类 RunnerImpl 中（Runner.java）：

private final LocalMessageQueue messageQueue = new LocalMessageQueue();
// ...
private LocalMessageQueue pubsub() {
    return messageQueue;
}

从上述代码可以看出，Runner 的进程内 pubsub 默认就是一个 LocalMessageQueue 实例——它由全局 Runner 持有、与进程生命周期绑定，作为本地消息入口的兜底实现。与此同时，RunnerImpl 还维护了一个独立的 distributedMessageQueue 字段（MessageQueueBase 类型），通过 distPubsub() 暴露，用于承接跨进程的分布式消息能力（Runner.java）。

这揭示了一个清晰的分层设计：

• 本地层：Runner.pubsub() → LocalMessageQueue，占位、无消息能力；
• 分布式层：Runner.getDistPubsub() → 分布式消息队列（如 Pulsar 或 Fake 实现），承载真正的跨进程 pub/sub。

与 mq 包其他类型的对照

LocalMessageQueue 只是 com.openjiuwen.core.runner.mq 包的一角。整个包围绕"本地消息队列抽象、内存实现以及请求/流式消息封装"组织，类型关系如下：

类型	角色
MessageQueueBase	定义 start/stop/subscribe/unsubscribe/produceMessage 的抽象队列契约
MessageQueueInMemory	基于内存主题路由的真实实现，使用阻塞队列与单线程消费者
SubscriptionBase	订阅抽象基类，管理消息处理器绑定与激活状态
SubscriptionInMemory	基于阻塞队列的内存订阅实现，激活后用后台线程消费
QueueMessage	通用消息载体，封装消息 ID、载荷与错误码/错误信息
InvokeQueueMessage	请求-响应模式消息，额外携带 CompletableFuture<Object> 响应句柄
StreamQueueMessage	流式响应模式消息，额外携带 CompletableFuture<Iterator<Object>> 响应句柄
AsyncMessageHandler	异步消息处理器函数式接口：handle(T) -> CompletableFuture<R>
LocalMessageQueue	占位实现，start()/stop() 直接返回 true

与 MessageQueueInMemory 的实质差异

LocalMessageQueue 与 MessageQueueInMemory 形成了"占位 vs 可用"的对照。后者继承了 MessageQueueBase，具备完整能力（MessageQueueInMemory.java）：

• 构造时默认 queueMaxSize = 10_000、timeout = 120 秒，支持无参构造与显式传参构造；
• start() 使用 OpenJiuwenExecutors.newSingleThreadExecutor("mq-inmemory", true) 启动后台消费者线程；
• subscribe(topic) 为每个主题创建 SubscriptionInMemory，重复订阅同一主题会抛出 IllegalArgumentException("Topic ... is already subscribed.")；
• produceMessage(topic, message) 将 TopicMessage 放入有界阻塞队列，由消费者线程按主题路由到对应订阅。

因此，如果业务确实需要在进程内做消息订阅/投递，应使用 MessageQueueInMemory；LocalMessageQueue 只承担"占位"职责。

相关测试与契约验证

原 API 文档标注的相关测试为 RunnerTest；在当前仓库中，对 LocalMessageQueue 行为的直接验证位于测试类 MessageQueueBaseTypesTest.java：

@Test
@DisplayName("local message queue start and stop return true")
void testLocalMessageQueueLifecycle() {
    LocalMessageQueue queue = new LocalMessageQueue();

    assertTrue(queue.start());
    assertTrue(queue.stop());
}

该测试以 assertTrue 断言 start()/stop() 恒为 true，与类注释和文档描述完全一致。测试类注释同时说明，其覆盖的契约面镜像自 Python 侧 tests/unit_tests/core/runner/test_message_queue.py，印证了 LocalMessageQueue 是跨语言一致的占位语义。

同文件中其他测试还覆盖了 mq 包的关键契约，可作为扩展阅读：

• testQueueMessageDefaults：验证 QueueMessage 默认值（messageId=""、errorCode=StatusCode.SUCCESS、errorMsg=""、payload=null）；
• testInvokeQueueMessageInitializesResponseFuture：验证 InvokeQueueMessage 初始化响应 Future；
• testStreamQueueMessageInitializesIteratorResponseFuture：验证 StreamQueueMessage 以 Iterator 完成响应；
• testMessageQueueBaseSubclassContract：验证 MessageQueueBase 五个抽象方法的子类可接线性。

实际使用建议

何时使用 LocalMessageQueue

• 单进程、无跨进程消息诉求：进程内 Runner.pubsub() 默认可直接使用，无需任何配置；
• 作为 Runner 启动流程的兜底：无论是否配置分布式队列，pubsub() 访问都不会失败；
• 替换占位实现：需要真实本地消息能力时，改用 MessageQueueInMemory；需要跨进程能力时，通过 Runner 配置接入分布式消息队列。

接入分布式消息队列的入口

Runner 的分布式消息队列类型由 MessageQueueType 枚举定义，目前支持两种取值：

• PULSAR("pulsar")：Apache Pulsar 分布式消息队列；
• FAKE("fake")：本地假队列，用于测试或占位。

配置对象为 MessageQueueConfig，其中 type 字段默认值为 MessageQueueType.PULSAR.getValue()，并携带 pulsarConfig 用于 Pulsar 连接参数。也就是说，默认配置下 Runner 倾向使用 Pulsar 作为分布式消息通道，而 LocalMessageQueue 只服务于进程内的占位 pub/sub 入口。

小结

LocalMessageQueue 是 openJiuwen agent-core-java runner 体系中一个刻意"简化到极致"的组件：它以两个恒为 true 的方法完成生命周期占位，既保证了 Runner.pubsub() 在任意配置下都可安全访问，又通过不继承 MessageQueueBase 的方式划清了"占位"与"真实队列"的边界。理解它的设计意图，有助于你在阅读 MessageQueueInMemory 与分布式消息队列接入时，快速分清本地占位、内存实现与跨进程通道三者之间的层次关系。