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

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 flock: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, nxTrue, # 仅当不存在时设置 exself.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 ftask:history:{task_name}:last_start self.redis.set(key, time.time()) def record_complete(self, task_name: str, success: bool): 记录任务完成 key ftask: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 ftask: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_ttl900) 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} (第 {attempt1}/{max_retries1} 次) ) # 执行任务 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} 失败 (第 {attempt1} 次): {e}, exc_infoTrue, ) 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, brokerredis://localhost:6379/1) app.task(bindTrue, max_retries3, default_retry_delay60) 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_ttl900) 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(exce) return success # Celery Beat 配置 app.conf.beat_schedule { daily-report: { task: daily_report_task, schedule: crontab(hour3, minute0), # 凌晨 3 点 options: { expires: 3600, # 1 小时后过期 }, }, } # ---- 使用示例 ---- reliable_task(lock_ttl300) 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 面板——显示每个任务最近一次执行时间、成功率、失败原因。当某个任务的距上次成功执行超过阈时触发告警。定时任务的特点是没有人天天盯着看异常往往在数据出问题后才被发现——监控面板能把这个时间从天缩短到分钟。