如何在指定时间间隔后移除Redis Stream中已完成的事件？

Redis的XTRIM命令确实只依据事件ID或流长度进行修剪，无法区分事件的处理状态（已完成/待处理）。要实现仅删除已完成（已ACK）事件、保留待处理（pending）事件的需求，目前最可靠的方式是通过Lua脚本实现原子筛选与删除，以下是具体方案：

核心思路 #

已完成事件是指被消费组内消费者成功ACK的事件，这类事件不在消费组的PEL（Pending Entries List）中；而待处理事件会保留在PEL里。我们可以通过Lua脚本原子完成以下步骤：

1. 拉取目标消费组的所有待处理事件ID
2. 筛选出指定时间范围内、不在PEL中的事件（即已完成事件）
3. 批量删除这些已完成事件

Lua脚本实现 #

-- 参数说明：
-- KEYS[1] = 目标Stream名称
-- KEYS[2] = 消费组名称
-- ARGV[1] = 待清理事件的起始时间戳（毫秒，如0表示最早）
-- ARGV[2] = 待清理事件的结束时间戳（毫秒，如当前时间戳）
-- ARGV[3] = 单次拉取的pending事件最大数量（避免内存溢出，默认10000）

local pending_ids = redis.call('XPENDING', KEYS[1], KEYS[2], '-', '+', tonumber(ARGV[3]) or 10000)
local pending_set = {}
for _, entry in ipairs(pending_ids) do
    pending_set[entry[1]] = true
end

-- 获取时间范围内的所有Stream事件
local stream_entries = redis.call('XRANGE', KEYS[1], ARGV[1], ARGV[2])
local to_delete = {}
for _, entry in ipairs(stream_entries) do
    local event_id = entry[1]
    if not pending_set[event_id] then
        table.insert(to_delete, event_id)
    end
end

-- 批量删除已完成事件
if #to_delete > 0 then
    redis.call('XDEL', KEYS[1], unpack(to_delete))
end

return to_delete

调用示例 #

使用EVAL命令执行脚本，比如清理mystream流中mygroup消费组下，时间戳在0到1717200000000（2024年6月1日）之间的已完成事件：

EVAL "上面的Lua脚本内容" 2 mystream mygroup 0 1717200000000 10000

注意事项 #

• 调整XPENDING的拉取数量：如果待处理事件较多，需增大ARGV[3]的值，避免遗漏pending事件导致误删
• 分批次处理：若Stream中事件量极大，建议缩小时间范围，分多次执行脚本，避免Redis长时间阻塞
• 定期执行：可通过外部定时任务（如crontab）或Redis的定时任务（需Redis 7.0+的SCHEDULE命令）定期触发清理
• 版本兼容：该方案支持Redis 5.0及以上版本（消费组功能从5.0开始引入）

内容的提问来源于stack exchange，提问作者Dhivagar