定时任务脚本如何避免重复执行

本文目录导读：

1. 文件锁（最常用）
2. 数据库锁
3. Redis分布式锁
4. 使用系统工具
5. 应用层方案（Python示例）
6. 进阶方案：使用ETCD/Consul
7. 最佳实践建议
8. 注意事项

避免定时任务重复执行是一个常见的运维和开发问题，以下是几种主流的解决方案,从简单到复杂排序：

文件锁（最常用）

#!/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()

注意事项

1. 处理死锁：设置锁的超时时间
2. 清理机制：确保锁在异常时能被清理
3. 幂等性：任务本身应该是幂等的
4. 监控告警：记录锁竞争情况
5. 手动干预：提供强制解锁的方法

选择哪种方案取决于你的具体场景：

• 单机任务：flock最简单可靠
• 微服务/集群：Redis/ETCD分布式锁
• 数据库中心：基于数据库的乐观锁