本文目录导读：
- 文件锁（最简单，适合单机单进程）
- 数据库乐观锁 / 分布式锁（适合集群/微服务）
- 状态机 + 幂等性设计（最可靠）
- 系统级调度工具（推荐用于定时任务）
- 针对不同场景的建议
- 最优雅的写法（Python + Redis）
对于脚本任务重复执行的问题,通常可以在任务启动前、任务执行中和任务完成后三个阶段设置防护机制，以下是几种成熟且经过实践检验的解决方案，按复杂度和适用场景排序：
文件锁（最简单，适合单机单进程）
这是最经典的Unix方式,通过创建一个临时文件来标识任务正在运行。
原理：尝试创建一个独占锁文件，如果文件已存在则退出。
Bash/Shell 示例：
LOCKFILE="/tmp/myscript.lock"
# 强制删除过期锁（防止脚本崩溃后锁残留，建议保留）
if [ -f "$LOCKFILE" ]; then
# 检查持有锁的进程是否还在运行
if kill -0 $(cat "$LOCKFILE") 2>/dev/null; then
echo "[ERROR] 脚本已在运行中 (PID: $(cat $LOCKFILE))"
exit 1
else
# 进程已死，清理旧锁
rm -f "$LOCKFILE"
fi
fi
# 创建新锁，记录当前PID
echo $$ > "$LOCKFILE"
# ---- 在这里写你的任务逻辑 ----
sleep 30 # 模拟耗时任务
# -------------------------
# 任务结束，释放锁
rm -f "$LOCKFILE"
Python 示例（使用fcntl多进程安全）：
import fcntl
import sys
lock_file = '/tmp/myapp.lock'
with open(lock_file, 'w') as f:
try:
fcntl.flock(f, fcntl.LOCK_EX | fcntl.LOCK_NB)
except IOError:
print("脚本已在运行中", file=sys.stderr)
sys.exit(1)
# 你的任务逻辑
...
# 自动释放锁（with语句结束）
缺点：如果服务器重启，锁文件可能残留，可配合信号处理或定时清理。
数据库乐观锁 / 分布式锁（适合集群/微服务）
当多个服务实例可能同时触发任务时,需要共享的锁存储。
方案A：MySQL 行锁
-- 创建任务状态表
CREATE TABLE task_lock (
task_name VARCHAR(50) PRIMARY KEY,
status ENUM('idle', 'running', 'finished') DEFAULT 'idle',
start_time DATETIME,
node_id VARCHAR(50)
);
-- 获取锁（原子操作）
UPDATE task_lock
SET status = 'running', start_time = NOW(), node_id = 'node-001'
WHERE task_name = 'daily_report' AND status = 'idle';
-- 如果影响行数>0，说明获取锁成功
方案B：Redis 分布式锁（高并发推荐）
import redis
import time
r = redis.Redis()
lock_key = "task:report:lock"
# 尝试获取锁，过期时间5分钟，防止死锁
if r.setnx(lock_key, "locked"):
r.expire(lock_key, 300) # 5分钟超时
try:
# 执行业务逻辑
run_report()
finally:
r.delete(lock_key) # 释放锁
else:
print("任务正在其他节点执行中")
注意：使用 Redis 时务必设置过期时间，防止节点崩溃导致锁永久占用。
状态机 + 幂等性设计（最可靠）
从业务层面保证即使多次执行也不会产生副作用,这是最彻底的方案。
核心原则：
- 任务具有唯一ID（如日期、批次号）。
- 执行前检查：去数据库/文件检查该ID是否已被处理。
- 执行后记录：将处理结果写入持久化存储。
示例：
def process_daily_order(today):
# 1. 幂等性检查
if db.exists("processed_dates", today):
print(f"{today} 已处理过，跳过")
return
# 2. 执行实际逻辑
data = fetch_orders(today)
process(data)
# 3. 记录已完成（事务性写入）
db.insert("processed_dates", date=today, status="done")
优点：即使脚本被多次触发，或中途崩溃重启，也不会重复处理数据。
系统级调度工具（推荐用于定时任务）
如果任务是定时执行的,使用专业调度器比写脚本来控制重复更可靠。
Linux crontab 结合 flock
# 使用flock命令包裹，自动处理文件锁 * * * * * /usr/bin/flock -n /tmp/myjob.lock /usr/local/bin/my_script.sh
systemd timer + 服务单元
[Unit] Description=My Daily Job [Service] Type=oneshot ExecStart=/usr/local/bin/my_script.sh
专业调度系统：Airflow、Celery Beat、Jenkins等，自带去重和状态管理。
针对不同场景的建议
| 场景 | 最佳方案 | 原因 |
|---|---|---|
| 单机Cron任务 | flock 命令 |
零代码侵入，Linux自带 |
| Kubernetes Job | Leader Election + 状态数据库 | 适合容器化环境 |
| 微服务定时任务 | 分布式锁 (Redis/Etcd) | 跨节点协调 |
| 数据仓库ETL | 幂等性设计 + 调度框架 | 数据准确性优先 |
| 低代码/快速开发 | 文件锁 + PID检测 | 实现简单，依赖少 |
最优雅的写法（Python + Redis）
如果你想找一个几乎通用的方案,可以考虑：
import redis
from contextlib import contextmanager
@contextmanager
def task_lock(task_id, ttl=600):
"""分布式任务锁上下文管理器"""
r = redis.Redis()
lock_key = f"lock:{task_id}"
if not r.setnx(lock_key, 'locked'):
raise RuntimeError(f"任务 {task_id} 已在执行")
r.expire(lock_key, ttl)
try:
yield
finally:
r.delete(lock_key)
# 使用
with task_lock("daily_cleanup"):
# 这里面的代码保证只会有一个实例执行
run_cleanup()
为什么推荐这个？
- 原子性：
setnx保证并发安全 - 自动过期：防止死锁
- 上下文管理：自动释放，不怕异常
- 可扩展：只需改为 Redis Cluster 即可支持集群
不要试图在脚本内部通过“等待再检查”来规避重复——这只会增加不确定性。最可靠的方法是在外部设置一把锁（文件/数据/分布式），然后让脚本在获取锁失败时直接退出，加上幂等设计作为兜底，就能彻底解决重复执行问题。
