You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

如何实现同时启动的两个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

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.08.12 23:55:26