多Python脚本操作单个TXT文件的冲突问题及解决方案咨询

可能出现的冲突问题 #

• 文件内容损坏：当一个脚本正在写入行时，另一个脚本修改或截断文件，会导致写入的内容乱码、不完整，或原有数据被破坏。
• 数据丢失：脚本A刚写入的行可能被脚本B的删除操作意外清除；或者脚本B读取到脚本A尚未写完的半行数据，导致处理逻辑出错。
• 竞态条件：两个脚本同时打开文件进行写操作时，最后执行关闭的脚本会覆盖之前的修改，导致预期外的结果（比如脚本A写入的新数据被脚本B的修改覆盖）。
• 资源占用异常：部分系统下，一个脚本打开文件后未释放句柄，另一个脚本会无法打开文件，抛出PermissionError或文件被占用的异常。

可靠的解决办法 #

1. 使用文件锁实现互斥访问 #

通过操作系统提供的文件锁机制，确保同一时间只有一个脚本能对文件进行写/修改操作。

• Unix/Linux 环境：用fcntl模块加排他锁

import fcntl

def write_data(file_path, data):
    with open(file_path, 'a') as f:
        # 加排他锁
        fcntl.flock(f, fcntl.LOCK_EX)
        f.write(data + '\n')
        f.flush()
        # 锁会随文件关闭自动释放

def process_data(file_path):
    with open(file_path, 'r+') as f:
        fcntl.flock(f, fcntl.LOCK_EX)
        # 读取所有行并处理
        lines = f.readlines()
        # 示例：移除第一行已处理数据
        processed_lines = lines[1:] if lines else []
        # 回到文件开头，写入处理后的内容
        f.seek(0)
        f.writelines(processed_lines)
        f.truncate()

• Windows 环境：用msvcrt模块的文件锁

import msvcrt

def lock_file(f):
    msvcrt.locking(f.fileno(), msvcrt.LK_LOCK, 1)

def unlock_file(f):
    msvcrt.locking(f.fileno(), msvcrt.LK_UNLCK, 1)

# 写入/处理函数逻辑类似，操作前后加锁、解锁

2. 用消息队列替代直接文件操作 #

彻底避免文件并发问题，让脚本A把数据发送到队列，脚本B从队列取数据处理，处理结果可存入数据库或单独归档：

• 进程内队列（适用于同进程/多进程场景）：

# 脚本A（生产者）
from multiprocessing import Queue
import time

def producer(queue):
    while True:
        data = f"new_data_{time.time()}"
        queue.put(data)
        time.sleep(60)

# 脚本B（消费者）
def consumer(queue):
    while True:
        data = queue.get()
        # 处理数据逻辑
        print(f"Processing {data}")
        # 处理完无需修改原文件，直接丢弃或存入数据库

• 跨进程/机器场景可使用Redis、RabbitMQ等中间件，实现更可靠的队列通信。

3. 分段文件策略 #

脚本A每次写入新的临时文件（按时间戳命名），脚本B只处理已完成的文件，不触碰正在写入的文件：

• 脚本A：

import time

def write_data():
    timestamp = int(time.time())
    file_path = f"data_{timestamp}.txt"
    with open(file_path, 'w') as f:
        f.write(f"new_data_{timestamp}\n")
    time.sleep(60)

• 脚本B：遍历目录下的文件，筛选出非当前写入的文件，处理后归档或删除。

4. 替换为轻量数据库（如SQLite） #

SQLite内置事务和并发控制，比TXT文件更适合并发读写场景：

• 脚本A插入数据：

import sqlite3
import time

def insert_data():
    conn = sqlite3.connect('data.db')
    cursor = conn.cursor()
    # 初始化表（仅第一次执行）
    cursor.execute('CREATE TABLE IF NOT EXISTS tasks (id INTEGER PRIMARY KEY AUTOINCREMENT, data TEXT, processed INTEGER DEFAULT 0)')
    data = f"new_data_{time.time()}"
    cursor.execute('INSERT INTO tasks (data) VALUES (?)', (data,))
    conn.commit()
    conn.close()
    time.sleep(60)

• 脚本B处理并标记/删除数据：

def process_task():
    conn = sqlite3.connect('data.db')
    cursor = conn.cursor()
    # 取未处理的第一条数据
    cursor.execute('SELECT id, data FROM tasks WHERE processed = 0 LIMIT 1')
    task = cursor.fetchone()
    if task:
        task_id, data = task
        # 处理数据逻辑
        print(f"Processing {data}")
        # 标记为已处理或直接删除
        cursor.execute('UPDATE tasks SET processed = 1 WHERE id = ?', (task_id,))
        # 或删除：cursor.execute('DELETE FROM tasks WHERE id = ?', (task_id,))
        conn.commit()
    conn.close()

5. 原子替换文件操作 #

脚本B修改文件时，先写入临时文件，再用原子操作替换原文件，避免中间状态被脚本A读取：

import os
import tempfile

def process_data(file_path):
    # 读取原文件内容
    with open(file_path, 'r') as f:
        lines = f.readlines()
    # 处理数据（示例：移除第一行）
    processed_lines = lines[1:] if lines else []
    # 写入临时文件
    with tempfile.NamedTemporaryFile(mode='w', delete=False) as temp_f:
        temp_f.writelines(processed_lines)
    # 原子替换原文件，避免中间状态
    os.replace(temp_f.name, file_path)

脚本A用追加模式写入，并确保内容立即落地：

import os
import time

def write_data(file_path, data):
    with open(file_path, 'a') as f:
        f.write(data + '\n')
        f.flush()
        os.fsync(f.fileno())  # 确保内容写入磁盘，避免缓存延迟
    time.sleep(60)

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