如何在Python中实现Linux文件的锁定及锁定状态检查

在Linux环境下，跨进程的文件锁可以通过Python内置的fcntl模块实现，这是系统级别的排他锁机制，能可靠保证多进程间的文件互斥访问。以下是匹配你需求的完整实现：

核心代码实现 #

import fcntl
import os
import time
from pathlib import Path

# 存储已锁定文件的句柄，防止被GC回收导致锁意外释放
locked_file_handles = {}

def get_file_path_from_poll(poll_dir="/path/to/your/target_dir"):
    """从指定目录获取待处理文件（示例逻辑，可根据业务调整）"""
    dir_path = Path(poll_dir)
    # 遍历目录下的普通文件，跳过已锁定的文件
    for file_path in dir_path.glob("*"):
        if file_path.is_file() and str(file_path) not in locked_file_handles:
            return str(file_path)
    return None

def try_lock_file(file_path):
    """尝试获取文件排他锁，成功返回True，失败返回False"""
    fd = None
    try:
        # 打开文件（确保文件存在且支持读写）
        fd = os.open(file_path, os.O_RDWR | os.O_CREAT)
        # 非阻塞模式加排他锁：LOCK_EX=排他锁，LOCK_NB=不等待直接返回
        fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
        locked_file_handles[file_path] = fd
        return True
    except BlockingIOError:
        # 锁已被其他进程持有
        if fd:
            os.close(fd)
        return False
    except Exception as e:
        print(f"锁定文件{file_path}失败: {str(e)}")
        if fd:
            os.close(fd)
        return False

def release_lock(file_path):
    """释放文件锁并关闭句柄"""
    if file_path in locked_file_handles:
        fd = locked_file_handles.pop(file_path)
        try:
            fcntl.flock(fd, fcntl.LOCK_UN)
            os.close(fd)
        except Exception as e:
            print(f"释放文件{file_path}锁失败: {str(e)}")

def main():
    locked = False
    while not locked:
        file_path = get_file_path_from_poll()
        if not file_path:
            # 无待处理文件时短暂休眠，避免空循环占用CPU
            time.sleep(1)
            continue
        
        locked = try_lock_file(file_path)
        if locked:
            try:
                # --------------------------
                # 这里替换为你的业务逻辑
                print(f"开始处理文件: {file_path}")
                time.sleep(3)  # 模拟业务操作耗时
                # --------------------------
            finally:
                # 无论业务成功与否，必须释放锁
                release_lock(file_path)
                locked = True

if __name__ == "__main__":
    main()

关键说明 #

• 锁机制：使用fcntl.flock的非阻塞排他锁，确保进程不会因等待锁而阻塞，完全匹配你原代码的循环逻辑。
• 句柄管理：通过locked_file_handles字典保存已锁定文件的句柄，防止Python垃圾回收自动关闭句柄导致锁失效。
• 可靠性保障：业务逻辑放在try-finally块中，确保即使代码抛出异常，锁也能被正常释放，避免死锁。
• 自定义适配：get_file_path_from_poll函数可根据你的实际需求修改，比如按文件创建时间排序、过滤特定后缀名等。

内容的提问来源于stack exchange，提问作者Julias