订单超时关单、优惠券到期提醒、15 分钟未支付自动取消、消息重投退避——这类「到点才执行」的需求，用定时任务扫数据库是最容易想到的方案，但只要表一大、并发一高，就会出现扫全表慢、多个实例重复执行、任务取到一半进程挂掉导致丢失等问题。Redis 的 ZSET（sorted set）天生适合做这件事：score 存到期时间戳，member 存任务内容，按 score 排序取出即按到期时间取出。
但「会往 ZSET 里塞数据」和「做出一个不丢、不重、可水平扩展的延迟队列」之间隔着不少坑：如何避免多个消费者拿到同一条任务？ZRANGEBYSCORE 取出来之后删掉之前进程崩溃怎么办？ZADD 用时间戳做 score 时，同一毫秒内大量任务会不会被 ZSET 去重？本文用一个可运行的最小实现逐层拆开这些问题，最终给出「到期扫描 + Lua 原子取出」的完整代码。
一、为什么 ZSET 适合做延迟队列
延迟队列的核心操作只有两个：按到期时间有序地找到该执行的任务、把任务从队列中安全地取走。ZSET 的底层是跳表 + 哈希表，score 有序、member 唯一，恰好覆盖第一个需求。
- 写入：
ZADD delay_queue <execute_at_ms> <task_id>，score 是任务的绝对到期毫秒时间戳，而不是相对延迟秒数。用绝对时间戳的好处是同一个任务的到期时刻与写入时刻无关，重试、补录、迁移都不会算错。 - 读取：
ZRANGEBYSCORE delay_queue -inf <now_ms> LIMIT 0 20，一次拿到已到期的最多 20 条。 - 删除：
ZREM delay_queue <task_id>。
对比其他 Redis 结构：LIST 只能 FIFO，无法按时间排序；普通 SET 无序，取到期任务需要全量遍历；Stream 虽然支持消费组和 ACK，但它的消息 ID 是时间自增的，不支持「未来某个时刻才可见」的语义，只能靠消费者自己过滤。所以 ZSET 是这个场景的主流选择。
这里有一个必须先说清的细节：ZSET 的 member 唯一。如果直接把任务内容 JSON 当 member，两条内容完全相同的任务会被合并，分数被后写的覆盖。正确做法是 member 只放全局唯一的 task_id（如 order:12345:close 或 UUID），任务正文放到独立的 HASH 或 STRING 里，用 task_id 关联。这样天然获得幂等：同一个业务事件重复投递时，ZADD 只是把 score 刷新，不会产生两条任务。
二、数据结构设计与关键参数
下面这套 key 命名在生产环境可以直接用，前缀统一方便排查与清理：
| Key | 类型 | 作用 | 过期策略 |
|---|---|---|---|
delay:zset:{topic} |
ZSET | 按到期时间排序的任务索引，member 为 task_id | 不设 TTL |
delay:job:{task_id} |
HASH | 任务正文：payload、retry_count、topic、created_at | TTL 7 天 |
delay:done:{task_id} |
STRING | 执行完成标记，用于幂等与调试 | TTL 24 小时 |
几个真实可调的参数，按经验值给出：
- 扫描周期：500ms ~ 1s。周期越短延迟精度越高，但空轮询的 QPS 也越高。500ms 下每实例每秒 2 次 ZRANGEBYSCORE，10 个实例才 20 QPS，Redis 毫无压力。
- 单批数量 BATCH：10 ~ 50。太大则单次 Lua 脚本执行时间变长、阻塞其他命令；太小则在积压时消费速度不足。
- 时间戳精度：统一用毫秒，
System.currentTimeMillis()或Date.now()。不要混用秒，这是最常见的低级错误，会导致任务「早到」1000 倍立即执行。 - 任务正文 TTL：必须大于业务上可能的最大延迟时间。一个延迟 7 天的任务，HASH 设 1 小时 TTL，到点执行时会读不到 payload。
注意：集群模式下要使用 hash tag，把同一个 topic 的 ZSET 与任务 HASH 落到同一个 slot 上，否则 Lua 脚本跨 slot 访问会报
CROSSSLOT Keys in request don't hash to the same slot。上面的 key 设计里{topic}与task_id的映射关系需要在业务层保证，简单做法是 topic 固定若干值，并把 task_id 拼成{topic}:uuid格式。
三、朴素实现的问题：取出与删除不是原子的
最直觉的消费者长这样：
import redis, time, json
r = redis.Redis(host="127.0.0.1", port=6379, decode_responses=True)
TOPIC = "order_close"
ZSET = f"delay:zset:{TOPIC}"
def naive_consume():
now_ms = int(time.time() * 1000)
# 1. 取到期任务
ids = r.zrangebyscore(ZSET, "-inf", now_ms, start=0, num=20)
for task_id in ids:
# 2. 删除（如果这一步之前进程崩溃，任务就丢了？不，是重复执行）
r.zrem(ZSET, task_id)
# 3. 读正文并执行
job = r.hgetall(f"delay:job:{task_id}")
if not job:
continue
print("execute", task_id, job.get("payload"))
这段代码能跑，但有两个严重缺陷：
- 重复消费：多个消费者实例同时执行
ZRANGEBYSCORE，会拿到同一批 task_id，然后都去ZREM。虽然只有一个人 ZREM 成功（返回 1），但失败者的代码没有检查返回值就继续执行了——任务被执行多次。如果任务涉及扣款、发券，这就是资损。 - 取出与删除之间的窗口：如果在 ZREM 之后、读完 job 之前进程被 kill，任务已经从 ZSET 移除，正文 HASH 还在但永远不会再被扫到，变成「静默丢失」。反过来，如果先读 job 再 ZREM 时崩溃，任务会在下一轮被重新扫到，属于「至少一次」，这个方向反而更安全。
所以结论是：取任务必须是一次原子操作，用 Lua 脚本把「查到期、删 ZSET、读正文、标记处理中」打包成一个事务执行。Redis 单线程执行 Lua，脚本内不会被其他命令打断。
四、Lua 原子取出：一次脚本完成查删取
下面这个脚本是本文的核心。它接受三个参数：ZSET key、当前时间戳、批量大小，返回一个数组，每个元素是「task_id + 正文 JSON」的字符串（用分隔符拼接）。关键在于 ZREM 的返回值被严格检查：只有真正删掉的那一条才会被返回给调用方，其余实例拿到空结果，从根上杜绝重复消费。
-- pop_due.lua
-- KEYS[1] = delay:zset:{topic}
-- ARGV[1] = now_ms, ARGV[2] = batch_size, ARGV[3] = job_key_prefix
local zset = KEYS[1]
local now_ms = tonumber(ARGV[1])
local batch = tonumber(ARGV[2])
local prefix = ARGV[3]
-- 1. 取出已到期的 task_id
local ids = redis.call('ZRANGEBYSCORE', zset, '-inf', now_ms, 'LIMIT', 0, batch)
if #ids == 0 then
return {}
end
local result = {}
for i = 1, #ids do
local task_id = ids[i]
-- 2. 只有成功删除的实例才拥有该任务
local removed = redis.call('ZREM', zset, task_id)
if removed == 1 then
local job_key = prefix .. task_id
local payload = redis.call('HGET', job_key, 'payload')
if payload then
-- 3. 保留正文，重试时还要用；记录取出时间供监控
redis.call('HSET', job_key, 'picked_at', now_ms)
result[#result + 1] = task_id .. '|' .. payload
else
-- 正文丢失：记一条审计日志，避免任务凭空消失
redis.call('ZADD', zset .. ':orphan', now_ms, task_id)
end
end
end
return result
配合的投递与消费代码（Python，redis-py 5.x 可直接运行）：
import json
import time
import uuid
import redis
POOL = redis.ConnectionPool(host="127.0.0.1", port=6379,
max_connections=32, decode_responses=True)
R = redis.Redis(connection_pool=POOL)
POP_DUE_LUA = R.register_script(open("pop_due.lua").read())
TOPIC = "order_close"
ZSET = f"delay:zset:{TOPIC}"
JOB_PREFIX = "delay:job:"
def enqueue(task_id: str, payload: dict, delay_ms: int) -> None:
"""投递一个延迟任务。delay_ms 为相对当前时刻的延迟毫秒数。"""
execute_at = int(time.time() * 1000) + delay_ms
job_key = JOB_PREFIX + task_id
pipe = R.pipeline(transaction=True)
pipe.hset(job_key, mapping={
"payload": json.dumps(payload, ensure_ascii=False),
"topic": TOPIC,
"retry_count": 0,
"created_at": int(time.time() * 1000),
})
pipe.expire(job_key, 7 * 24 * 3600)
pipe.zadd(ZSET, {task_id: execute_at})
pipe.execute()
def consume_once(batch: int = 20):
"""取一批到期任务，返回 [(task_id, payload_dict), ...]。"""
now_ms = int(time.time() * 1000)
raw = POP_DUE_LUA(keys=[ZSET],
args=[now_ms, batch, JOB_PREFIX])
jobs = []
for item in raw:
task_id, _, payload = item.partition("|")
jobs.append((task_id, json.loads(payload)))
return jobs
if __name__ == "__main__":
enqueue(f"order:{uuid.uuid4().hex}:close", {"order_id": 20240101}, 3000)
for _ in range(10):
got = consume_once()
if got:
for tid, body in got:
print("handle", tid, body)
break
time.sleep(0.5)
注意 partition("|") 而非 split("|")：payload 是 JSON，里面可能含竖线（比如 remark 字段），用 partition 只按第一个分隔符切，剩下的整体作为 payload，不会切坏。
五、执行语义与可靠性增强
5.1 至少一次，靠业务幂等兜底
上面的脚本保证「一个任务同一时刻只有一个实例能取到」，但保证不了「只执行一次」：任务被取走后处理到一半崩溃，它已经从 ZSET 消失，不会再被扫描到。这属于「最多一次」的风险。要变成「至少一次」，需要引入处理中超时回队。
做法是脚本里不直接删 ZSET，而是改用一个 waiting list（LIST）存放正在处理的任务，并记录取出时间。后台每隔 30 秒扫一次 waiting list，把取出时间超过 5 分钟（大于最大处理耗时）的任务重新 ZADD 回主 ZSET。因为 ZSET 的 member 是 task_id，重复回队也只是刷新 score，不会产生重复条目。
-- requeue_timeout.lua 的超时判定片段
-- KEYS[1] = waiting list key, ARGV[1] = 超时毫秒阈值, ARGV[2] = zset key
local waiting_key = KEYS[1]
local zset_key = KEYS[2]
local timeout_ms = tonumber(ARGV[1])
local now_ms = int(time.time() * 1000) if false else tonumber(ARGV[2])
local items = redis.call('LRANGE', waiting_key, 0, 99)
local moved = 0
for _, v in ipairs(items) do
local task_id, picked = v:match('^(.-)|(%d+)$')
if task_id and now_ms - tonumber(picked) > timeout_ms then
redis.call('ZADD', zset_key, now_ms, task_id)
redis.call('LREM', waiting_key, 1, v)
moved = moved + 1
end
end
return moved
生产环境务必用
SET delay:lock:requeue <instance_id> NX PX 30000给回队扫描加分布式锁，否则多个实例同时回队会互相干扰。不要用 Redlock 之外的「看门狗」简化版做有资损的操作，回队本身是幂等的，简单锁就够了。
5.2 失败重试与退避
业务处理抛异常时不要直接丢弃。把 retry_count 加 1，按指数退避重算 score 后 ZADD 回队列：第 n 次重试的延迟取 min(60s * 2^n, 3600s)，n 超过 5 次则写入死信 ZSET delay:dlq:{topic}，由人工或告警系统介入。重试时一定要检查 delay:done:{task_id} 是否存在，防止「已经成功的任务因为 ACK 超时被重复执行」。
5.3 批量与背压
如果 consume_once 返回满额 batch（比如 20 条），说明有积压，应立即循环再取一次，而不是死等下一个 500ms 周期；返回为 0 时再 sleep。这就是最简单的自适应消费：
def run_forever():
while True:
got = consume_once(batch=20)
if not got:
time.sleep(0.5)
continue
for task_id, payload in got:
handle(task_id, payload) # 内部自行 try/except 并决定重试
六、排查与监控：几个真实报错的解法
报错 1：WRONGTYPE Operation against a key holding the wrong kind of value
通常是 key 名和业务其他模块冲突了。比如有人用 order:close 存了一个 STRING 锁，而你把它当 ZSET 用。解决办法是把 key 全量加命名空间前缀 delay:，上线前用 SCAN 0 MATCH delay:* COUNT 1000 检查类型，TYPE delay:zset:order_close 确认为 zset。
报错 2：CROSSSLOT Keys in request don't hash to the same slot
Redis Cluster 下 Lua 脚本访问的多个 key 不在同一个 slot。解法是给 ZSET 和 HASH 都用同一个 hash tag：delay:zset:{order_close} 与 delay:job:{order_close}:<uuid>，脚本里两者的 {...} 内容必须一致。注意这会把同一个 topic 的所有任务压到一个分片上，热点 topic 要考虑按业务再拆子 topic。
报错 3：BUSY Redis is busy running a script
Lua 脚本执行超过 lua-time-limit（默认 5000ms）。常见原因是 batch 过大加上 payload 巨大（比如把整个订单快照塞进 HASH 的 payload 字段）。把 payload 控制在 10KB 以内，大对象只存 ID，正文另行查询。
必须监控的四个指标，建议每 10 秒采集一次：
- 积压量：
ZCARD delay:zset:{topic}，持续上涨说明消费能力不足。 - 到期未消费数：
ZCOUNT delay:zset:{topic} -inf <now_ms - 30000>，即已经到期超过 30 秒还在队列里的任务数，这是延迟精度的直接度量，大于 0 就该告警。 - 消费耗时 P99：在
handle里打点，P99 必须小于回队超时阈值，否则任务会被误判超时导致重复执行。 - 孤儿任务数：
delay:zset:{topic}:orphan的 ZCARD，非 0 说明正文 HASH 的 TTL 设短了或被人为删除。
小结
用 ZSET 做延迟队列的骨架只有三件事：score 存绝对到期毫秒时间戳、member 存唯一的 task_id（正文另存）、用 Lua 把「查到期 + 删 + 读正文」合成一个原子操作并检查 ZREM 返回值为 1。再叠加处理中超时回队、失败指数退避、死信队列，就得到一个可上线的最小实现。它的延迟精度取决于扫描周期（500ms 级别），不适合毫秒级精确定时；单 ZSET 的吞吐在万级 QPS 以内都很轻松，再高就按业务拆 topic 分片。
FAQ
Q1：为什么 score 用绝对时间戳而不是延迟秒数？
延迟秒数只在写入那一刻有意义，任务一旦重试、迁移或补录，score 就失去了参照系。绝对时间戳是全局单调的（同一台机器上），任何时刻读取都能直接与 now 比较，天然支持重试改期。
Q2：两个任务的 task_id 相同会怎样？
后写入的 ZADD 会覆盖前者的 score，即「以最后一次为准」。这通常正是业务想要的幂等效果。如果你的业务需要同一事件多次延迟执行，task_id 里必须带唯一后缀，比如 order:123:notify:1、order:123:notify:2。
Q3：能不能用 Redis 的 Keyspace Notification 做到期通知？
不能。Redis 的键过期通知是「惰性 + 定期清理」驱动的，过期事件可能延迟数分钟甚至在被访问前一直不触发，且 pub/sub 不保证送达。它只适合缓存失效提示，不能承载任务调度。
Q4：ZRANGEBYSCORE 的 LIMIT 一定要加吗？
一定要。某次故障中积压了 80 万条到期任务，不加 LIMIT 的单次调用返回了全部 member，直接打爆客户端内存并阻塞了 Redis 数秒。永远是 LIMIT 0 <batch>，再用循环消费积压。
Q5：延迟精度能到多少？
理论下限是扫描周期。500ms 轮询下，实际误差通常在 0~600ms。若业务要求秒级以内精确，可以缩短到 100ms，但要把每实例的轮询 QPS 和 Redis 连接数算进去，或者改用「就近唤醒」：维护一个最近到期时间，睡到那个时间点再醒，而不是固定周期空转。