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

如何防止Python脚本重复运行?现有fcntl文件锁代码失效求助

解决Python脚本单实例运行(含锁过期检测)的问题

你的原代码问题在于**fcntl.flock是基于内核的文件建议锁**——它不是通过锁文件的存在性来判断实例是否运行的:如果之前的进程正常退出,内核会自动释放锁,哪怕锁文件还留在磁盘上,这时候你的脚本依然能获取到锁,自然不会触发sys.exit()。另外这种方式仅支持Unix系系统,还没处理进程意外崩溃后锁文件残留但锁已失效的情况。

下面是一个更可靠的实现,包含锁文件有效性验证、进程存活检测、过期时间判断,能确保脚本同一时间仅运行一个实例:

import os
import sys
import time
import fcntl
from typing import Optional

def is_process_running(pid: int) -> bool:
    """检查指定PID的进程是否还在运行(Unix系通用)"""
    try:
        # 发送0信号,仅检查进程是否存在,不会终止进程
        os.kill(pid, 0)
        return True
    except OSError:
        return False

def acquire_lock(lock_file_path: str, expire_seconds: int = 300) -> bool:
    """
    获取单实例运行锁,支持过期检测
    :param lock_file_path: 锁文件存储路径
    :param expire_seconds: 锁过期时间(秒),默认5分钟
    :return: 成功获取锁返回True,否则返回False
    """
    fp: Optional[os.FileDescriptor] = None
    try:
        # 以读写+创建模式打开锁文件
        fp = open(lock_file_path, 'a+')
        # 尝试获取排他锁,非阻塞模式(获取失败立即抛出异常)
        fcntl.flock(fp, fcntl.LOCK_EX | fcntl.LOCK_NB)
        
        # 锁获取成功,写入当前进程PID和锁创建时间戳
        fp.seek(0)
        fp.truncate()
        fp.write(f"{os.getpid()}\n{time.time()}")
        fp.flush()
        return True
    except IOError:
        # 获取锁失败,检查旧锁的有效性
        fp = open(lock_file_path, 'r')
        try:
            lines = fp.readlines()
            if len(lines) >= 2:
                pid = int(lines[0].strip())
                lock_time = float(lines[1].strip())
                current_time = time.time()
                
                # 两种情况可以接管锁:进程已死,或者锁已过期
                if not is_process_running(pid) or (current_time - lock_time) > expire_seconds:
                    fp.close()
                    os.unlink(lock_file_path)
                    # 删除旧锁后重新尝试获取
                    return acquire_lock(lock_file_path, expire_seconds)
        except (ValueError, IndexError):
            # 锁文件内容损坏,直接删除重试
            fp.close()
            os.unlink(lock_file_path)
            return acquire_lock(lock_file_path, expire_seconds)
        finally:
            if fp:
                fp.close()
        # 旧锁有效,返回False
        return False
    finally:
        # 注意:不要在这里关闭文件!关闭会自动释放锁,锁需要保持到进程结束
        pass

# 脚本主逻辑
if __name__ == "__main__":
    LOCK_FILE = "myfile.lock"
    if not acquire_lock(LOCK_FILE):
        print("脚本已在运行,或锁未过期", file=sys.stderr)
        sys.exit(1)
    
    # 这里写你的业务逻辑
    print("脚本开始运行...")
    try:
        # 模拟脚本运行时长
        time.sleep(60)
    finally:
        # 可选:正常退出时清理锁文件(内核会自动释放锁,清理只是更整洁)
        try:
            os.unlink(LOCK_FILE)
        except OSError:
            pass

关键逻辑说明

  1. 进程存活检测:用os.kill(pid, 0)检查PID是否存在,这是Unix系下最可靠的方式,不会对目标进程造成任何影响。
  2. 过期锁处理:锁文件中记录创建时间,超过设定时长则判定为无效锁,允许新实例接管。
  3. 锁的自动释放:即使脚本意外崩溃,内核会在进程终止后自动释放fcntl.flock锁;正常退出时可以主动删除锁文件保持磁盘整洁。
  4. 异常兼容:处理锁文件损坏、权限不足等边缘情况,避免脚本卡死。

如果需要兼容Windows系统,可以改用跨平台锁库portalocker(需先pip install portalocker),核心逻辑类似,仅替换锁的实现部分即可。

内容的提问来源于stack exchange,提问作者MoreScratch

相关产品推荐
方舟 Agent Plan

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

最近更新时间:2026.05.28 06:40:00