Python中如何快速循环读取文件并避免与写入程序的冲突？

核心思路 #

结合你的需求（读操作高频低CPU、写操作低频允许短暂中断），最简洁高效的方案是原子文件替换+缓存读取，完全不需要复杂的锁、线程或队列机制：

• 写操作（程序B）：每次更新时，先将内容写入临时文件（比如F.tmp），完成后用os.rename()替换原文件F。由于rename在主流文件系统（Unix/Linux、Windows NTFS）中是原子操作，程序A永远只会读到完整的旧文件或新文件，不会出现半写的残缺数据。
• 读操作（程序A）：维护一份文件内容的缓存，同时记录文件的最后修改时间。每次循环时：
  1. 快速检查文件的修改时间（os.path.getmtime(F)）
  2. 如果修改时间未变化，直接返回缓存内容，无需实际读取文件
  3. 如果修改时间更新，再读取文件内容，更新缓存和修改时间

为什么不选其他方案？ #

• 锁：不管是线程锁还是文件锁，都会增加额外的开销，尤其是读操作高频的场景下，锁的获取释放会浪费CPU资源。而原子替换的方式完全避免了锁的需求，实现更简单。
• 中断：信号中断是用来处理外部事件的，无法解决文件读写的同步问题，完全不适用这个场景。
• 线程：线程是执行单元，不是同步机制。如果A和B是独立程序，线程根本没用；如果是同一进程内的两个线程，也不如原子替换+缓存的方案高效。
• 队列：队列是用来在线程/进程间传递数据的，你的需求是直接读写文件，用队列会完全改变架构，不符合原需求。

代码示例 #

程序A（读取逻辑）

import os
import time

def read_file_with_cache(file_path):
    cache = None
    last_mtime = 0
    while True:
        current_mtime = os.path.getmtime(file_path)
        if current_mtime != last_mtime:
            with open(file_path, 'r') as f:
                cache = f.read()
            last_mtime = current_mtime
        # 这里处理缓存内容，根据实际需求替换逻辑
        print(f"读取内容: {cache}")
        time.sleep(0.1)  # 控制每秒10次循环

if __name__ == "__main__":
    read_file_with_cache("F")

程序B（写入逻辑）

import os
import time

def write_file_atomically(file_path, content):
    tmp_path = f"{file_path}.tmp"
    with open(tmp_path, 'w') as f:
        f.write(content)
    # 原子替换原文件，保证读操作不会读到不完整数据
    os.rename(tmp_path, file_path)

if __name__ == "__main__":
    while True:
        # 生成更新内容，根据实际需求替换逻辑
        update_content = f"更新时间: {time.strftime('%Y-%m-%d %H:%M:%S')}"
        write_file_atomically("F", update_content)
        time.sleep(30 * 60)  # 每30分钟执行一次写入

优势 #

• 读操作几乎无CPU开销：大部分循环只是检查文件修改时间，只有当文件更新时才会执行一次IO读取
• 完全避免读写冲突：原子替换保证读操作永远拿到完整数据
• 实现简单：不需要处理复杂的锁、线程同步逻辑，代码可读性高
• 符合需求：写操作无需追求时效性，每30分钟一次的替换对读操作的影响可以忽略（甚至感觉不到中断）

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