轻量主题模型

轻量主题名称

• 定义：轻量主题的名称，用于标识轻量主题，轻量主题名称在所属主题内全局唯一。

• 取值：当主题类型为Lite时，用户对message进行了setLiteTopic，如对应的轻量主题不存在，系统会自动创建。

• 约束：请参见参数限制。

过期时间

• 定义：轻量主题的过期时间，当轻量主题距离最近一次消息写入时间超过过期时间后，该轻量主题会被自动删除。删除是指释放轻量主题占用的个数，即占用总数-1。

• 取值：当创建主题类型为Lite时，可以设置过期时间 expiration 值。

• 约束：请参见参数限制。

应用场景 1：Multi-Agent 的异步通信，解决长耗时调用阻塞痛点

随着AI需求场景变得更加复杂，大多数单Agent在复杂场景中面临着局限性：缺乏专业化分工、难以对多领域进行整合；无法实现动态协作决策。单Agent应用和单Agent工作流会逐步转向 Multi-Agent 应用。但因为 AI 任务长耗时的特点，同步调用会造成调用者的线程阻塞，存在大规模协作扩展性问题。

如上图所示 Multi-Agent的工作流程为：Supervisor Agent 负责将需求拆分给两个子Agent；两个子Agent负责各自领域问题解答并将结果返回给Supervisor Agent； Supervisor Agent将结果汇总返回给Web端。使用 RocketMQ 异步通信方案流程如下：

1. 在接收请求的流程中

   1. 为每个子Agent创建一个Topic (Request)用于请求任务的缓冲队列，可以是优先级Topic，能先处理高优任务。

   2. Supervisor Agent 将拆分任务信息发送到对应的请求主题中。

2. 在返回响应结果流程中

   1. Supervisor Agent 创建 Lite 类型的Topic (Response) 并订阅这个Topic。

   2. 子 Agent 处理将每个任务的响应结果发送到Topic (Response) 的LiteTopic中。LiteTopic可以用任务ID命名，为每个任务创建一个专属的LiteTopic。

   3. Supervisor Agent 通过订阅实时获取结果，然后通过HTTP SSE协议推送给Web端。

应用场景 2：分布式会话状态管理，终结 AI 应用的会话状态管理难题

AI 应用的交互模式具有特殊性，即长耗时、多轮次且高度依赖高成本计算的会话。当应用依赖 SSE 等长连接时，一旦连接中断（如网关重启、连接超时、网络不稳定触发），不仅会导致当前会话上下文的丢失，更会直接造成已投入的 AI 任务作废，从而浪费宝贵的算力资源。

如上图所示，和场景 1的返回响应结果流程相同，使用Lite 类型的Topic作为实时通知响应结果。这里每个LiteTopic可以使用 SessionID 命名（例如 chatbot/{sessionID}），这样会话结果都作为消息在这个主题中有序传递。长连接重连后继续保持会话的连续性的解决方案如下：

1. Web端2 和应用服务节点 1 建立长链接，创建会话Session2。

2. 应用服务节点 1 监听 LiteTopic [chat/SessionID2]

3. 大模型任务调度组件，根据请求的SessionID信息，将返回结果发送到 LiteTopic [chat/SessionID2]

4. 由于网络等异常，WebSocket 自动重连到了应用服务节点2。

5. 应用服务节点 1 取消订阅LiteTopic [chat/SessionID2]，应用服务节点 2 订阅LiteTopic [chat/SessionID2]。

6. LiteTopic [chat/SessionID2] 根据之前的消费进度，将后续未消费的消息继续推送给应用服务节点 2。保证了会话状态和会话数据的连续性。

发送消息

Producer producer = provider.newProducerBuilder()
    .setTopics(topic)
    .setClientConfiguration(clientConfiguration)
    .build();

final Message message = provider.newMessageBuilder()
    .setTopic(topic)
    //设置消息索引键，可根据关键字精确查找某条消息。
    .setKeys("messageKey")
    //设置LiteTopic
    .setLiteTopic("lite-topic-1")
    //消息体
    .setBody("messageBody".getBytes())
    .build();

try {
    final SendReceipt sendReceipt = producer.send(message);
    log.info("Send message successfully, messageId={}", sendReceipt.getMessageId());
} catch (LiteTopicQuotaExceededException e) {
    // LiteTopic配额超限，评估并增加配额
    log.error("Lite topic quota exceeded", e);
} catch (Throwable t) {
    log.error("Failed to send message", t);
}

消费消息

需要使用 LitePushConsumer 消费类：

//初始化LitePushConsumer，需要绑定消费者分组ConsumerGroup、目标主题Topic、通信参数等。
LitePushConsumer litePushConsumer = provider.newLitePushConsumerBuilder()
    .setClientConfiguration(clientConfiguration)
    // 控制台创建ConsumerGroup时绑定的Topic
    .bindTopic(topicName)
    //设置消费者分组
    .setConsumerGroup(consumerGroup)
    .setMessageListener(messageView -> {
        //处理消息并返回消费结果。
        LOGGER.info("Consume message={}", messageView);
        return ConsumeResult.SUCCESS;
    })
    .build();

try {
    // 订阅感兴趣的LiteTopic集合
    litePushConsumer.subscribeLite("lite-topic-1");
    litePushConsumer.subscribeLite("lite-topic-2");
    litePushConsumer.subscribeLite("lite-topic-3");
} catch (LiteSubscriptionQuotaExceededException e) {
    // LiteTopic配额超限，评估并增加配额
    log.error("Lite subscription quota exceeded", e);
} catch (Throwable t) {
    log.error("Failed to subscribe lite topic", t);
}

// 业务处理完毕后，及时取消不再使用的LiteTopic订阅
litePushConsumer.unsubscribeLite("lite-topic-3");

// 获取当前订阅的LiteTopic集合
Set<String> liteTopicSet = litePushConsumer.getLiteTopicSet();

动态修改订阅关系

/**
 * 动态添加订阅关系
 * subscribeLite() 方法会发起网络请求并执行配额验证，因此可能会调用失败。
 * 请务必检查此调用的结果，以确保订阅被成功添加。
 * 可能的失败场景包括：
 * 1. 网络请求错误，可以重试。
 * 2. 配额验证失败，抛出 LiteSubscriptionQuotaExceededException 异常。
 * 请评估配额是否满足需求，并及时使用 unsubscribeLite() 取消订阅不再使用的主题以释放资源。
 */

litePushConsumer.subscribeLite("lite-topic-1");

//动态删除订阅关系
litePushConsumer.unsubscribeLite("lite-topic-1");

Serverless系列实例

非Serverless系列（包年包月、按量付费）实例