Linux下Python如何判断文件是否被其他进程打开写入且未关闭？

首先明确：可以判断文件是否处于写入未关闭状态，但这种方法存在竞态条件和性能问题，更推荐用原子操作或锁机制从根源避免冲突，解决你遇到的读进程读取未完成文件的问题。

一、不推荐：直接判断文件是否被写入中 #

在Linux上可以通过/proc文件系统检查文件是否被其他进程以写入模式打开，但缺点很明显：

• 遍历所有进程的文件描述符，性能差；
• 存在竞态条件——刚检查完文件就被关闭/打开；
• 需要root权限才能查看所有进程的fd信息。

示例代码（仅作演示，不推荐生产用）：

import os

def is_file_being_written(file_path):
    try:
        file_inode = os.stat(file_path).st_ino
    except FileNotFoundError:
        return False

    for pid in os.listdir('/proc'):
        if not pid.isdigit():
            continue
        fd_dir = f'/proc/{pid}/fd'
        try:
            for fd in os.listdir(fd_dir):
                try:
                    link_path = os.readlink(f'{fd_dir}/{fd}')
                    if link_path != file_path:
                        continue
                    # 检查文件打开模式是否为写入
                    fd_info_path = f'/proc/{pid}/fdinfo/{fd}'
                    with open(fd_info_path, 'r') as f_info:
                        for line in f_info:
                            if line.startswith('flags:'):
                                flags = int(line.split()[1], 8)
                                # O_WRONLY(0o2) 或 O_RDWR(0o20000) 表示可写入
                                if (flags & 0o2) or (flags & 0o20000):
                                    return True
                except (OSError, PermissionError):
                    continue
        except (OSError, PermissionError):
            continue
    return False

二、推荐：从根源避免冲突的可靠方案 #

1. 原子重命名（最优解） #

Linux下同一文件系统内的文件重命名是原子操作——要么重命名完成，要么完全没发生。写进程先写到临时文件，写完后再重命名为目标文件名，读进程只读取最终的目标文件。

写进程代码：

import os

temp_path = 'cache.tmp'
target_path = 'cache.dat'

# 先写入临时文件
with open(temp_path, 'w') as f:
    f.write('你的缓存数据')
    # 强制把数据刷到磁盘，避免系统缓存延迟
    f.flush()
    os.fsync(f.fileno())

# 原子替换目标文件，读进程只会看到完整的文件
os.rename(temp_path, target_path)

读进程代码：

target_path = 'cache.dat'

if os.path.exists(target_path):
    with open(target_path, 'r') as f:
        data = f.read()

这个方案完全不会出现读进程读取到半完成文件的情况，性能也最优。

2. fcntl文件锁（适合多进程读写场景） #

利用Linux的fcntl实现建议性锁：写进程加排他锁，读进程加共享锁，确保同一时间只有一个进程写入，或多个进程读取。

写进程代码：

import fcntl
import os

file_path = 'cache.dat'

with open(file_path, 'w') as f:
    # 加排他锁，其他进程的锁请求会等待
    fcntl.flock(f, fcntl.LOCK_EX)
    f.write('你的缓存数据')
    f.flush()
    os.fsync(f.fileno())
    # 文件关闭时自动释放锁

读进程代码：

import fcntl
import os

file_path = 'cache.dat'

if os.path.exists(file_path):
    with open(file_path, 'r') as f:
        # 加共享锁，等待排他锁释放后再读取
        fcntl.flock(f, fcntl.LOCK_SH)
        data = f.read()

注意：建议性锁需要所有进程都遵守锁规则，如果有进程直接跳过锁读写，还是会出问题。

3. 独立锁文件（简单场景适用） #

写进程开始写入时创建锁文件，写完后删除；读进程检查锁文件是否存在，存在则等待或跳过。关键用os.O_EXCL原子创建锁文件，避免多个写进程同时创建锁。

写进程代码：

import os

lock_path = 'cache.lock'
target_path = 'cache.dat'

# 原子创建锁文件，避免竞态
try:
    fd = os.open(lock_path, os.O_CREAT | os.O_EXCL | os.O_RDWR)
    os.close(fd)
except FileExistsError:
    # 已有进程在写，可重试或退出
    pass

try:
    with open(target_path, 'w') as f:
        f.write('你的缓存数据')
        f.flush()
        os.fsync(f.fileno())
finally:
    # 无论成功失败都清理锁文件
    if os.path.exists(lock_path):
        os.unlink(lock_path)

读进程代码：

import os
import time

lock_path = 'cache.lock'
target_path = 'cache.dat'

# 等待锁文件消失
while os.path.exists(lock_path):
    time.sleep(0.1)

if os.path.exists(target_path):
    with open(target_path, 'r') as f:
        data = f.read()

总结 #

直接判断文件是否在写入状态不是靠谱的解决方案，优先用原子重命名，这是最可靠且高效的方式；如果需要多进程同时读写（比如追加写入），用fcntl锁；锁文件适合简单场景，但要注意异常情况下锁的清理。

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