Stream：可靠消息队列

stream: orders
+---------------------+----------------------------+
| ID                  | fields                     |
+---------------------+----------------------------+
| 1722310000000-0     | action=create sku=2001     |
| 1722310000000-1     | action=pay   order=8801    |   <- 同毫秒第二条, 序号+1
| 1722310005123-0     | action=ship  order=8801    |
+---------------------+----------------------------+
        ID 格式: 毫秒时间戳-序号, 严格递增

# 星号表示让 Redis 自动生成 ID（毫秒时间戳-序号）
XADD orders * action create sku 2001
XADD orders * action pay order 8801
XLEN orders
# 控制长度：近似裁剪到 1 万条（波浪号表示允许略多，性能好）
XADD orders MAXLEN ~ 10000 * action ship order 8801
# 减号和加号是最小/最大 ID，读取全部消息
XRANGE orders - +

# 阻塞读取新消息: $ 表示"只要从现在开始的新消息"
127.0.0.1:6379> XREAD BLOCK 5000 STREAMS orders $
（挂起, 等到新消息或 5 秒超时）

# 这里手写 ID 便于后面 XACK 引用，生产中用星号自动生成
XADD orders 1-1 action create sku 2001
XADD orders 1-2 action pay order 8801
# 从头（0）开始消费；用美元符号则只消费此后的新消息
XGROUP CREATE orders g1 0
# 消费者 c1 领取 2 条新消息，大于号表示"从未投递给本组的消息"
XREADGROUP GROUP g1 c1 COUNT 2 STREAMS orders >
# 再读一次，已经没有新消息了
XREADGROUP GROUP g1 c1 COUNT 2 STREAMS orders >

XADD orders 1-1 action create sku 2001
XADD orders 1-2 action pay order 8801
XGROUP CREATE orders g1 0
XREADGROUP GROUP g1 c1 COUNT 2 STREAMS orders >
# 处理完成，确认第一条
XACK orders g1 1-1
# 组内待确认概况：总数、最小/最大 ID、各消费者的挂起数
XPENDING orders g1
# 详细列表：每条的消费者、空闲毫秒数、投递次数
XPENDING orders g1 - + 10

XADD orders 1-1 action pay order 8801
XGROUP CREATE orders g1 0
# c1 领走消息但不 ACK，模拟处理到一半崩溃
XREADGROUP GROUP g1 c1 COUNT 1 STREAMS orders >
XPENDING orders g1 - + 10
# c2 认领空闲超过 0ms 的消息（生产中填 60000 之类的阈值）
XAUTOCLAIM orders g1 c2 0 0
# 现在这条消息归 c2 了
XPENDING orders g1 - + 10

while True:
    # 1. 优先处理新消息
    msgs = XREADGROUP GROUP g1 c1 COUNT 10 BLOCK 2000 STREAMS orders >
    for id, fields in msgs:
        try:
            process(fields)
            XACK orders g1 id
        except Exception:
            pass          # 不 ACK, 留在 PEL 等待重试/认领
    # 2. 周期性认领僵尸消息（也可由独立任务做）
    XAUTOCLAIM orders g1 c1 60000 0

127.0.0.1:6379> XINFO STREAM orders
 1) "length"
 2) (integer) 3
 3) "radix-tree-keys"
 4) (integer) 1
 ...
127.0.0.1:6379> XINFO GROUPS orders
1) 1) "name"
   2) "g1"
   3) "consumers"
   4) (integer) 2
   5) "pending"
   6) (integer) 1
   7) "last-delivered-id"
   8) "1722310000000-1"
127.0.0.1:6379> XDEL orders 1722310000000-0     # 删除指定消息
(integer) 1
127.0.0.1:6379> XTRIM orders MAXLEN ~ 1000      # 手动裁剪
(integer) 0