Python 定时任务可靠性：Celery Beat + 数据库锁避免重复执行

一、凌晨 3 点的 Cron 任务跑了两次，数据翻倍了

任务执行了 45 分钟，但 Cron 以为它 5 分钟就该完成。
因为 celery beat 配置的是每 5 分钟检查一次。
到了第 6 分钟，beat 认为任务该跑了——上一个还在跑。
于是同样的任务启动了两个实例。

结果：营收报表翻倍，清晨的日报数字吓坏了 CEO。
原因不是代码错误，是缺少互斥机制。

定时任务的可靠性有三层要求：
第一，任务只能单实例运行。
第二，任务失败要能重试。
第三，任务延迟要能告警。

这个场景在数据团队中太常见了。Celery Beat 的调度模型是"定时触发 + 异步执行"——Beat 只管按时发任务，不关心任务是否在执行中。
这种"发射后不管"的设计在短任务（< 1 分钟）中没问题，但在长任务（10-60 分钟）中就是定时炸弹。

更隐蔽的问题是：即使你加了锁防止同一节点上的重复执行，在多 Worker 部署的场景下，锁还必须是分布式的。
同一台机器上的进程锁（fcntl/threading.Lock）管不到其他机器，而生产环境的 Celery Worker 通常部署在多台机器上。
所以定时任务的可靠性保证，本质是一个分布式协调问题。

二、可靠定时任务的防重复执行机制

flowchart TB A[Celery Beat 触发] --> B[尝试获取分布式锁] B --> C{获取成功?} C -->|是| D[执行任务] C -->|否| E[记录日志: 任务正在执行中] D --> F{执行成功?} F -->|是| G[释放锁 + 记录完成时间] F -->|否| H{可重试错误?} H -->|是| I[指数退避重试] H -->|否| J[释放锁 + 告警] I --> B

这个流程中有三个关键设计决策。

第一，锁的获取必须是原子操作（RedisSET NX EX）。
如果"检查锁是否存在"和"设置锁"是两个独立操作，中间会有竞态条件——两个进程同时检查通过，同时获取锁。

第二，锁必须有 TTL（过期时间）。
如果任务获取锁后崩溃或被杀，锁永远不释放，后续任务永远被跳过。
这就是死锁问题——在分布式锁场景中，TTL 是防止死锁的唯一保证。

第三，锁的释放必须是安全的。
不能用简单的DEL删除锁——因为如果任务执行时间超过 TTL，锁已经被自动释放了，另一个实例可能已经获取了新的锁。
此时用DEL会误删别人的锁。必须用 Lua 脚本验证锁的持有者身份后才能删除。

三、完整实现

下面的代码实现了一个完整的可靠定时任务框架，包含分布式锁、任务历史记录和装饰器封装。

""" reliable_scheduler.py - 基于 Celery Beat + 数据库锁的可靠定时任务 """ import hashlib import json import logging import time from datetime import datetime, timedelta from typing import Optional, Callable from contextlib import contextmanager from functools import wraps import redis logger = logging.getLogger(__name__) class DistributedLock: """基于 Redis 的分布式锁""" def __init__( self, redis_client: redis.Redis, lock_name: str, expire_seconds: int = 600, # 锁超时 10 分钟 ): self.redis = redis_client self.lock_key = f"lock:task:{lock_name}" self.expire = expire_seconds self.lock_id = self._generate_lock_id() def _generate_lock_id(self) -> str: """生成唯一的锁标识""" return hashlib.md5( f"{time.time()}:{id(self)}".encode() ).hexdigest() def acquire(self) -> bool: """ 尝试获取锁 返回 True 表示获取成功 """ return self.redis.set( self.lock_key, self.lock_id, nx=True, # 仅当不存在时设置 ex=self.expire, ) def release(self): """释放锁（Lua 脚本保证原子性）""" script = """ if redis.call("get", KEYS[1]) == ARGV[1] then return redis.call("del", KEYS[1]) else return 0 end """ self.redis.eval(script, 1, self.lock_key, self.lock_id) def extend(self, extra_seconds: int = 300): """续期锁（长任务需要）""" script = """ if redis.call("get", KEYS[1]) == ARGV[1] then return redis.call("expire", KEYS[1], ARGV[2]) else return 0 end """ self.redis.eval( script, 1, self.lock_key, self.lock_id, extra_seconds ) class TaskHistory: """任务执行历史（记录到 Redis）""" def __init__(self, redis_client: redis.Redis): self.redis = redis_client def record_start(self, task_name: str): """记录任务开始""" key = f"task:history:{task_name}:last_start" self.redis.set(key, time.time()) def record_complete(self, task_name: str, success: bool): """记录任务完成""" key = f"task:history:{task_name}:last_complete" data = json.dumps({ "success": success, "timestamp": time.time(), }) self.redis.set(key, data) def time_since_last_start(self, task_name: str) -> Optional[float]: """距上次任务开始的时间（秒）""" key = f"task:history:{task_name}:last_start" val = self.redis.get(key) if val: return time.time() - float(val) return None @contextmanager def task_lock( redis_client: redis.Redis, task_name: str, lock_ttl: int = 600, auto_extend: bool = True, ): """ 任务锁上下文管理器 用法: with task_lock(redis, "daily_report", lock_ttl=900) as lock: if lock is None: logger.warning("任务正在执行中，跳过") return run_report() """ lock = DistributedLock(redis_client, task_name, lock_ttl) if not lock.acquire(): logger.warning(f"任务 {task_name} 正在执行中，跳过本次调度") yield None return logger.info(f"获取锁成功，开始执行 {task_name}") try: yield lock finally: lock.release() logger.info(f"任务 {task_name} 锁已释放") def reliable_task( redis_url: str = "redis://localhost:6379/0", lock_ttl: int = 600, max_retries: int = 3, retry_backoff: int = 60, ): """ 可靠任务装饰器 功能: - 分布式锁防止重复执行 - 自动重试（指数退避） - 执行历史记录 """ def decorator(func: Callable): @wraps(func) def wrapper(*args, **kwargs): redis_client = redis.from_url(redis_url) task_name = func.__name__ history = TaskHistory(redis_client) with task_lock(redis_client, task_name, lock_ttl) as lock: if lock is None: return {"status": "skipped", "reason": "already_running"} history.record_start(task_name) for attempt in range(max_retries + 1): try: logger.info( f"执行 {task_name} (第 {attempt+1}/{max_retries+1} 次)" ) # 执行任务 result = func(*args, **kwargs) history.record_complete(task_name, True) logger.info(f"任务 {task_name} 完成") return {"status": "success", "result": result} except Exception as e: logger.error( f"任务 {task_name} 失败 (第 {attempt+1} 次): {e}", exc_info=True, ) if attempt < max_retries: wait = retry_backoff * (2 ** attempt) logger.info(f"等待 {wait}s 后重试...") time.sleep(wait) else: history.record_complete(task_name, False) logger.error( f"任务 {task_name} 重试 {max_retries} 次后仍失败" ) raise return {"status": "failed"} return wrapper return decorator # ---- Celery 集成 ---- from celery import Celery from celery.schedules import crontab app = Celery("tasks", broker="redis://localhost:6379/1") @app.task(bind=True, max_retries=3, default_retry_delay=60) def daily_report_task(self): """每日报表任务""" redis_client = redis.from_url("redis://localhost:6379/0") task_name = "daily_report" with task_lock(redis_client, task_name, lock_ttl=900) as lock: if lock is None: logger.warning("报表任务正在执行中") return "skipped" try: # 1. 查询数据库 # 2. 计算指标 # 3. 生成报表 # 4. 发送邮件 logger.info("报表生成完成") except Exception as e: logger.error(f"报表任务异常: {e}") # Celery 自动重试 raise self.retry(exc=e) return "success" # Celery Beat 配置 app.conf.beat_schedule = { "daily-report": { "task": "daily_report_task", "schedule": crontab(hour=3, minute=0), # 凌晨 3 点 "options": { "expires": 3600, # 1 小时后过期 }, }, } # ---- 使用示例 ---- @reliable_task(lock_ttl=300) def sync_user_data(): """同步用户数据（带自动锁和重试）""" logger.info("开始同步用户数据...") time.sleep(5) # 模拟耗时 # 实际同步逻辑 return {"synced": 1000} if __name__ == "__main__": # 手动执行 result = sync_user_data() print(result) # 第二次立即执行会被跳过 result2 = sync_user_data() print(result2) # {"status": "skipped", "reason": "already_running"}

DistributedLock.release使用 Lua 脚本的原因很容易被忽视，但它是一个关键的原子性保证。
如果用两条 Redis 命令——先 GET 锁值比对，再 DEL 删除——中间可能被另一个进程插入。
Lua 脚本在 Redis 中是原子执行的，要么全部成功要么全部失败，保证了"只有锁的持有者才能释放锁"。

extend方法解决了长任务场景下的锁续期问题。
如果你的任务预估执行 10 分钟，但锁 TTL 只有 5 分钟，你可以每过 3 分钟调用一次extend把 TTL 再延长。
这是一种"心跳续期"模式——只要你还在执行，锁就保持有效；一旦你崩溃，锁在 TTL 后自动释放。

四、锁的可靠性考量

Redis 锁不是完美的。
Redis 宕机可能导致锁丢失。
核心任务建议使用 Redlock 算法或数据库行锁。

锁超时设置要大于任务最长的正常执行时间。
如果锁 TTL 是 600 秒、任务实际需要 700 秒。
任务后半段锁已过期，另一个实例可能启动。
需要通过extend()方法续期或设置充足的 TTL。

在生产环境中，Redis 单节点锁的不确定性来源于主从切换。
如果任务获取了锁，锁数据写入了主节点，主节点崩溃，从节点提升为新主——但这个从节点可能还没有同步到这条锁数据。
新主上没有锁，另一个实例可以获取"同一把锁"，导致两个任务同时执行。

如果你的任务绝对不能重复执行（如对账、结算），建议使用 Redlock 算法（Redis 官方推荐的分布式锁方案）或者直接用数据库的SELECT ... FOR UPDATE行锁。
Redlock 通过在多个独立的 Redis 节点上获取锁来提高可靠性；数据库行锁则依赖数据库的 ACID 特性保证互斥。

五、总结

定时任务可靠性需要三层保障：互斥锁、自动重试、历史记录。
Redis 分布式锁防止重复执行，TTL 防止死锁。
@reliable_task装饰器统一封装锁+重试逻辑。
锁超时需要大于任务最大执行时间，或支持续期。
Celery Beat 配合数据库锁，构建生产级任务调度。

最后一条建议：给你的定时任务加一个监控面板。
用TaskHistory记录的数据生成一个简单的 Grafana 面板——显示每个任务最近一次执行时间、成功率、失败原因。
当某个任务的"距上次成功执行"超过阈时触发告警。
定时任务的特点是"没有人天天盯着看"，异常往往在数据出问题后才被发现——监控面板能把这个时间从"天"缩短到"分钟"。