如何实现同时启动的两个Cron触发Python脚本仅一个执行?
解决Cron重复触发Python脚本的互斥执行问题
针对两个Cron服务同时触发同一Python脚本,需要保证仅一个实例执行的需求,以下是几种可靠的实现方案:
方法一:原子性文件锁(推荐)
利用操作系统原子性文件创建的特性,直接避免竞争条件。只有第一个进程能成功创建锁文件,后续进程会因文件已存在直接退出。
import os import sys LOCK_FILE = "/tmp/my_script.lock" def acquire_lock(): try: # O_EXCL标记确保文件不存在时才创建,原子操作无竞争 fd = os.open(LOCK_FILE, os.O_CREAT | os.O_EXCL | os.O_RDWR) os.close(fd) return True except FileExistsError: return False def release_lock(): try: os.remove(LOCK_FILE) except FileNotFoundError: pass if __name__ == "__main__": if not acquire_lock(): print("已有脚本实例在运行,退出") sys.exit(1) try: # 此处编写你的业务逻辑 print("执行脚本核心逻辑...") finally: # 无论执行成功/失败,最终释放锁 release_lock()
方法二:PID锁文件(处理锁残留)
在锁文件中写入当前进程PID,启动时不仅检查锁文件是否存在,还验证PID对应的进程是否存活,避免脚本意外崩溃导致锁残留的问题。
import os import sys import psutil LOCK_FILE = "/tmp/my_script.pid" def acquire_lock(): if os.path.exists(LOCK_FILE): with open(LOCK_FILE, "r") as f: pid = int(f.read().strip()) # 检查PID对应的进程是否还在运行 if psutil.pid_exists(pid): return False else: # 清理无效锁文件 os.remove(LOCK_FILE) # 原子性写入当前PID try: with open(LOCK_FILE, "x") as f: f.write(str(os.getpid())) return True except FileExistsError: # 被其他进程抢先创建锁文件 return False def release_lock(): try: os.remove(LOCK_FILE) except FileNotFoundError: pass if __name__ == "__main__": if not acquire_lock(): print("已有脚本实例在运行,退出") sys.exit(1) try: # 此处编写你的业务逻辑 print("执行脚本核心逻辑...") finally: release_lock()
注:需先安装psutil库:pip install psutil
方法三:flock文件锁(系统自动释放)
利用Linux的flock机制,通过文件描述符加排他锁,进程退出时系统会自动释放锁,无需手动删除锁文件。
import os import sys import fcntl LOCK_FILE = "/tmp/my_script.flock" def acquire_lock(): try: fd = open(LOCK_FILE, "w") # 非阻塞排他锁,拿不到锁直接抛出异常 fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB) return fd except BlockingIOError: return None if __name__ == "__main__": lock_fd = acquire_lock() if not lock_fd: print("已有脚本实例在运行,退出") sys.exit(1) try: # 此处编写你的业务逻辑 print("执行脚本核心逻辑...") finally: # 关闭文件描述符自动释放锁 lock_fd.close()
关于你提到的时间戳锁方案的问题
基于时间戳生成锁文件的思路存在原子性缺失的致命问题:两个进程可能在同一时间窗口(甚至同一毫秒)生成相同的时间标识,导致锁判断失效,因此不推荐使用。
内容的提问来源于stack exchange,提问作者paras goyal
相关产品推荐
相关产品推荐

