本文目录导读：
避免定时任务重复执行是一个常见的运维和开发问题，以下是几种主流的解决方案,从简单到复杂排序：
文件锁（最常用）
#!/bin/bash
LOCKFILE="/tmp/my_task.lock"
# 检查锁文件是否存在
if [ -f "$LOCKFILE" ]; then
echo "任务已在运行中，退出"
exit 1
fi
# 创建锁文件
touch "$LOCKFILE"
# 任务执行
echo "开始执行任务..."
sleep 10
echo "任务完成"
# 清理锁文件
rm -f "$LOCKFILE"
优化版（带PID检查）：
LOCKFILE="/tmp/my_task.lock"
# 检查锁文件并验证进程是否存在
if [ -f "$LOCKFILE" ]; then
PID=$(cat "$LOCKFILE")
if kill -0 "$PID" 2>/dev/null; then
echo "进程 $PID 仍在运行，退出"
exit 1
fi
# 进程已不存在，删除过期锁
rm -f "$LOCKFILE"
fi
# 写入当前PID
echo $$ > "$LOCKFILE"
# 确保退出时清理
trap "rm -f $LOCKFILE" EXIT
# 执行任务
# ...
数据库锁
-- 创建锁表
CREATE TABLE task_locks (
task_name VARCHAR(100) PRIMARY KEY,
lock_time TIMESTAMP,
expire_time TIMESTAMP
);
-- 获取锁（MySQL）
INSERT INTO task_locks (task_name, lock_time, expire_time)
VALUES ('my_task', NOW(), DATE_ADD(NOW(), INTERVAL 1 HOUR))
ON DUPLICATE KEY UPDATE
lock_time = IF(expire_time < NOW(), VALUES(lock_time), lock_time),
expire_time = IF(expire_time < NOW(), VALUES(expire_time), expire_time);
-- 检查是否获取成功
SELECT ROW_COUNT() > 0 AS got_lock;
Redis分布式锁
import redis
import time
r = redis.Redis(host='localhost', port=6379, db=0)
def acquire_lock(lock_name, expire_time=10):
"""获取分布式锁"""
lock_key = f"task_lock:{lock_name}"
# 使用SETNX + EXPIRE原子操作
return r.set(lock_key, "locked", nx=True, ex=expire_time)
def release_lock(lock_name):
"""释放锁"""
lock_key = f"task_lock:{lock_name}"
r.delete(lock_key)
# 使用示例
if acquire_lock("my_task", timeout=30):
try:
# 执行任务
print("开始执行任务")
time.sleep(5)
finally:
release_lock("my_task")
else:
print("任务已被其他实例执行")
使用系统工具
flock（Linux推荐）
#!/bin/bash # 使用flock确保唯一实例 exec 200>/var/lock/my_task.lock flock -n 200 || exit 1 # 任务代码 echo "Start task at $(date)" sleep 10 echo "End task at $(date)"
使用cron+flock
# crontab配置 */5 * * * * /usr/bin/flock -n /tmp/my_task.lock /path/to/script.sh
应用层方案（Python示例）
import fcntl
import sys
import os
class SingleInstance:
def __init__(self, lock_file="/tmp/my_app.lock"):
self.lock_file = lock_file
self.fp = None
def __enter__(self):
self.fp = open(self.lock_file, 'w')
try:
fcntl.flock(self.fp.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
except IOError:
print("另一个实例正在运行")
sys.exit(1)
return self
def __exit__(self, *args):
fcntl.flock(self.fp.fileno(), fcntl.LOCK_UN)
self.fp.close()
os.remove(self.lock_file)
# 使用
with SingleInstance("/tmp/my_task.lock"):
# 执行任务
print("Running task...")
进阶方案：使用ETCD/Consul
import etcd3
client = etcd3.client(host='localhost', port=2379)
# 获取租约
lease = client.lease(ttl=30) # 30秒过期
# 尝试获取锁
lock_key = "/locks/my_task"
try:
# 使用租约创建key，自动过期防止死锁
client.put(lock_key, "locked", lease=lease)
except etcd3.exceptions.RepeatedKeyError:
print("锁已被占用")
exit(1)
try:
# 执行任务
print("Executing task...")
finally:
# 释放锁
client.delete(lock_key)
最佳实践建议
简单场景（单机）
#!/bin/bash
# 使用flock + timeout组合
LOCKFILE="/tmp/my_task.lock"
TIMEOUT=300 # 5分钟超时
(
flock -x -w $TIMEOUT 200 || {
echo "无法获取锁（超时或已被占用）"
exit 1
}
echo "任务开始: $(date)"
# 你的任务代码
sleep 60
echo "任务结束: $(date)"
) 200>$LOCKFILE
分布式场景（推荐Redis）
import redis
import uuid
import time
class TaskLock:
def __init__(self, redis_client, lock_name):
self.redis = redis_client
self.lock_name = f"task_lock:{lock_name}"
self.identifier = str(uuid.uuid4())
self.lock_timeout = 30
def acquire(self):
return self.redis.set(
self.lock_name,
self.identifier,
nx=True,
ex=self.lock_timeout
)
def release(self):
# 使用Lua脚本确保原子性
script = """
if redis.call("get",KEYS[1]) == ARGV[1] then
return redis.call("del",KEYS[1])
else
return 0
end
"""
return self.redis.eval(script, 1, self.lock_name, self.identifier)
# 使用
lock = TaskLock(redis_client, "daily_report")
if lock.acquire():
try:
generate_daily_report()
finally:
lock.release()
注意事项
- 处理死锁：设置锁的超时时间
- 清理机制：确保锁在异常时能被清理
- 幂等性：任务本身应该是幂等的
- 监控告警：记录锁竞争情况
- 手动干预：提供强制解锁的方法
选择哪种方案取决于你的具体场景：
- 单机任务：flock最简单可靠
- 微服务/集群：Redis/ETCD分布式锁
- 数据库中心：基于数据库的乐观锁
