Redis Streams 深度剖析¶

为什么选择 Redis Streams¶

mindmap
  root((Redis Streams))
    持久化
      ["消息落盘不丢失"]
      ["AOF / RDB 持久化"]
    消费组
      ["Consumer Group 竞争消费"]
      ["自动负载均衡"]
    范围查询
      ["按 ID 范围读取历史"]
      ["支持 XRANGE / XREVRANGE"]
    ACK 机制
      ["PEL 待确认列表"]
      ["未 ACK 自动重投"]
    高性能
      ["单分区 O(1) 写入"]
      ["支持批量读取"]

Key Schema 版本：v1 / v2（单机 / 集群兼容）¶

本页下面「完整 Redis Key 参考」列出的 key 都是默认的 v1 格式——不带任何 Redis Cluster hash tag，只在单机 Redis 下保证正确。切到 Redis Cluster 部署（参见部署指南 · 集群部署）时，SDK 会切换成 v2 格式：给同一实体（如同一个 session_id / worker_id / trace_id）下的相关 key 统一加上 hash tag，保证它们落在同一个 slot，多 key 操作（MULTI/EXEC、Lua 脚本等）才不会触发 CROSSSLOT 错误。

为什么 v1 格式在 Cluster 下会出问题¶

Redis Cluster 把 keyspace 切成 16384 个 slot，一条命令只能在一个 slot 上运行；跨 slot 的多 key 事务会被直接拒绝。 v1 格式的 key（比如 byai_gateway:registry:worker:online:{worker_id}，这里的花括号只是本文档用来表示"实际值填在这里" 的占位符写法，不是 Redis key 里的字面字符）落在哪个 slot 完全由 key 的完整字符串哈希决定，同一个 worker 下的多个相关 key（在线租约、执行记录、心跳统计……）大概率会散落在不同 slot 上。只要业务代码里有把它们放进同一个 pipeline / 事务的路径，就会在 Cluster 下报 CROSSSLOT Keys in request don't hash to the same slot。

v2 格式：用 hash tag 把同一实体的 key 锁进同一个 slot¶

v2 格式给 key 加上统一前缀 byai_gateway:v2:，并且给"同一实体"（same-entity）分组的 key 用 { } 包住共同的 ID 部分—— Redis 计算 slot 时只看花括号里的内容，两个 key 只要花括号内容相同就必定落在同一个 slot。以 worker_online_lease 为例：

Key Schema	实际 key（`worker_id="worker-07"`）
v1（默认）	`byai_gateway:registry:worker:online:worker-07`
v2（Cluster）	`byai_gateway:v2:registry:worker:{worker-07}:online`

三语言的 key 工厂方法（Python RedisKeys.* / TS QueueNames.*、RegistryKeys.* / Java Constants.*）内部已经把这条拼装规则统一收口，业务代码永远不应该手写字面量拼 Redis key——一是容易在两种 schema 下漏加/错加 hash tag，二是升级到 v2 之后手写的字面量会一直指向旧的 v1 key，SDK 实际写入的是 v2 key，读永远读到空数据且不会报错，非常难排查。

不是所有 key 都需要 hash tag¶

只有"同一实体下有多个相关 key、且确实存在把它们放进同一个 pipeline/事务的代码路径"才需要 hash tag（上表归类为 same-entity）；只有一个变量维度、从不与其他 key 组合操作的 key（single-key，比如 agent_type 控制流）在 v2 下只加统一前缀，不需要额外的花括号分组。判断某个自定义 key 要不要加 hash tag，看的是"它会不会和别的 key 一起出现在同一次 MULTI/EXEC 或 Lua 脚本里"，而不是"它属于哪个功能域"。

如何启用¶

设置 REDIS_CLUSTER_HOST 即可——SDK 会同时把部署形态切到 cluster、把 key schema 切到 v2，不需要分别配置。如果只是想在单机部署下先验证 v2 key 格式本身是否有问题（不切集群），可以单独显式设置 REDIS_KEY_SCHEMA_VERSION=v2。详见部署指南 · 集群部署。

v1 → v2 是唯一涉及存量数据的破坏性变更

v2 格式落地后，同一批 key 的物理名字全部改变，旧 key 下的历史数据在新代码里读不到。多数 key 带 TTL，部署新版本后旧 key 自然过期即可；只有极少数无 TTL 的 key 需要一次性迁移脚本。升级前务必先在测试环境走一遍完整流程。

完整 Redis Key 参考¶

Stream Keys（消息队列）¶

Key 模式	类型	用途
`byai_gateway:ctrl:agent_type:{agent_type}`	Stream	Agent type 控制流，Worker 竞争消费
`byai_gateway:ctrl:worker:{worker_id}`	Stream	Worker 定向控制流，单 Worker 消费
`byai_gateway:session:{session_id}:data_stream`	Stream	会话级数据流，Worker 输出通道

Registry Keys（注册中心）¶

Key 模式	类型	用途
`byai_gateway:registry:workers`	Set	所有已知 Worker ID 集合
`byai_gateway:registry:worker:online:{worker_id}`	String	Worker 在线租约（TTL：Python 30s，Java / TypeScript 15s —— 三语言常量当前不一致，见下方提示）
`byai_gateway:registry:worker:lock:{worker_id}`	String	Worker 启动互斥锁（TTL 60s）
`byai_gateway:registry:worker:agent_types:{worker_id}`	Set	Worker 声明的 Agent type 集合
`byai_gateway:registry:agent_type:workers:{agent_type}`	Set	Agent type 的成员 Worker 集合
`byai_gateway:registry:worker:status:{worker_id}`	String	Worker 当前状态
`byai_gateway:registry:worker:admin:{worker_id}`	Hash	WorkerManager 管理的生命周期状态（`lifecycle`/`reason`/`updated_at`），无 TTL

Worker 在线租约 TTL 三语言不一致

Python 的 WORKER_DEFAULT_LEASE_TTL_SECONDS 是 30，Java / TypeScript 都是 15，而三者的心跳间隔（WORKER_DEFAULT_HEARTBEAT_INTERVAL_SECONDS）统一是 5——也就是说 Python worker 要连续错过 6 次心跳才会被判定离线，Java / TypeScript worker 只需错过 3 次。这看起来更像历史遗留的实现漂移，而不是有意为之的设计差异，混部场景下需要注意这个差异带来的离线判定时间不一致。

Execution Keys（执行追踪）¶

执行记录不是存成独立的顶层 key，而是作为 Hash 字段挂在会话级聚合注册表和 Worker 级 ZSET 下：

Key 模式	类型	用途
`byai_gateway:session:{session_id}:registry`	Hash	会话级聚合注册表；内部按字段前缀细分两类数据：`exec:{execution_id}` 字段存单次执行详情（JSON），`msg_map:{message_id}` 字段存消息 ID → 执行 ID 的映射
`byai_gateway:registry:worker:executions:{worker_id}`	Sorted Set	Worker 处理过的执行 ID，score 为更新时间
`byai_gateway:registry:worker:active_execution_index:{worker_id}`	Sorted Set	Worker 当前活跃（未终态）执行 ID，score 为更新时间
`byai_gateway:registry:worker:active_snapshots:{worker_id}`	Hash	活跃执行 ID → 轻量快照
`byai_gateway:registry:worker:history_snapshots:{worker_id}`	Hash	历史执行 ID → 轻量快照

Task Group Keys（任务组）¶

Key 模式	类型	用途
`byai_gateway:task_group:{group_id}`	Hash	任务组计数器 (total / completed)
`byai_gateway:task_group:{group_id}:results`	Hash	任务组结果集（TTL 86400s）

Service Discovery Keys（服务发现）¶

Key 模式	类型	用途
`byai_gateway:sd:services`	Set	全局服务名称索引
`byai_gateway:sd:active:{service_name}`	Sorted Set	活跃实例（score = 心跳时间戳）
`byai_gateway:sd:instances:{service_name}`	Hash	实例详情（host, port, metadata）

Consumer Group¶

Key 模式	说明
`byai_gateway:consumer_group:agent_engines`	默认消费者组名

Consumer Group 路由架构¶

graph TB
    Client["GatewayClient"]

    subgraph AgentTypeRoute["Agent Type 路由 (默认)"]
        direction TB
        S1["byai_gateway:ctrl:agent_type:chat_agent"]
        CG["Consumer Group<br/>byai_gateway:consumer_group:agent_engines"]
        W1["Worker-1"]
        W2["Worker-2"]
        W3["Worker-3"]

        S1 --> CG
        CG -->|"竞争消费"| W1
        CG -->|"竞争消费"| W2
        CG -->|"竞争消费"| W3
    end

    subgraph DirectRoute["Worker 定向路由"]
        direction TB
        S2["byai_gateway:ctrl:worker:worker-2"]
        W2D["Worker-2"]
        S2 -->|"独占消费"| W2D
    end

    Client -->|"targetAgentType"| S1
    Client -.->|"targetWorkerId"| S2

Agent Type 路由 (默认)¶

消息写入 byai_gateway:ctrl:agent_type:{agent_type}
在 requireOnlineWorker=true 时验证是否存在在线 Worker
同一 Agent type 的所有 Worker 通过 Consumer Group 竞争消费

Worker 定向路由¶

传入 targetWorkerId 后消息写入 byai_gateway:ctrl:worker:{worker_id}
适合 debug、取消任务或定向控制

消息确认机制¶

stateDiagram-v2
    [*] --> Pending: XADD 写入
    Pending --> Delivered: XREADGROUP 分发
    Delivered --> Processing: Worker 开始处理
    Processing --> Acknowledged: XACK 确认
    Processing --> Redelivered: 超时未 ACK
    Redelivered --> Delivered: 重新分发
    Acknowledged --> [*]: 从 PEL 中移除

Pending: 消息已写入 Stream
Delivered: 被 Consumer Group 分发给某个 Worker
Processing: Worker 正在处理
Acknowledged: Worker 发送 XACK，消息从 PEL (Pending Entries List) 移除
Redelivered: 未及时 ACK 时自动重新分发

消费者组配置¶

PythonJavaTypeScript

run_worker(
    consumer_group="agent_engines",  # 消费者组名
    max_concurrency=50,              # 最大并发
    fetch_count=10,                  # 批量获取数量
)

WorkerRunner runner = new WorkerRunner(worker);
// 消费者组名默认为 "byai_gateway:consumer_group:agent_engines"
// 通过 WorkerConfig 配置并发和批量获取参数
runner.start();

runWorker(MyWorker, {
    consumerGroup: "agent_engines",  // 消费者组名
    maxConcurrency: 50,              // 最大并发
    fetchCount: 10,                  // 批量获取数量
});

Key 命名空间规范¶

所有 Redis Key 均以 byai_gateway: 为前缀，按功能域划分子命名空间：

graph LR
    Root["byai_gateway:"]

    Root --> ctrl["ctrl:<br/>控制流"]
    Root --> session["session:<br/>会话数据"]
    Root --> registry["registry:<br/>注册中心"]
    Root --> sd["sd:<br/>服务发现"]
    Root --> task_group["task_group:<br/>任务组"]
    Root --> consumer_group["consumer_group:<br/>消费者组"]

    ctrl --> ctrl_at["agent_type:{type}"]
    ctrl --> ctrl_wk["worker:{id}"]

    session --> sess_ds["{sid}:data_stream"]
    session --> sess_reg["{sid}:registry"]

    registry --> reg_workers["workers"]
    registry --> reg_worker["worker:*"]
    registry --> reg_at["agent_type:*"]
    registry --> reg_exec["execution:*"]