延迟任务无处不在
下完订单 30 分钟未支付，自动关闭。优惠券到期前 1 小时，推送提醒。注册后 3 天未登录，发送召回邮件。这些场景的本质是同一个问题：在某个未来时间点执行一个任务。
延迟队列的难点不在于"能不能实现"，而在于在高并发下既不丢任务、又不重复执行，同时保持可接受的资源开销。下面从最简单的方案讲起，逐步演进。
方案一：数据库轮询
最直觉的做法——把延迟任务存到数据库表里，起一个定时任务每秒扫一遍。
1
2
3
4
5
6
7
8
9
|
CREATE TABLE delay_tasks (
id BIGINT PRIMARY KEY AUTO_INCREMENT,
task_type VARCHAR(64) NOT NULL,
payload TEXT NOT NULL,
execute_at DATETIME NOT NULL,
status TINYINT DEFAULT 0, -- 0=待执行 1=执行中 2=已完成
created_at DATETIME DEFAULT NOW(),
INDEX idx_execute_at (execute_at, status)
);
|
消费逻辑用 Go 实现如下：
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
28
29
30
31
|
func PollTasks(db *sql.DB) {
ticker := time.NewTicker(1 * time.Second)
defer ticker.Stop()
for range ticker.C {
rows, err := db.Query(`
SELECT id, task_type, payload
FROM delay_tasks
WHERE execute_at <= NOW() AND status = 0
LIMIT 50`)
if err != nil {
log.Printf("query error: %v", err)
continue
}
var tasks []Task
for rows.Next() {
var t Task
rows.Scan(&t.ID, &t.Type, &t.Payload)
tasks = append(tasks, t)
}
rows.Close()
for _, t := range tasks {
// 用 UPDATE 抢占任务，防止多实例重复执行
res, _ := db.Exec(
`UPDATE delay_tasks SET status = 1 WHERE id = ? AND status = 0`, t.ID)
if n, _ := res.RowsAffected(); n > 0 {
go executeTask(t)
}
}
}
}
|
优点：实现简单，事务保证强，不丢数据。
缺点：轮询空转浪费资源，数据库压力大；精度受轮询间隔限制；扩展性差。
适用场景：任务量小（<1000/秒），精度要求秒级，团队不想引入额外中间件。
方案二：Redis ZSet 有序集合
用 Redis 的 ZSet，把任务执行时间作为 score，消费端不断拉取 score <= 当前时间的任务。
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
28
29
30
31
32
33
34
|
type RedisDelayQueue struct {
client *redis.Client
key string
}
func (q *RedisDelayQueue) Push(taskID string, executeAt time.Time) error {
return q.client.ZAdd(q.key, redis.Z{
Score: float64(executeAt.Unix()),
Member: taskID,
}).Err()
}
func (q *RedisDelayQueue) Consume() {
for {
now := time.Now().Unix()
// 取出所有到期的任务
members, err := q.client.ZRangeByScore(q.key, &redis.ZRangeBy{
Min: "0",
Max: strconv.FormatInt(now, 10),
Count: 50,
}).Result()
if err != nil || len(members) == 0 {
time.Sleep(500 * time.Millisecond)
continue
}
for _, member := range members {
// 原子删除：只有第一个删成功的才会执行
n, err := q.client.ZRem(q.key, member).Result()
if err == nil && n > 0 {
go executeTask(member)
}
}
}
}
|
关键在于 ZRem 的原子性——多个消费者同时抢同一个任务，只有 ZRem 返回 1 的那个才会真正执行，天然防重复。
优点：内存操作极快，单机支撑 10 万+ QPS；ZRem 原子保证不重复执行。
缺点：断电丢数据（可通过 AOF 持久化缓解但无法完全避免）；任务执行失败需要手动重试机制。
适用场景：任务量大、允许极小概率丢失、已有 Redis 基础设施。
方案三：Redis + Lua 脚本（解决并发竞争）
方案二在超高并发下，ZRangeByScore 和 ZRem 之间存在一个窗口——多个消费者可能同时拿到同一个任务，然后都尝试 ZRem。虽然最终只有一个成功，但无效的竞争白白浪费资源。
用 Lua 脚本将"查询 + 删除"合为一个原子操作：
1
2
3
4
5
6
7
8
|
-- KEYS[1] = 延迟队列的 ZSet key
-- ARGV[1] = 当前时间戳
-- ARGV[2] = 一次取多少个
local tasks = redis.call('ZRANGEBYSCORE', KEYS[1], '0', ARGV[1], 'LIMIT', 0, ARGV[2])
if #tasks > 0 then
redis.call('ZREM', KEYS[1], unpack(tasks))
end
return tasks
|
Go 调用代码：
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
|
var consumeScript = redis.NewScript(`
local tasks = redis.call('ZRANGEBYSCORE', KEYS[1], '0', ARGV[1], 'LIMIT', 0, ARGV[2])
if #tasks > 0 then
redis.call('ZREM', KEYS[1], unpack(tasks))
end
return tasks
`)
func (q *RedisDelayQueue) ConsumeWithLua() {
for {
now := strconv.FormatInt(time.Now().Unix(), 10)
result, err := consumeScript.Do(q.client, q.key, now, 50).Result()
if err != nil {
log.Printf("lua error: %v", err)
time.Sleep(500 * time.Millisecond)
continue
}
tasks := result.([]interface{})
if len(tasks) == 0 {
time.Sleep(500 * time.Millisecond)
continue
}
for _, t := range tasks {
go executeTask(t.(string))
}
}
}
|
Lua 脚本在 Redis 单线程内执行，天然原子，彻底消除竞争窗口。
优点：无竞争开销，吞吐更高。
缺点：Lua 脚本会阻塞 Redis 线程，如果单次取太多任务会影响其他命令；Redis Cluster 下需要按 hash tag 路由到同一分片。
方案四：分层时间轮
对于海量延迟任务（百万级），无论数据库还是 Redis ZSet 都会遇到瓶颈。时间轮是一种 O(1) 复杂度的经典数据结构，Kafka 和 Netty 内部都在用。
核心思路：想象一个时钟，有 N 个格子。任务按延迟时间的取模放入对应格子，指针每 tick（如 100ms）前进一格，执行格子里的所有任务。
1
2
3
4
5
6
7
8
|
分层时间轮结构：
[0] [1] [2] [3] [4] [5] [6] [7] ... [59] ← 秒级轮（60格）
|
+-- 超过60秒的任务，升级到分钟轮
[0] [1] [2] ... [59] ← 分钟轮（60格）
|
+-- 超过60分钟的任务，升级到小时轮
[0] [1] ... [23] ← 小时轮（24格）
|
简化版 Go 实现：
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
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
|
type TimerTask struct {
executeAt int64
fn func()
}
type TimeWheel struct {
slots [][]*TimerTask
curSlot int
tickDur time.Duration
slotNum int
mu sync.Mutex
}
func NewTimeWheel(tick time.Duration, slots int) *TimeWheel {
tw := &TimeWheel{
slots: make([][]*TimerTask, slots),
tickDur: tick,
slotNum: slots,
}
return tw
}
func (tw *TimeWheel) Add(delay time.Duration, fn func()) {
tw.mu.Lock()
defer tw.mu.Unlock()
ticks := int(delay / tw.tickDur)
slot := (tw.curSlot + ticks) % tw.slotNum
tw.slots[slot] = append(tw.slots[slot], &TimerTask{
executeAt: time.Now().Add(delay).Unix(),
fn: fn,
})
}
func (tw *TimeWheel) Start() {
ticker := time.NewTicker(tw.tickDur)
defer ticker.Stop()
for range ticker.C {
tw.mu.Lock()
tw.curSlot = (tw.curSlot + 1) % tw.slotNum
tasks := tw.slots[tw.curSlot]
tw.slots[tw.curSlot] = nil
tw.mu.Unlock()
for _, t := range tasks {
if time.Now().Unix() >= t.executeAt {
go t.fn()
}
}
}
}
|
优点：O(1) 入队和出队，内存开销小，适合百万级任务。
缺点：单机内存方案，宕机丢数据；实现复杂度高（完整分层时间轮涉及任务降级逻辑）；精度受 tick 粒度限制。
适用场景：超大规模延迟任务、内存可接受丢失、追求极致性能。
选型对比
| 方案 |
精度 |
吞吐量 |
可靠性 |
实现复杂度 |
适用场景 |
| 数据库轮询 |
秒级 |
低 |
高 |
低 |
任务少、强一致 |
| Redis ZSet |
毫秒级 |
高 |
中 |
中 |
通用首选 |
| Redis + Lua |
毫秒级 |
很高 |
中 |
中高 |
超高并发 |
| 时间轮 |
tick级 |
极高 |
低 |
高 |
海量任务、内存态 |
实践建议
-
大多数场景选 Redis ZSet 方案：在性能和可靠性之间取得了最佳平衡，社区成熟，踩坑少。
-
务必实现失败重试：任务从队列取出后如果执行失败，需要写回队列或转入死信队列。可以用独立的 retry ZSet，score 为下次重试时间。
-
监控队列积压：用 ZCARD 检查队列长度，设置告警阈值。积压说明消费者处理不过来或任务执行时间过长。
-
幂等设计：无论哪种方案，任务都可能被重复投递（比如服务重启后重新加载），消费者必须做幂等处理——用唯一 ID + 去重表或 Redis SETNX。
-
优雅关闭：消费协程要响应 context.Done()，关闭前处理完已拉取的任务，避免丢弃。
延迟队列没有银弹，选型时先看可靠性和吞吐量要求，再看团队技术栈，最后看运维成本。简单的数据库轮询在 80% 的场景下就够用了，别过度设计。