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

wen python案例 2026-06-27 27

本文目录导读：

文件锁（File Lock）
PID文件方式
使用atexit模块（更可靠）
使用数据库/Redis锁（分布式场景）
使用第三方库
使用Python内置的portalocker库
最佳实践建议

在Python中规避定时任务重复执行,主要有以下几种常用方法：

文件锁（File Lock）

import os
import fcntl
import time
def acquire_lock():
    lock_file = '/tmp/my_script.lock'
    try:
        fd = open(lock_file, 'w')
        fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
        return fd
    except IOError:
        return None
def main():
    lock = acquire_lock()
    if not lock:
        print("脚本已在运行，退出")
        return
    # 执行实际任务
    try:
        print("开始执行任务...")
        time.sleep(30)
        print("任务完成")
    finally:
        fcntl.flock(lock, fcntl.LOCK_UN)
        lock.close()
        try:
            os.remove('/tmp/my_script.lock')
        except:
            pass
if __name__ == "__main__":
    main()

PID文件方式

import os
import sys
import time
class SingleInstance:
    def __init__(self, pid_file='/tmp/my_script.pid'):
        self.pid_file = pid_file
    def check_running(self):
        if os.path.exists(self.pid_file):
            try:
                with open(self.pid_file, 'r') as f:
                    pid = int(f.read().strip())
                # 检查进程是否存在
                os.kill(pid, 0)  # 信号0表示检查进程是否存在
                return True
            except (OSError, ValueError):
                # 进程不存在或PID无效
                pass
        return False
    def create_pid_file(self):
        with open(self.pid_file, 'w') as f:
            f.write(str(os.getpid()))
    def remove_pid_file(self):
        try:
            os.remove(self.pid_file)
        except:
            pass
    def __enter__(self):
        if self.check_running():
            sys.exit("脚本已在运行")
        self.create_pid_file()
        return self
    def __exit__(self, exc_type, exc_val, exc_tb):
        self.remove_pid_file()
# 使用示例
def main():
    with SingleInstance():
        print("开始执行任务...")
        time.sleep(30)
        print("任务完成")
if __name__ == "__main__":
    main()

使用atexit模块（更可靠）

import atexit
import os
import sys
import time
class SingleInstance:
    def __init__(self, lock_file='/tmp/my_app.lock'):
        self.lock_file = lock_file
        self.fd = None
    def acquire(self):
        try:
            self.fd = os.open(self.lock_file, os.O_CREAT | os.O_EXCL | os.O_RDWR)
            return True
        except FileExistsError:
            return False
    def release(self):
        if self.fd:
            os.close(self.fd)
            try:
                os.remove(self.lock_file)
            except:
                pass
# 全局单例
singleton = SingleInstance()
@atexit.register
def cleanup():
    singleton.release()
def main():
    if not singleton.acquire():
        print("脚本已在运行，退出")
        sys.exit(1)
    print("开始执行任务...")
    time.sleep(30)
    print("任务完成")
if __name__ == "__main__":
    main()

使用数据库/Redis锁（分布式场景）

import redis
import time
class RedisLock:
    def __init__(self, redis_client, lock_key='my_task_lock', timeout=3600):
        self.redis = redis_client
        self.lock_key = lock_key
        self.timeout = timeout
    def acquire(self):
        # 使用SET NX命令实现分布式锁
        result = self.redis.setnx(self.lock_key, str(time.time()))
        if result:
            self.redis.expire(self.lock_key, self.timeout)
            return True
        return False
    def release(self):
        self.redis.delete(self.lock_key)
# 使用示例
def main():
    r = redis.Redis(host='localhost', port=6379, db=0)
    lock = RedisLock(r)
    if not lock.acquire():
        print("任务已在执行")
        return
    try:
        print("开始执行任务...")
        time.sleep(30)
        print("任务完成")
    finally:
        lock.release()
if __name__ == "__main__":
    main()

使用第三方库

filelock库（推荐）

pip install filelock

from filelock import FileLock
import time
def main():
    lock = FileLock("/tmp/my_script.lock", timeout=0)  # timeout=0 立即超时
    try:
        with lock.acquire():
            print("开始执行任务...")
            time.sleep(30)
            print("任务完成")
    except:
        print("脚本已在运行，退出")
if __name__ == "__main__":
    main()

使用Python内置的`portalocker`库

import portalocker
import time
def main():
    lock_file = '/tmp/my_script.lock'
    try:
        with open(lock_file, 'w') as f:
            portalocker.lock(f, portalocker.LOCK_EX | portalocker.LOCK_NB)
            print("开始执行任务...")
            time.sleep(30)
            print("任务完成")
    except portalocker.LockException:
        print("脚本已在运行，退出")
if __name__ == "__main__":
    main()