本文目录导读：
- 数据库乐观锁（最推荐，适合有数据库的场景）
- Redis 分布式锁（适合高性能、跨服务器场景）
- 文件锁（适合纯本地、无外部依赖的场景）
- 任务自身幂等性（终极方案）
- 调度器层面的限制（适用于Crontab）
- 如何选择？
- 特别提醒
- 一个综合示例（Python + Redis + 幂等）
定时脚本并发执行时避免冲突，核心思路是确保同一时刻只有一个实例在处理关键资源，根据不同的场景和资源类型，有几种主流的解决方案,按推荐程度排序如下：
数据库乐观锁（最推荐，适合有数据库的场景）
这是最通用、实现成本最低的方法，在数据库表中增加一个版本号字段（如 version 或 updated_at）。
原理：
-- 伪代码：更新时检查版本号是否和读取时一致 UPDATE tasks SET status = 'processing', version = version + 1 WHERE id = ? AND status = 'pending' AND version = ?
UPDATE 影响的行数（affected_rows）为0，说明状态已被其他实例修改,当前实例直接放弃或重试。
优点： 无需额外组件，数据库自带事务保证。 缺点： 不适合无数据库的脚本（如纯文件处理）。
Redis 分布式锁（适合高性能、跨服务器场景）
使用 Redis 的 SET NX PX 命令创建一把全局锁。
代码示例（Python + Redis）：
import redis
import time
r = redis.Redis(decode_responses=True)
lock_key = "cron:my_task_lock"
# 尝试加锁，过期时间30秒，防止死锁
locked = r.set(lock_key, "1", nx=True, px=30000)
if locked:
try:
print("获得锁，开始执行任务...")
# 执行你的定时逻辑
time.sleep(5)
finally:
# 释放锁（建议用Lua脚本保证原子性）
r.delete(lock_key)
else:
print("未获得锁，跳过本次执行")
进阶（Redlock算法）： 在分布式系统中，单点Redis有风险，高可用场景需考虑Redlock，但大多数中小项目,单点Redis配合自动过期已足够。
文件锁（适合纯本地、无外部依赖的场景）
对于在单台服务器上执行的简单脚本,文件锁非常轻量。
Shell 示例（flock）：
# 使用文件描述符加锁
(
flock -n 200 || exit 1
echo "开始执行任务..."
# 你的脚本逻辑
sleep 60
) 200>/var/run/my_script.lock
flock -n 无法获得锁（另一个实例正在运行），exit 1 会直接退出,不会等待。
Python 示例（fcntl）：
import fcntl
import sys
lock_file = open('/var/run/my_script.lock', 'w')
try:
fcntl.flock(lock_file, fcntl.LOCK_EX | fcntl.LOCK_NB)
print("获得锁")
# 执行任务
except BlockingIOError:
print("锁被占用，跳过")
sys.exit(1)
任务自身幂等性（终极方案）
核心思想： 让任务执行多次也不会产生错误结果。
常见实现：
- 数据库： 使用
INSERT ... ON DUPLICATE KEY UPDATE或REPLACE INTO。 - 消息队列： 确保消费者支持消息去重（如记录已处理的唯一ID）。
- 文件处理： 处理前检查文件是否已被处理（如记录文件Hash）。
- API调用： 接口设计为天然幂等（多次请求效果一致）。
关键： 即使锁机制失效（如 Redis 挂了），幂等性可以作为最后一道防线,防止数据错乱。
调度器层面的限制（适用于Crontab）
方式： 使用 flock 直接在 crontab 中限制。
# 每分钟执行一次，但若前一次未执行完则跳过 * * * * * /usr/bin/flock -n /tmp/my_task.lock /usr/bin/python /path/to/script.py
如何选择？
| 场景 | 推荐方案 | 理由 |
|---|---|---|
| 有数据库 | 乐观锁 | 零额外依赖，稳定可靠 |
| 跨服务器、高并发 | Redis分布式锁 | 性能高，易于协调 |
| 单机、极简单脚本 | 文件锁 (flock) | 最轻量，无需任何配置 |
| 处理不可重试的任务 | 幂等设计 + 上述任一种 | 双重保障 |
| 不想改变代码 | 调度器层面flock | 无需修改脚本，配置即生效 |
特别提醒
- 锁超时时间：预估任务最大执行时间，设置超时时间（例如任务最长10s，设置超时15s），过短会导致任务未完成锁被释放,过长可能导致死锁。
- 锁释放：一定使用
try...finally或封装好的资源管理器确保锁最终释放。 - 加锁粒度：尽可能细粒度，例如不要锁整个“数据同步”任务，而是锁“同步某张表”的资源,提高并发吞吐量。
一个综合示例（Python + Redis + 幂等）
def execute_with_lock(task_id, unique_key):
lock_key = f"cron:{task_id}"
dedup_key = f"cron_dedup:{unique_key}"
# 1. 检查是否已处理（幂等性）
if redis.exists(dedup_key):
print("任务已执行过，跳过")
return
# 2. 获取分布式锁
if not redis.set(lock_key, "1", nx=True, ex=300):
print("其他实例正在执行，跳过")
return
try:
# 3. 执行任务主逻辑
process_data()
# 4. 标记为已处理（幂等去重）
redis.setex(dedup_key, 86400, "1") # 24小时过期
finally:
# 5. 释放锁
redis.delete(lock_key)
这种模式基本可以覆盖99%的冲突场景，根据你的具体技术栈选择最轻量、最贴合的方式即可。
