怎样实现延迟任务脚本

wen 实用脚本 2026-06-30 28

从原理到代码全解析

目录导读

延迟任务的核心概念与常见场景
延迟任务的四种主流实现方案
基于Redis的延迟队列（推荐）
基于MySQL轮询的简单实现
使用Linux Crontab定时执行
借助消息中间件（RabbitMQ/RocketMQ）
性能对比与选型建议
常见问题QA
如何选择最适合你的延迟任务方案

延迟任务的核心概念与常见场景

什么是延迟任务？
延迟任务是指在未来某个指定时间点执行的任务，而非立即触发，典型场景包括：

怎样实现延迟任务脚本

订单支付超时自动取消（30分钟后）
用户注册后发送欢迎邮件（延迟5分钟）
定时数据备份（每天凌晨2点）
优惠券到期前48小时提醒

关键难点：
延迟任务需要准确计时、持久化存储（防止系统重启丢失）以及高并发下的可靠性。

延迟任务的四种主流实现方案

方案	适用规模	精准度	成本	代码复杂度
Redis有序集合（ZSET）	中高频使用	秒级	低	中
MySQL轮询	低频少量任务	分钟级	极低	低
Linux Crontab	固定周期任务	分钟级	极低	极低
消息中间件（MQ）	大流量场景	毫秒级	较高	高

方案一：基于Redis的延迟队列（推荐）

实现原理

利用Redis的有序集合（Zset）数据结构，以任务的执行时间戳作为Score，原子化地获取到期的任务。

核心代码示例（Python）

import redis
import time
r = redis.Redis(host='localhost', port=6379)
# 添加延迟任务
def add_delayed_task(task_id, delay_seconds, data):
    score = time.time() + delay_seconds
    r.zadd('delay_queue', {task_id: score})
    r.set(f'task_data:{task_id}', data)
# 消费延迟任务
def consume_tasks():
    while True:
        # 获取到期的任务（Score小于当前时间）
        tasks = r.zrangebyscore('delay_queue', 0, time.time(), start=0, num=1)
        if tasks:
            task_id = tasks[0].decode()
            # 从队列移除（保证幂等性）
            if r.zrem('delay_queue', task_id):
                data = r.get(f'task_data:{task_id}')
                print(f'执行任务{task_id}: {data}')
                r.delete(f'task_data:{task_id}')
        else:
            time.sleep(1)  # 避免CPU空转

优点：

性能高（Redis单机QPS可达10万+）
支持任意秒级的延迟
缺点：
需要额外处理任务失败重试
依赖Redis内存，大数据量需估算

方案二：基于MySQL轮询的简单实现

实现方法

创建一张任务表 delayed_tasks，包含字段：id, execute_at, status(pending/done), task_data。
使用一个守护进程每隔X秒扫描表中 execute_at <= NOW() 且 status=pending 的记录并执行。

SQL查询示例：

SELECT * FROM delayed_tasks 
WHERE status = 'pending' AND execute_at <= NOW() 
ORDER BY execute_at ASC 
LIMIT 100;

注意事项：

必须使用索引（status, execute_at）避免全表扫描
需要事务锁或乐观锁防止重复执行
轮询频率越高，数据库压力越大（通常设为10~30秒）

使用场景：项目初期、任务量<1000/天、对延迟不敏感。

方案三：使用Linux Crontab定时执行

适用场景

固定周期的延迟任务，

每天凌晨4点清理日志
每30分钟检测系统健康状态

Shell脚本示例

# 每天23:30执行备份脚本（延迟到指定时刻）
30 23 * * * /usr/local/bin/backup.sh

局限性：

只能到分钟级（支持25种特殊语法，如*/5表示每5分钟）
无法处理“用户点击后延迟30分钟”的动态任务
需要手动编辑crontab文件，不适合动态添加

方案四：借助消息中间件（RabbitMQ/RocketMQ）

实现原理

以RabbitMQ的死信队列（DLX）或RocketMQ的定时消息为例。

生产者发送消息时指定 delay_seconds
消息先存储在延迟队列，到期后自动转发到业务队列
消费者从业务队列获取消息并执行

RabbitMQ延迟插件示例（rabbitmq_delayed_message_exchange）

import pika
conn = pika.BlockingConnection()
channel = conn.channel()
# 声明延迟交换机
channel.exchange_declare('delayed_exchange', 'x-delayed-message', 
                         arguments={'x-delayed-type': 'direct'})
# 发送消息，延迟5秒
channel.basic_publish(exchange='delayed_exchange', routing_key='task', 
                      body='Hello', 
                      properties=pika.BasicProperties(headers={'x-delay': 5000}))

优点：