RocketMQ 5.x 与 AI 场景（含 LiteTopic）

一句话：AI 工作流的并发单位是「会话/任务」，不是「应用」；LiteTopic 用「父 Topic 做命名空间 + 子通道按会话自动创建回收」承接海量低流量会话，普通 Topic 继续承载业务主链路。

适用版本：LiteTopic 需服务端 ≥ 5.5.0（RIP-83）、gRPC 客户端 ≥ 5.1.0；普通 Topic、Proxy、Timer 在 5.0+ 可用。文中未确认的参数以官方 Parameter limits 为准。

为什么需要

RAG 摄入、Agent 协作、推理排队都有共同形态：任务数量多、单任务流量低、任务之间要隔离、失败要能单独重试、客户端要能断线续传。

传统做法有三条路，各有硬伤：每个会话建一个普通 Topic，元数据和 ConsumeQueue 文件会爆炸；所有会话挤一个 Topic 用 sessionId 过滤，共享队列和位点，慢会话拖累别人；会话状态放应用内存，应用必须有状态，扩缩容和重连都难。

LiteTopic 把「会话」做成 Broker 管理的二级通道：自动创建、空闲过期回收、位点存在 Broker，应用可以无状态。

核心机制

LiteTopic 的模型

• 消息仍写统一 CommitLog，不为每个子通道复制消息体。
• 索引放 RocksDB（LiteCQ），创建/删除子通道是 KV 操作，不是建文件，才能支撑百万级通道。
• Ready Set 只唤醒有新消息的通道，替代对所有通道的全量轮询。
• 位点在 Broker 侧持久化，客户端断线后可按会话续传。
• 支持按通道 Suspend：主动暂停投递，到点恢复，不算消费失败，也不进重试/DLQ。

限制：每个子通道单 Queue，吞吐靠通道数横向扩展；不支持延迟、事务、优先级消息；单个通道高吞吐的场景不适用。

三个典型 AI 场景

RAG 文档摄入：上传 → doc-ingest → 解析/切分 → doc-chunk → Embedding → 向量库 → doc-indexed。每一步是独立消费组，失败可单独重试；大文档处理耗时分钟级，用异步链路避免用户同步等待。消息体带 docId、chunkId、embeddingModel，消费端按 docId + chunkId 幂等，避免重复写向量库。

Agent 工作流编排：TaskCreated → Planning → SubtaskBatch → Tool Agents → SubtaskDone → Judge → ResultReady。普通 Topic 承载任务事件，LiteTopic 承载某个用户会话的连续消息。需要超时控制时，用普通 Topic 的延迟消息（5.x Timer）做兜底，不要指望 LiteTopic 的延迟能力。事务消息只用于「本地状态 + 事件发送」的原子性，同样走普通 Topic。

推理请求削峰：API Gateway → inference-queue（普通 Topic，按模型分 Queue）→ Consumer（数量 ≈ GPU 并发槽位）→ 推理 → WebSocket/SSE 回推结果。队列的作用是削峰和限流：入队后立即告诉用户「处理中」，Consumer 按 GPU 能力消费。超时兜底用延迟消息；结果回推走长连接或普通 Topic，不要在这条链路里混入其他 MQ。

5.x 的相关能力与边界

• Proxy / gRPC：多语言客户端（Python/Go）通过无状态 Proxy 接入，适合 AI 服务栈。
• Timer：任意时间点延迟，受 timerMaxDelaySec 上限约束；用于超时检查、延迟重试。
• Pop 消费：拉取模式，消费速率更可控，适合弹性消费端。
• 分层存储：冷数据下沉对象存储，能力与版本/产品有关，以官方文档为准。
• 没有「自动扩缩 Queue」：开源 5.x 未见按负载自动增减 Queue 的机制；云产品的「弹性 TPS」是配额/规格层面的弹性，不等于自动改 Queue 数。

最小上手

Broker 侧（5.5，参数名以官方 Parameter limits 为准）：

# broker.conf
enableLmq = true
enableMultiDispatch = true
defaultRocksDB = true        # LiteTopic 的索引依赖 RocksDB ConsumeQueue

创建 LITE 父 Topic（属性名在不同 5.5 构建中可能是 message.type；先跑 mqadmin updateTopic -h 确认）：

sh bin/mqadmin updateTopic -n 127.0.0.1:9876 -c DefaultCluster \
  -t ai-session -a "+message.type=LITE,+lite.topic.expiration=3600"

lite.topic.expiration 是子通道空闲过期时间（秒），按会话时长设置；设得过大，RocksDB 里会堆积大量死通道。

gRPC 客户端（≥ 5.1.0）发送到某个会话：

Message message = provider.newMessageBuilder()
    .setTopic("ai-session").setLiteTopic("session-123")   // 不存在则自动创建
    .setBody("hello".getBytes()).build();
provider.send(message);

消费端用 LiteSubscription 按 topic + liteTopic 订阅（API 名称以 SDK 为准）。

# 父 Topic 是否存在、类型是否为 LITE
sh bin/mqadmin topicList -n 127.0.0.1:9876 | grep ai-session
sh bin/mqadmin topicStatus -n 127.0.0.1:9876 -t ai-session
# 发送后确认消息进入父 Topic 的存储与消费进度
sh bin/mqadmin consumerProgress -n 127.0.0.1:9876 -g ai-consumer

生产动作

• 按租户/业务拆父 Topic，不要把所有会话塞进一个父 Topic；配额、TTL、权限都按父 Topic 管理。
• TTL 按会话生命周期设置，并监控活跃子通道数；过期通道不回收会持续占用 RocksDB 空间。
• 用 Suspend 做主动流控，不要用消费失败来限速；Suspend 可恢复、不产生 DLQ。
• 会话 ID 要稳定且不可猜测，重连、换实例后仍能映射到同一子通道。
• 消费端必须幂等：自动重试、断线续传、Suspend 恢复都可能带来重复。
• 容量按「通道数 × 单通道 TPS」估算，不要用单通道吞吐推整体；RocksDB 的磁盘、compaction、读写延迟要单独监控。

边界

• LiteTopic 不能替代普通 Topic。它不支持延迟、事务、优先级；业务主链路、订单、支付仍用普通 Topic。
• 单通道吞吐有限。高吞吐日志、流式数据不要按会话拆 LiteTopic。
• 不是「无限 Topic」。父 Topic 数量、子通道配额、RocksDB 容量都有上限。
• 开源 5.x 没有自动扩缩 Queue；容量规划仍要人工做。
• 版本门槛明确：服务端 < 5.5 或客户端用旧 Remoting SDK，都无法使用 LiteTopic。
• 结果回推不要混用其他 MQ。用 WebSocket/SSE 或 RocketMQ 普通 Topic，保持依赖单一。

讲清楚

AI 场景和传统业务消息的区别在并发单位。传统业务按应用分 Topic，AI 按会话/任务分通道：会话数量可能上百万，每个通道流量很低，但要隔离、要能单独挂起、要能断线续传。普通 Topic 要么一个会话一个 Topic 撑爆元数据，要么共用一个 Topic 互相阻塞。

LiteTopic 的做法是父 Topic 加子通道：父 Topic 管配额和 TTL，子通道首次使用时自动创建、空闲过期回收；消息仍写统一 CommitLog，索引放 RocksDB，位点存 Broker，所以应用可以无状态。它牺牲单通道吞吐换隔离和规模，适合 RAG 摄入、Agent 会话、推理排队这类负载。

落地时记住三个边界：LiteTopic 不支持延迟和事务，超时用普通 Topic 的 Timer；单通道是单 Queue，吞吐靠通道数；开源 5.x 没有自动扩缩 Queue，容量要提前规划。业务主链路继续用普通 Topic，LiteTopic 只做会话层。