Linux/UNIX下Python脚本互斥运行问题求助：避免并发执行报错

为什么threading.Lock()没用？ #

threading.Lock()是线程锁，只能限制同一个进程内的多个线程，不同进程的锁实例完全独立，所以同时运行多个脚本进程时，锁根本起不到互斥作用。要实现进程间互斥，必须用进程级别的锁，比如文件锁。

解决方案一：用文件锁实现进程互斥（推荐） #

通过创建一个锁文件，借助系统级的文件排他锁控制，只有拿到锁的进程才能执行删除逻辑，其他进程直接提示退出。同时在遍历文件时增加异常处理，避免文件被删后报错。

修改后的代码示例：

import os
import datetime
from datetime import date, timedelta
# 跨平台文件锁处理
import sys
if sys.platform.startswith('win'):
    import msvcrt
else:
    import fcntl

def delete_files(ini_path, days_num):
    config = read_config_file(ini_path)
    # 锁文件路径，Windows可改为类似"C:\\temp\\delete_files.lock"
    lock_file_path = "/tmp/delete_files.lock"  
    lock_fd = None
    try:
        # 打开锁文件，不存在则自动创建
        lock_fd = open(lock_file_path, 'w')
        # 获取非阻塞排他锁
        if sys.platform.startswith('win'):
            msvcrt.locking(lock_fd.fileno(), msvcrt.LK_NBLCK, 1)
        else:
            fcntl.flock(lock_fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
        
        # 拿到锁后执行核心逻辑
        log_processes(str(config[1]))
        folder_path = str(config[0])
        if not os.listdir(folder_path):
            print("Empty list")
        else:
            print("List loaded")
        
        # 遍历文件时处理文件已被删除的情况
        for file in os.listdir(folder_path):       
            file_path = os.path.join(folder_path, file)
            try:
                if os.path.isdir(file_path):
                    continue
                file_datestamp = datetime.fromtimestamp(os.path.getmtime(file_path))
                start_of_deletion_date = date.today() - timedelta(days=int(days_num))
                if date.today() > file_datestamp.date() >= start_of_deletion_date:
                    os.remove(file_path)
                else:
                    compress_file(file_path)
            except FileNotFoundError:
                # 文件已被其他进程删除，直接跳过
                continue
            except Exception as e:
                log_processes(config[1], e)
                print(f"处理文件{file_path}出错: {e}")
                
    except BlockingIOError:
        # 拿不到锁，说明已有进程在运行
        print("脚本已在运行中，请稍后再试")
    except Exception as e:
        log_processes(config[1], e)
        print(e)
    finally:
        # 释放锁并关闭文件
        if lock_fd:
            if sys.platform.startswith('win'):
                try:
                    msvcrt.locking(lock_fd.fileno(), msvcrt.LK_UNLCK, 1)
                except:
                    pass
            else:
                try:
                    fcntl.flock(lock_fd, fcntl.LOCK_UN)
                except:
                    pass
            lock_fd.close()

关键改动说明 #

• 进程锁实现：用系统级文件锁保证同一时间只有一个进程进入核心逻辑
• 非阻塞锁：拿不到锁直接提示退出，无需等待
• 异常捕获：增加FileNotFoundError捕获，遇到已被删除的文件直接跳过
• 路径优化：用os.path.join替代手动拼接路径，避免格式错误

解决方案二：用PID文件（简单但有缺陷） #

创建一个PID文件记录当前运行的进程ID，启动时检查PID对应的进程是否存活：

import os
import psutil

def check_running():
    pid_file = "/tmp/delete_files.pid"
    if os.path.exists(pid_file):
        with open(pid_file, 'r') as f:
            pid = int(f.read().strip())
        if psutil.pid_exists(pid):
            return True
        else:
            # 进程已异常退出，删除无效PID文件
            os.remove(pid_file)
    # 写入当前进程PID
    with open(pid_file, 'w') as f:
        f.write(str(os.getpid()))
    return False

def delete_files(ini_path, days_num):
    if check_running():
        print("脚本已在运行中，请稍后再试")
        return
    # 后续原有逻辑...

这种方法的缺陷是：如果进程异常崩溃，PID文件可能未被删除，导致后续脚本无法正常启动，可靠性不如文件锁。

内容的提问来源于stack exchange，提问作者Rabie Moses Santillan