openJiuwen agent-core-java 消息队列流式响应：StreamQueueMessage 设计与使用全解析

StreamQueueMessage 是 openJiuwen agent-core-java（com.openjiuwen.core.runner.mq 包）中面向流式响应模式的消息载体，它在通用消息基类 QueueMessage 之上额外持有一个 CompletableFuture<Iterator<Object>> 响应对象，用于承载按迭代器逐条产出（而非一次性返回单个值）的异步结果。本文以官方 API 文档 StreamQueueMessage.md 为主线，结合仓库源码与测试用例，深入讲解该类的字段语义、构造方式、底层流转机制、类型约束与最佳实践，帮助你完整掌握在 openJiuwen 消息队列体系中实现流式响应的正确姿势。

一、StreamQueueMessage 的定位：流式响应的消息载体

openJiuwen 的 runner 消息队列模块（core/src/main/java/com/openjiuwen/core/runner/mq/）围绕"消息"抽象了两种响应模式：

消息类型	响应对象类型	适用场景
InvokeQueueMessage	CompletableFuture<Object>	请求-响应（request-response）模式，消费端一次性返回单个结果
StreamQueueMessage	CompletableFuture<Iterator<Object>>	流式（streaming / iterator）模式，消费端以迭代器逐条产出结果

StreamQueueMessage 的类声明如下（与 Python 版 openjiuwen/core/runner/message_queue_base.py 中的 StreamQueueMessage 一一对应）：

public class StreamQueueMessage extends QueueMessage

从源码 StreamQueueMessage.java 可以看到，它的核心职责是：在通用消息属性（messageId、payload、errorCode、errorMsg）之外，为流式响应单独提供一个可异步完成的 CompletableFuture<Iterator<Object>>。当消费端处理完消息后，通过向这个 Future 写入一个迭代器（或异常）来通知生产端流式结果就绪。

二、字段解析：response 的语义与序列化行为

文档中 StreamQueueMessage 仅有一个新增字段：

字段	类型	默认值	说明
response	CompletableFuture<Iterator<Object>>	new CompletableFuture<>()	流式响应对象，保存消费端返回的迭代器

对应源码 StreamQueueMessage.java：

@JsonIgnore
private final CompletableFuture<Iterator<Object>> response = new CompletableFuture<>();

这里有三点值得注意：

1. final 修饰 + 字段初始化：response 在对象创建时即为一个全新的、尚未完成的 CompletableFuture，不允许被替换。这保证了每个消息对象自始至终只有一个响应 Future，生产端可以安全地对该 Future 进行 complete(...) 或 completeExceptionally(...)。
2. @JsonIgnore 注解：响应 Future 是运行期的内部状态，不应随消息一起参与 Jackson 序列化（例如消息落盘、跨进程传输时）。因此该字段被显式排除在 JSON 之外。这与基类 QueueMessage 上的 @JsonNaming(PropertyNamingStrategies.SnakeCaseStrategy.class)（见 QueueMessage.java）配合，确保真正需要传输的消息体是干净、可序列化的。
3. 泛型设计：Iterator<Object> 而非具体类型，意味着流式响应中的每个元素可以是任意对象（如分块文本、事件对象、结构化数据片段），由消费端自行决定产出类型。

三、继承自 QueueMessage 的通用消息属性

StreamQueueMessage 直接继承了基类 QueueMessage 的四个通用字段，任何 StreamQueueMessage 实例天然具备这些属性（源码见 QueueMessage.java）：

字段	类型	默认值	说明
messageId	String	""	消息唯一标识；由内存订阅在投递时自动补充 UUID（见下文第六节）
payload	Object	null	消息携带的业务数据
errorCode	int	StatusCode.SUCCESS.getCode()（即 0）	消费结果错误码，成功时为 0
errorMsg	String	""	消费结果错误描述，成功时为空字符串

基类为每个字段提供了配套的 getter / setter（getMessageId/setMessageId、getPayload/setPayload、getErrorCode/setErrorCode、getErrorMsg/setErrorMsg）。这些默认值被测试 MessageQueueBaseTypesTest.java 明确锁定：新建消息的 messageId 为空串、errorCode 为成功码、errorMsg 为空串、payload 为 null。

因此，一个 StreamQueueMessage 实例实际上是"通用消息信封 + 流式响应句柄"的组合：发送方通过 payload 传递请求数据，接收方通过 response Future 回传迭代器结果，并通过 errorCode / errorMsg 反馈处理成败。

四、构造方法：两种创建方式

StreamQueueMessage 提供两个构造方法：

签名	说明
public StreamQueueMessage()	无参构造，messageId 与 payload 保持基类默认值，适用于随后通过 setter 填充的场景
public StreamQueueMessage(String messageId, Object payload)	带标识与载荷构造，直接指定消息 ID 与业务数据

源码 StreamQueueMessage.java 中，带参构造直接委托基类：

public StreamQueueMessage(String messageId, Object payload) {
    super(messageId, payload);
}

两种构造方式下，response Future 都会在对象创建时自动初始化。测试 MessageQueueBaseTypesTest.java 验证了带参构造的行为：创建 new StreamQueueMessage("stream-1", "payload") 后，messageId 与 payload 被正确赋值，getResponse() 返回的 Future 初始未完成（isDone() 为 false），随后可以主动 complete(iterator) 并从中取回同一个迭代器实例（assertSame）。

五、getResponse()：异步获取流式结果

StreamQueueMessage 唯一新增的方法为：

public CompletableFuture<Iterator<Object>> getResponse()

该方法直接返回内部持有的 response Future，调用方可以：

• 同步阻塞等待结果：message.getResponse().get(timeout, TimeUnit.SECONDS) 拿到迭代器，再通过迭代器逐条消费；
• 异步编排：使用 thenApply、whenComplete、thenAccept 等 CompletableFuture 组合手段，在结果就绪后继续处理；
• 感知失败：若消费过程抛出异常，Future 将以 completeExceptionally 结束，get() 会抛出 ExecutionException，其 cause 为消费端产生的异常。

需要强调：这个 Future 的完成时机不由消息本身控制，而是由消息队列消费链路在后台异步写入（详见下一节）。生产端创建消息后，只管将其投递到队列，随后通过 getResponse() 等待消费端把流式结果"回填"进来。

六、底层流转机制：从 produceMessage 到 response 完成

StreamQueueMessage 的响应不是凭空产生的，它依赖 runner 消息队列模块的完整消费链路。以内存实现为例，整条链路为：

生产者 produceMessage(topic, StreamQueueMessage)
    └─> MessageQueueInMemory 内部阻塞队列 + 单线程消费者
            └─> 按 topic 路由到对应 SubscriptionInMemory（须已 activate）
                    └─> pushMessage（空 messageId 自动补 UUID）
                            └─> 订阅消费者调用 AsyncMessageHandler.handle(payload)
                                    └─> 返回 CompletableFuture<Object>
                                            └─> handleResponse：类型校验后
                                                    complete(iterator) 或 completeExceptionally

关键节点对应源码如下：

1. 队列路由：MessageQueueInMemory（MessageQueueInMemory.java）通过 produceMessage(topic, message) 将消息封装为 TopicMessage 放入有界阻塞队列，单线程消费者取出后按 topic 找到已激活的 SubscriptionInMemory 并投递。
2. 订阅消费：SubscriptionInMemory（SubscriptionInMemory.java）的消费循环取出消息，调用通过 setMessageHandler(AsyncMessageHandler) 注册的处理器——该处理器是函数式接口（AsyncMessageHandler.java）：

@FunctionalInterface
public interface AsyncMessageHandler<T, R> {
    CompletableFuture<R> handle(T message);
}

处理器返回的 CompletableFuture 会挂上 orTimeout 超时保护（默认 120_000 秒，可在构造 MessageQueueInMemory / SubscriptionInMemory 时传入 Duration 调整）。

3. 结果回填：SubscriptionInMemory.handleResponse（SubscriptionInMemory.java）按消息类型分发——对 StreamQueueMessage，将处理器结果强制转为 Iterator<Object> 后 complete 到 getResponse() 上：

if (message instanceof StreamQueueMessage streamQueueMessage) {
    @SuppressWarnings("unchecked")
    Iterator<Object> iterator = (Iterator<Object>) requireIterator(response);
    streamQueueMessage.getResponse().complete(iterator);
}

4. 异常回填：若处理器抛出异常、返回 null、或类型校验失败，failMessage（SubscriptionInMemory.java）会将 errorCode / errorMsg 写入消息，并在 response Future 尚未完成时对其 completeExceptionally，让生产端能够感知消费失败。

这一"投递消息 → 异步处理 → Future 回填"的闭环，使 StreamQueueMessage 真正具备流式语义：消息一经投递即返回控制权，结果以迭代器的形式在未来某个时刻就绪。

七、流式响应的类型约束与校验规则

消费端回填到 StreamQueueMessage 的结果必须是 Iterator 类型，这是框架层的硬性校验。requireIterator（SubscriptionInMemory.java）给出了两条规则：

• 结果为 null 时抛 IllegalArgumentException("response is empty")；
• 结果不是 Iterator 时抛 IllegalArgumentException("StreamQueueMessage need Iterator response")。

与之相对，InvokeQueueMessage 走另一套规则（SubscriptionInMemory.java）：结果不能是 Iterator，且不能是 Python 语义下的"falsy"值（null、false、数值 0、空字符串、空集合、空 Map，见 isPythonFalsy）。若在流式 topic 上发送 InvokeQueueMessage，或流式处理器返回非迭代器结果，都会触发消费错误。

测试 MessageQueueInMemoryTest.java 的 recordsWrongResponseTypeAsConsumeError 用例专门验证了这一点：订阅处理器返回 CompletableFutureFactory.completedIterator("item")（迭代器），而发送方用的是 InvokeQueueMessage，最终 getResponse().get(...) 抛出 ExecutionException，底层 cause 为 IllegalArgumentException，且消息 errorCode 被置为 StatusCode.MESSAGE_QUEUE_MESSAGE_CONSUME_ERROR。

实践建议：流式消费端应使用 List.of(...).iterator()、Stream.iterator() 或自定义迭代器作为处理器返回值；同时，同一个 topic 上的消息类型应保持一致，避免流式/请求-响应混用。

八、完整可运行示例：流式响应的端到端使用

结合内存队列 MessageQueueInMemory，一个可运行的端到端流式示例如下（参考 MessageQueueInMemoryTest.java 的 routesInvokeAndStreamMessages 用例改造）：

import com.openjiuwen.core.runner.mq.MessageQueueInMemory;
import com.openjiuwen.core.runner.mq.StreamQueueMessage;
import com.openjiuwen.core.runner.mq.SubscriptionInMemory;

import java.time.Duration;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;

public class StreamQueueMessageExample {

    public static void main(String[] args) throws Exception {
        // 1. 创建并启动内存消息队列（队列容量 100，消费超时 5 秒）
        MessageQueueInMemory queue = new MessageQueueInMemory(100, Duration.ofSeconds(5));
        queue.start();
        try {
            // 2. 订阅主题 topic_stream，注册流式处理器
            SubscriptionInMemory subscription = queue.subscribe("topic_stream");
            subscription.setMessageHandler(payload ->
                    CompletableFuture.completedFuture(
                            List.of("stream response for " + payload, "stream done").iterator()));
            subscription.activate();

            // 3. 构造并投递 StreamQueueMessage
            StreamQueueMessage streamMessage = new StreamQueueMessage("stream-1", "payload-a");
            queue.produceMessage("topic_stream", streamMessage);

            // 4. 异步等待迭代器结果，逐条消费
            var iterator = streamMessage.getResponse().get(1, TimeUnit.SECONDS);
            while (iterator.hasNext()) {
                System.out.println(iterator.next());
            }
            // 5. 校验消费成功状态
            System.out.println("errorCode = " + streamMessage.getErrorCode());
            System.out.println("errorMsg = " + streamMessage.getErrorMsg());
        } finally {
            queue.stop();
        }
    }
}

运行输出为：

stream response for payload-a
stream done
errorCode = 0
errorMsg = 

要点回顾：投递前必须先 subscribe(topic) 并 activate() 订阅（重复订阅同一 topic 会抛 IllegalArgumentException("Topic '...' is already subscribed.")，见 MessageQueueInMemory.java）；投递时可留空 messageId，订阅投递阶段会自动补充 UUID（SubscriptionInMemory.java）。

九、相关测试：行为契约的证据链

文档指向的 MessageQueueInMemoryTest 与同类测试共同锁定了 StreamQueueMessage 的行为契约：

• MessageQueueInMemoryTest.java：routesInvokeAndStreamMessages 验证流式消息的完整链路——投递 StreamQueueMessage("stream-1", "payload-a") 后，getResponse().get(1, TimeUnit.SECONDS) 返回的迭代器依次产出 "stream response for payload-a"、"stream done"，且消息 errorCode 为成功码、errorMsg 为空；
• MessageQueueBaseTypesTest.java：testStreamQueueMessageInitializesResponseFuture 验证消息初始化即拥有未完成的响应 Future，且可被 complete(iterator) 后通过 get() 原样取回；
• 同一测试文件还覆盖了基类默认值、SubscriptionBase 惰性生命周期、AsyncMessageHandler 契约与 MessageQueueBase 子类接线等基础面，可作为理解整个 mq 包的入口。

十、注意事项与最佳实践

1. 类型匹配：流式 topic 的处理器返回值必须是 Iterator；StreamQueueMessage 与 InvokeQueueMessage 不要在同一 topic 上混用，否则会触发 MESSAGE_QUEUE_MESSAGE_CONSUME_ERROR。
2. 异步等待：response Future 由后台消费者线程完成，生产端若需同步等待，务必使用 get(timeout, TimeUnit) 形式设置超时，避免无限阻塞。
3. 错误感知：消费异常会同时体现在两处——消息的 errorCode / errorMsg 字段，以及 response Future 的异常完成。两者配合可定位"消费失败 + 失败原因"。
4. 序列化边界：response 被 @JsonIgnore 排除，流式 Future 是进程内句柄，不要期望它随消息跨进程传输；跨进程场景应自行在 payload 中设计流式协议（如分块消息序列）。
5. 生命周期管理：使用内存队列时记得 queue.stop() 释放消费者线程；订阅不再使用时调用 unsubscribe(topic) 触发 deactivate()。

综上，StreamQueueMessage 是 openJiuwen runner 消息队列中面向流式输出的一等公民消息类型：它以"通用消息信封 + 迭代器响应 Future"的结构，将流式结果的异步回填、类型校验与错误透传封装在一个类中。掌握它，你就掌握了在 openJiuwen 多智能体协作与异步任务框架中传递分块、增量结果的正确姿势。