Redis Stream

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
// 使用 Redis Stream 的 AUTOCLAIM 机制，自动从 PENDING 队列中回收“超时未确认”的消息
AutoClaimResult<String, String> result = stream.autoClaim(
        CONSUMER_GROUP,        // 消费者组名称（Stream 的消费组，比如 stream-processor）
        CONSUMER_NAME,         // 当前执行 reclaim 的消费者（比如 reclaim-worker）
        
        idleTimeoutMs,         // 关键参数：消息在 PENDING 中“空闲多久算超时”
                              // idle_time >= 该值（单位由下面 TimeUnit 决定）才会被回收
        
        TimeUnit.MILLISECONDS, // idleTimeoutMs 的时间单位（毫秒/秒等）
        
        startId,               // 扫描 PENDING 的起始游标（一般第一次用 "0-0"）
                              // Redis 会从这个位置开始遍历 pending entries list
        
        CLAIM_BATCH_SIZE       // 本次最多回收多少条消息（防止一次性处理过多）
);

// ---------------------- 返回结果说明 ----------------------

// result.getNextId()
// → 下一次 AUTOCLAIM 扫描的起点（cursor）
// → 如果返回 "0-0"，说明本轮扫描已结束

// result.getMessages()
// → 本次成功被“抢回”的消息列表（从旧 consumer 转移到当前 consumer）

// result.getDeletedIds()
// → 在 PENDING 中已不存在的消息 ID（可能已被 XACK 或删除）

idleTimeoutMs

XADD 写入消息
↓
XREADGROUP 被 consumer1 读到
↓ idle_time = 0
consumer1 开始处理
↓（没 XACK）
时间流逝 30s
↓ idle_time = 30000ms
XAUTOCLAIM 被 reclaim-worker 抢走
↓ idle_time = 0（重新开始计时）