QueueMessage 消息载体详解：openJiuwen agent-core-java 消息队列的通用消息模型

QueueMessage 是 openJiuwen agent-core-java 中消息队列通信的通用消息载体，统一封装消息 ID、业务载荷与错误信息，是 runner 模块异步通信（invoke 请求-响应与流式 streaming）的数据基座。本文围绕 QueueMessage.java 与其继承体系、队列路由实现和测试用例，讲清该载体的字段语义、序列化约定、错误处理机制，以及如何基于它构建可落地的消息收发代码。

一、类定义与定位

package com.openjiuwen.core.runner.mq;

@JsonNaming(PropertyNamingStrategies.SnakeCaseStrategy.class)
public class QueueMessage

QueueMessage 是消息队列通用消息载体，封装 messageId、payload、errorCode 与 errorMsg 四个核心属性，源码位于 core/src/main/java/com/openjiuwen/core/runner/mq/QueueMessage.java。从源码注释看，它是对 Python 侧 openjiuwen/core/runner/message_queue_base.py 中 QueueMessage 的 Java 镜像实现，保证了多语言实现间消息结构的一致性。

它在 runner 消息体系中的定位可概括为三点：

1. 消息投递的统一入参：MessageQueueBase.produceMessage(String topic, QueueMessage message) 要求所有投递到队列的消息都以 QueueMessage（或其子类）作为载体，见 MessageQueueBase.java。
2. 两类通信模式的基类：InvokeQueueMessage（请求-响应）与 StreamQueueMessage（流式迭代）均继承自它，并在此基础上扩展各自的 CompletableFuture 响应通道。
3. 分布式消息的公共父类：分布式 runner 的消息 DmqMessage 同样继承 QueueMessage，并将 payload 语义重定义为 body，见 DmqMessage.java。

二、字段详解与序列化约定

原文档给出的字段表如下：

字段	类型	默认值	说明
messageId	String	""	消息唯一标识
payload	Object	-	业务载荷，可为任意对象
errorCode	int	0	错误码，0 表示成功
errorMsg	String	""	错误信息描述

结合源码补充三个重要细节：

（1）errorCode 的默认值并非普通字面量 0，而是成功状态码常量。 源码中 errorCode = StatusCode.SUCCESS.getCode()，即 0（对应 "success"），参见 StatusCode.java。这意味着：一个新构造的 QueueMessage 天然处于“成功”语义，只有在消费失败时才被填充错误码。

（2）序列化采用蛇形命名（snake_case）。 类级注解 @JsonNaming(PropertyNamingStrategies.SnakeCaseStrategy.class) 使 Jackson 在序列化/反序列化时将驼峰字段转为蛇形：messageId ↔ message_id、errorCode ↔ error_code、errorMsg ↔ error_msg。这与 Python 侧实现命名风格对齐，便于跨语言（Java ↔ Python）消息互认。

（3）payload 是 Object 类型。 不限制具体业务类型，字符串、Map、领域对象均可直接承载，由订阅方在消费时自行解析。

三、构造方法与读写方法

原文档列出的 API 全部存在，源码逐一印证：

签名	说明
public QueueMessage()	无参构造，messageId 与 errorMsg 为空串，errorCode 为 0
public QueueMessage(String messageId, Object payload)	一次指定消息 ID 与业务载荷

方法签名	说明
public String getMessageId() / setMessageId(String)	读写消息唯一标识
public Object getPayload() / setPayload(Object)	读写业务载荷
public int getErrorCode() / setErrorCode(int)	读写错误码
public String getErrorMsg() / setErrorMsg(String)	读写错误描述

一个典型的构造与读取示例：

// 构造消息
QueueMessage message = new QueueMessage("msg-001", Map.of("task", "codegen"));
message.setErrorCode(0);
message.setErrorMsg("");

// 读取消息
String messageId = message.getMessageId();   // "msg-001"
Object payload = message.getPayload();       // 业务载荷
int code = message.getErrorCode();           // 0 表示成功

四、继承体系：从通用载体到两种通信模式

QueueMessage 之所以设计为“通用载体”，是因为它作为基类被两种异步通信模式复用：

• InvokeQueueMessage（请求-响应）：在 InvokeQueueMessage.java 中扩展 @JsonIgnore 标注的 CompletableFuture<Object> response，通过 getResponse() 异步等待单个结果。
• StreamQueueMessage（流式输出）：在 StreamQueueMessage.java 中扩展 @JsonIgnore 标注的 CompletableFuture<Iterator<Object>> response，通过 getResponse() 异步等待迭代器，逐项消费流式结果。

两者都保留 QueueMessage 的四个基类字段，因此 messageId、payload、errorCode、errorMsg 在两种模式下语义完全一致；response 标记 @JsonIgnore，避免响应通道在消息序列化传输时被意外携带。

五、底层路由与错误填充机制

在内存队列实现 MessageQueueInMemory.java 与订阅实现 SubscriptionInMemory.java 中，QueueMessage 贯穿投递、路由、消费全流程：

1. 投递：produceMessage(topic, message) 将消息连同 topic 包装为 TopicMessage 放入有界阻塞队列（默认容量 10,000）。
2. 路由：后台单线程消费者取出消息，按 topic 找到对应订阅，调用 subscription.pushMessage(message)。
3. 自动补 ID：pushMessage 中若 messageId 为空，会自动填充 UUID.randomUUID().toString()，保证每条消息都有唯一标识（见 SubscriptionInMemory.java）。
4. 消费与回填错误：订阅处理结果分发给消息时——成功则 complete(response)；失败则回填 errorCode/errorMsg，并以 completeExceptionally 结束响应 Future。非 BaseError 类型的异常统一标记为 StatusCode.MESSAGE_QUEUE_MESSAGE_CONSUME_ERROR（错误码 110212，"consume message error"），参见 StatusCode.java。

六、测试用例验证：行为即契约

原文档指出的相关测试 MessageQueueInMemoryTest.java 从行为层面验证了 QueueMessage 体系的关键约定：

• 响应类型契约：InvokeQueueMessage 收到 Iterator 响应会抛 IllegalArgumentException 并回填消费错误码 110212；StreamQueueMessage 则强制要求 Iterator 响应。
• Python falsy 兼容：InvokeQueueMessage 的空串、空集合、空 Map、0 数值等“假值”响应会被视为消费失败，错误信息包含 "response is empty"。
• 嵌套发送不死锁：在订阅处理回调内再次 produceMessage 可正常完成多级调用链（如 A(B(C))），验证了异步消费模型对递归投递的支撑。
• 重复订阅拦截：同一 topic 二次 subscribe 抛出 "Topic 'topic' is already subscribed."。
• 缺 ID 自动补全：空 messageId 投递后消费成功且 ID 被回填。

这些测试不仅保护了队列本身，也固化了 QueueMessage 及其子类在“成功/失败/类型/ID”四个维度上的对外契约，是阅读该载体行为语义的最佳起点。

七、实际使用入口

QueueMessage 体系的真实使用场景集中在 runner 模块：

• 本地 pub-sub 透出：Runner.pubsub() 返回 LocalMessageQueue（no-op 存根），见 Runner.java 与 LocalMessageQueue.java。
• 分布式 runner：MessageQueueFactory.create(MessageQueueConfig) 按配置类型（默认 PULSAR，见 MessageQueueConfig.java）创建真实 MessageQueueBase，远程客户端 MqRemoteClient 通过 mq.produceMessage(topic, ...) 投递请求消息，见 MessageQueueFactory.java。

如需在本地快速验证消息收发，可按 MessageQueueInMemoryTest.java 的模式编写基于 MessageQueueInMemory + SubscriptionInMemory + InvokeQueueMessage/StreamQueueMessage 的最小示例：start() 启动队列、subscribe(topic) 注册订阅并 activate()、produceMessage 投递、getResponse().get(timeout, unit) 等待结果，最后在 finally 中 stop() 释放资源。

小结

QueueMessage 以极简的四个字段（messageId/payload/errorCode/errorMsg）承载了 openJiuwen agent-core-java 消息体系中最核心的语义约定：成功即默认（errorCode=0）、失败必回填（消费错误码 110212）、缺 ID 自动补全、蛇形命名跨语言序列化。它向下被 invoke/stream 两类消息子类复用，向上支撑本地与分布式 runner 的异步通信，是理解整个 runner 消息队列机制的入口类型。