应用场景

问题分析

延时队列的核心需求可以抽象为三点：

1. 存储能力：能够持久化大量任务（百万级甚至亿级）
2. 查询效率：快速找到”时间到了”的任务（时间维度查询）
3. 删除能力：支持取消任务（如用户支付后取消超时任务）

这本质上是在时间维度上做高效的检索和调度。

技术方案对比

推荐决策树

是否需要支持分布式？
├─ 否 → 单机方案
│   ├─ 任务量 < 10万，精度要求不高 → 数据库 + 轮询
│   └─ 任务量 > 10万，或要求秒级精度 → Redis ZSet
│
└─ 是 → 分布式方案
    ├─ 已在使用 RocketMQ/RabbitMQ → 使用内置延时队列
    ├─ 需要 > 18天延时 → 自研（时间轮 + 持久化）
    └─ 高并发 + 精度要求 → 时间轮 + 消息队列

最佳实践建议

1. 小团队起步：先用数据库方案快速验证业务价值
2. 中等规模：迁移到 Redis ZSet，性价比最高
3. 大规模系统：考虑分层设计
   • 1小时内短延时：Redis ZSet（高精度）
   • 1小时以上长延时：数据库轮询（低成本）

代码示例 - Redis ZSet 方案

import redis
import time
import json

class RedisDelayQueue:
    def __init__(self, queue_name, redis_host='localhost'):
        self.redis = redis.Redis(host=redis_host, decode_responses=True)
        self.queue_key = f"delay_queue:{queue_name}"
    
    def add_task(self, task_id, task_data, delay_seconds):
        """添加延时任务"""
        execute_at = int(time.time()) + delay_seconds
        # score 为执行时间戳，member 为 task_id
        self.redis.zadd(
            self.queue_key,
            {json.dumps({'task_id': task_id, 'data': task_data}): execute_at}
        )
        return task_id
    
    def get_ready_tasks(self, batch_size=100):
        """获取到期的任务（批量）"""
        now = int(time.time())
        # 查询 score <= now 的任务
        ready_tasks = self.redis.zrangebyscore(
            self.queue_key,
            0,
            now,
            start=0,
            num=batch_size
        )
        return [json.loads(task) for task in ready_tasks]
    
    def remove_task(self, task_id):
        """删除任务（如用户支付后取消超时任务）"""
        # 需要遍历找到对应的 task（实际应用中可以建立反向索引）
        tasks = self.redis.zrange(self.queue_key, 0, -1)
        for task in tasks:
            task_data = json.loads(task)
            if task_data['task_id'] == task_id:
                self.redis.zrem(self.queue_key, task)
                return True
        return False
    
    def consume_tasks(self, handler, batch_size=100):
        """消费到期任务"""
        while True:
            ready_tasks = self.get_ready_tasks(batch_size)
            if not ready_tasks:
                time.sleep(1)  # 没有到期任务，休眠1秒
                continue
            
            for task in ready_tasks:
                try:
                    # 执行任务
                    handler(task['data'])
                    # 完成后删除
                    self.redis.zrem(self.queue_key, json.dumps(task))
                except Exception as e:
                    # 异常处理：记录日志，根据业务决定是否重试
                    print(f"任务执行失败: {task['task_id']}, 错误: {e}")

关键洞察

• 没有银弹：不同阶段适合不同方案
• Redis ZSet 是最通用的中间方案：覆盖 90% 的场景
• 分层设计是大规模系统的终极方案：短延时用内存，长延时用磁盘

问题分析

“45% 的任务在 30 分钟以内”这个数据点透露了几个关键信息：

1. 短延时任务占主导：近一半的任务对时间精度敏感
2. 轮询方案的缺陷被放大：1 分钟轮询间隔对 30 分钟任务来说，误差达 3.3%
3. 优化方向明确：优化短延时任务的收益最大

精度影响分析

假设采用”每分钟轮询数据库”的方案：

结论：对于 30 分钟以内的任务，1 分钟的轮询间隔带来的误差是不可接受的。

分层设计策略

基于延时分布，我们可以采用分层架构来平衡精度和成本：

┌─────────────────────────────────────────────────────────────┐
│                     延时任务分层架构                         │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  ┌──────────────────┐                                      │
│  │  热数据层（内存）  │  ← 30分钟内的任务（45%）            │
│  │  • Redis ZSet    │     • 高精度（秒级）                  │
│  │  • 时间轮        │     • 快速响应                        │
│  │  • 精度：1秒     │     • 内存成本可控（短延时任务很快消费）│
│  └──────────────────┘                                      │
│           ↓                                                │
│  ┌──────────────────┐                                      │
│  │  温数据层（数据库）│  ← 30分钟~7天的任务（35%）          │
│  │  • MySQL索引     │     • 中等精度（分钟级）              │
│  │  • 每分钟轮询    │     • 成本适中                        │
│  │  • 精度：1分钟   │                                      │
│  └──────────────────┘                                      │
│           ↓                                                │
│  ┌──────────────────┐                                      │
│  │  冷数据层（归档）  │  ← >7天的任务（20%）                 │
│  │  • 定时归档      │     • 低精度（小时级）                │
│  │  • 每小时轮询    │     • 最小成本                        │
│  └──────────────────┘                                      │
│                                                             │
└─────────────────────────────────────────────────────────────┘

成本优化

使用分层设计后的成本对比（假设每天 100 万任务）：

技术选型建议

基于”45% 任务在 30 分钟内”这个数据：

1. 优先级：短延时任务的精度优化是 P0 优先级

2. 技术栈选择：

   • 不要选择纯轮询方案（精度不足）
   • 不要全部用内存方案（成本太高）
   • 选择分层方案（精度和成本的平衡）

3. 实现路径：

   • 第一阶段：全部用 Redis ZSet（快速上线）
   • 第二阶段：引入分层设计（降低成本）
   • 第三阶段：自动化任务迁移（热→温→冷）

代码示例 - 分层延时队列

class TieredDelayQueue:
    def __init__(self):
        # 热数据层：Redis ZSet（30分钟内）
        self.hot_queue = RedisDelayQueue('hot_tasks')
        # 温数据层：数据库轮询（30分钟~7天）
        self.warm_queue = DatabaseDelayQueue('warm_tasks', poll_interval=60)
        # 冷数据层：数据库低频轮询（>7天）
        self.cold_queue = DatabaseDelayQueue('cold_tasks', poll_interval=3600)
    
    SHORT_DELAY_THRESHOLD = 30 * 60  # 30分钟
    MEDIUM_DELAY_THRESHOLD = 7 * 24 * 60 * 60  # 7天
    
    def add_task(self, task_id, task_data, delay_seconds):
        """根据延时时间自动路由到对应的层级"""
        if delay_seconds <= self.SHORT_DELAY_THRESHOLD:
            # 热数据：Redis ZSet，高精度
            return self.hot_queue.add_task(task_id, task_data, delay_seconds)
        elif delay_seconds <= self.MEDIUM_DELAY_THRESHOLD:
            # 温数据：数据库，1分钟精度
            return self.warm_queue.add_task(task_id, task_data, delay_seconds)
        else:
            # 冷数据：数据库，1小时精度
            return self.cold_queue.add_task(task_id, task_data, delay_seconds)
    
    def upgrade_tasks(self):
        """将快要到期的温/冷任务升级到热层"""
        # 例如：将剩余时间 < 1小时的任务从温层移到热层
        # 这样可以保证所有任务在到期前都在高精度层
        pass

关键洞察

1. 数据驱动设计：理解业务数据的分布特征（如延时分布）是技术选型的前提
2. 分层是通用优化思路：不仅延时队列，缓存系统、存储系统都用分层来平衡成本和性能
3. 精度不是越细越好：根据业务需求选择合适的精度，避免过度工程

问题分析

订单支付场景是一个典型的”主动触发取消被动任务”的模式：

用户下单 → 创建"30分钟后取消订单"任务
        ↓
    [29分钟后]
        ↓
    用户支付 → 需要立即取消"取消订单"任务
        ↓
    如果取消失败 → 1分钟后订单被错误取消 → 用户投诉！

这揭示了延时队列设计的三个核心挑战：

1. 任务可寻址性：如何快速找到要取消的任务？
2. 删除原子性：如何确保任务不会被执行两次？
3. 并发安全：支付回调执行的同时，任务刚好到期怎么办？

技术要求详解

要求 1：支持任务取消（API 设计）

延时队列必须提供主动删除接口，不能依赖”执行时检查状态”：

# ❌ 错误做法：依赖执行时检查
class BadDelayQueue:
    def execute_ready_tasks(self):
        tasks = self.get_ready_tasks()
        for task in tasks:
            order = Order.get(task['order_id'])
            if order.status == 'paid':  # 执行时才检查
                continue  # 跳过已支付订单
            # 但是库存已经被锁定了29分59秒！
            self.cancel_order(task['order_id'])

# ✅ 正确做法：支付时主动删除
class GoodDelayQueue:
    def cancel_task(self, task_id):
        """主动删除任务"""
        # 使用 task_id 快速定位并删除
        return self.storage.delete(task_id)

# 支付回调中调用
def on_payment_success(order_id):
    # 1. 更新订单状态
    order = Order.get(order_id)
    order.status = 'paid'
    order.save()
    
    # 2. 主动取消超时任务（关键！）
    delay_queue.cancel_task(
        task_id=f'order_timeout_{order_id}'
    )

要求 2：高效的删除操作

不同存储方案下的删除效率对比：

最佳实践：为每个任务分配全局唯一的 task_id，建立 task_id → 任务 的索引。

import redis
import hashlib

class IndexedDelayQueue:
    def __init__(self):
        self.redis = redis.Redis(decode_responses=True)
        # 任务存储：ZSet，score 为执行时间
        self.queue_key = "delay_queue:tasks"
        # 反向索引：Hash，task_id → 完整任务数据
        self.index_key = "delay_queue:index"
    
    def add_task(self, order_id, delay_seconds):
        """添加延时任务"""
        task_id = f"order_timeout_{order_id}"
        execute_at = int(time.time()) + delay_seconds
        
        task_data = json.dumps({
            'task_id': task_id,
            'order_id': order_id,
            'execute_at': execute_at
        })
        
        pipe = self.redis.pipeline()
        # 1. 添加到队列
        pipe.zadd(self.queue_key, {task_data: execute_at})
        # 2. 建立反向索引（关键！）
        pipe.hset(self.index_key, task_id, task_data)
        pipe.execute()
        
        return task_id
    
    def cancel_task(self, task_id):
        """取消任务（O(1) + O(log n)）"""
        pipe = self.redis.pipeline()
        
        # 1. 从索引中获取任务数据（O(1)）
        task_data = self.redis.hget(self.index_key, task_id)
        if not task_data:
            return False  # 任务不存在或已被执行
        
        # 2. 从队列中删除（O(log n)）
        pipe.zrem(self.queue_key, task_data)
        # 3. 删除索引
        pipe.hdel(self.index_key, task_id)
        pipe.execute()
        
        return True

要求 3：并发安全处理

考虑竞态条件：支付回调执行的同时，任务刚好到期

# 并发场景下的处理
def on_payment_success(order_id):
    """支付成功回调"""
    task_id = f"order_timeout_{order_id}"
    
    # 使用 Redis 事务（或分布式锁）
    with self.redis.pipeline() as pipe:
        while True:
            try:
                # WATCH 监控订单状态
                pipe.watch(f"order:{order_id}")
                order = Order.get(order_id)
                
                if order.status == 'paid':
                    pipe.unwatch()
                    return  # 已处理，直接返回
                
                # 开启事务
                pipe.multi()
                # 更新订单状态
                order.update(status='paid')
                # 取消超时任务
                pipe.zrem(self.queue_key, task_id)
                # 执行事务
                pipe.execute()
                break  # 成功
            except redis.WatchError:
                # 订单被其他进程修改，重试
                continue

生产级最佳实践

1. 幂等性设计

   def on_payment_success(order_id):
       # 1. 先更新订单状态（带幂等检查）
       order = Order.get(order_id)
       if order.status == 'paid':
           return  # 已处理，幂等返回
       
       order.status = 'paid'
       order.save()
       
       # 2. 再取消任务（允许取消不存在的情况）
       delay_queue.cancel_task(f'order_timeout_{order_id}')

2. 任务设计模式

   # 任务内部做二次检查（Defense in Depth）
   def execute_order_cancel_task(order_id):
       order = Order.get(order_id)
       # 任务执行时再次确认订单状态
       if order.status == 'paid':
           return  # 用户已支付，跳过取消
       
       # 确认取消
       order.status = 'cancelled'
       order.save()

3. 监控和告警

   def cancel_task(task_id):
       result = delay_queue.cancel_task(task_id)
       if not result:
           # 任务可能已被执行，需要检查并补偿
           alert_service.send(
               f"任务取消失败: {task_id}，可能已被执行，请检查订单状态"
           )
       return result

架构层面的建议

支付流程中的任务取消：

┌─────────────┐
│  用户支付    │
└──────┬──────┘
       │
       ▼
┌─────────────────────────────────┐
│  1. 更新订单状态为"已支付"       │ ← 主流程
│     (事务保护，幂等设计)          │
└──────────┬──────────────────────┘
           │
           ├──────────────────┐
           │                  │
           ▼                  ▼
    ┌─────────────┐    ┌─────────────┐
    │  取消超时任务 │    │  发送支付成功 │
    │  (异步尝试)   │    │  通知        │
    └─────────────┘    └─────────────┘
           │
           ▼
    ┌──────────────────────┐
    │  失败？记录监控日志   │
    │  + 订单状态兜底检查   │
    └──────────────────────┘

关键洞察

1. 任务取消是延时队列的核心功能：不是可选项，而是必须项
2. 删除效率很重要：O(n) 的删除在百万级任务下不可接受
3. 并发安全需要端到端考虑：从 API 设计到任务执行，每一步都要考虑竞态条件
4. Defense in Depth：多层级防护（幂等 + 事务 + 任务内检查）确保系统健壮性