如何用Python实现基于链接/文件夹创建的跨平台协作文件锁

我完全理解你的需求：一个轻量、跨平台且能在NFS这类远程分区上可靠运行的文件锁机制，还要无第三方依赖、代码简洁——原子操作确实是实现这类锁的最优路径，因为它能从根本上避免竞态条件。

下面是基于Python标准库实现的方案，核心利用原子文件创建（NFS v3+和几乎所有现代操作系统都支持这个特性）来实现排他锁：

import os
import time
from typing import Optional

class AtomicFileLock:
    def __init__(self, lock_path: str, timeout: float = 30.0):
        self.lock_path = lock_path
        self.timeout = timeout
        self._lock_fd: Optional[int] = None

    def acquire(self, block: bool = True) -> bool:
        """获取锁，返回是否成功"""
        start_time = time.time()
        while True:
            try:
                # 原子创建锁文件：O_CREAT | O_EXCL 确保只有一个进程能成功
                self._lock_fd = os.open(
                    self.lock_path,
                    os.O_CREAT | os.O_EXCL | os.O_WRONLY,
                    0o644
                )
                # 写入当前进程信息（可选，用于调试和更严谨的锁检查）
                os.write(self._lock_fd, f"{os.getpid()}".encode())
                os.close(self._lock_fd)
                self._lock_fd = None
                return True
            except FileExistsError:
                if not block:
                    return False
                # 检查锁是否过期，避免进程崩溃导致的死锁
                if time.time() - start_time > self.timeout:
                    return False
                # 检查锁文件的修改时间
                try:
                    mtime = os.path.getmtime(self.lock_path)
                    if time.time() - mtime > self.timeout:
                        # 锁已过期，强制删除（谨慎使用，根据业务场景调整）
                        os.remove(self.lock_path)
                except FileNotFoundError:
                    # 锁文件已被其他进程删除，继续尝试
                    pass
                time.sleep(0.1)

    def release(self) -> None:
        """释放锁"""
        try:
            os.remove(self.lock_path)
        except FileNotFoundError:
            # 锁文件已不存在，可能已被超时清理或其他进程释放
            pass

    def __enter__(self):
        self.acquire()
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.release()

关键特性说明 #

• 原子性保障：os.open的O_CREAT | O_EXCL组合是核心——这个操作要么成功创建文件（获取锁），要么因为文件已存在而抛出异常，完全没有中间状态，完美避免竞态条件，NFS环境下同样有效。
• 跨平台兼容：仅依赖Python标准库os和time，Windows、Linux、macOS都能直接运行。
• 无第三方依赖：完全符合你的“特定开源代码需无依赖”的要求。
• 超时机制：内置超时检查，避免因进程意外崩溃导致的锁残留死锁，你可以根据业务场景调整timeout参数。
• 上下文管理器支持：用with语句自动管理锁的获取和释放，代码更简洁安全：

# 示例用法
lock_file_path = "/nfs/shared/locks/my_resource.lock"

with AtomicFileLock(lock_file_path, timeout=60):
    # 在这里执行需要加锁的操作，比如读写共享文件
    print("获取到锁，执行临界区操作...")
    time.sleep(5)
# 退出with块后自动释放锁

注意事项 #

1. NFS版本要求：确保你的NFS服务器是v3或更高版本（v2不支持O_EXCL的原子性），大部分现代NFS部署都满足这个条件。
2. 超时的谨慎性：强制删除过期锁的逻辑是一把双刃剑——如果你的临界区操作确实需要超过timeout的时间，可能会导致锁被误删。如果需要更严谨的验证，可以在锁文件中写入进程ID和主机名，释放或检查超时前先验证这些信息。
3. 锁文件路径：必须确保所有协作进程都能访问到锁文件所在的NFS路径，且有读写权限。

内容的提问来源于stack exchange，提问作者Eric O. Lebigot