如何在Python中防止多进程并发写入同一文件发生冲突？-Python教程-PHP中文网

多进程直接用 open(..., 'a') 写文件会因竞态导致内容覆盖或交错，根本原因是 write() 非原子且 lseek+write 不受 o_append 保护；推荐用 multiprocessing.queue 由单写进程统一落盘，辅以 flush() 和合理批处理。

如何在python中防止多进程并发写入同一文件发生冲突？

为什么直接用 `open(..., 'a')` 多进程写文件会出问题

因为 open 的底层 write() 系统调用不是原子的，尤其在追加模式下：多个进程同时调用 lseek() 到文件末尾 + write()，可能产生竞态——两个进程读到同一偏移，后写的覆盖前写的，或内容交错（比如 “log1\nlog2\n” 变成 “lolog1\ng2\n”）。这不是 Python 层面能靠锁规避的，是操作系统级问题。

即使每个进程都用 'a' 模式，Linux 的 O_APPEND 保证单次 write() 原子性，但仅限于单次系统调用；如果写入内容跨多个 write()（比如大字符串被内核分片），仍可能穿插
print(..., file=f) 或 f.write(s); f.flush() 不改变根本风险
多线程用 threading.Lock 无效——进程间内存不共享，锁不互通

推荐方案：用 `multiprocessing.Manager().dict()` + 单写进程中转

核心思路是避免所有工作进程直接碰文件，改由一个专用写进程统一接收日志/数据并顺序落盘。其他进程只往共享队列或字典发消息，解耦 I/O 和计算。

用 multiprocessing.Queue 最稳妥：生产者（工作进程）put()，消费者（写进程）get()，天然线程/进程安全
若需带结构化元数据（如时间戳、进程 ID），把日志封装成字典：queue.put({'pid': os.getpid(), 'msg': 'done', 'ts': time.time()})
写进程循环 queue.get(timeout=0.1) 防止卡死，空队列时做 time.sleep(0.01) 降频
注意 Queue 有默认大小（约 32767），高吞吐场景要设 maxsize=0（无界）或捕获 queue.Full

# 写进程示例
def writer_process(log_queue, filepath):
    with open(filepath, 'a') as f:
        while True:
            try:
                item = log_queue.get(timeout=0.1)
                if item == 'STOP':  # 退出信号
                    break
                f.write(f"[{item['ts']:.3f}] P{item['pid']}: {item['msg']}\n")
                f.flush()  # 确保立即写入磁盘
            except queue.Empty:
                continue

替代方案：用文件锁 `flock()`（仅 Linux/macOS）

如果必须让每个进程自己写，且能接受 POSIX 系统限制，可用 flock() 对文件描述符加建议锁。注意这不是强制锁，所有进程必须主动参与协作。

Claw Code Suite (Python-Only Edition)

仅使用 Python 将 Claw Code 虚拟化框架与 OpenClaw 集成，提供 184 种工具和 200 多条命令，用于安全分析和代码质量...

下载

Python 通过 fcntl.flock(fd, fcntl.LOCK_EX) 加锁，LOCK_UN 解锁，锁随文件描述符关闭自动释放
必须在 open() 后立刻 flock()，不能先写再锁——已写内容不受保护
Windows 不支持 flock()，要用 msvcrt.locking()，但行为不一致，不推荐跨平台使用
锁粒度是整个文件，高并发时会成为瓶颈，写入延迟明显上升

# 工作进程中安全写片段
import fcntl
with open('log.txt', 'a') as f:
    fcntl.flock(f.fileno(), fcntl.LOCK_EX)
    try:
        f.write(f"Process {os.getpid()} done\n")
        f.flush()
    finally:
        fcntl.flock(f.fileno(), fcntl.LOCK_UN)  # 必须确保释放