解决WSGI多进程下Flask应用写入同一CSV文件的数据丢失问题

针对mod_wsgi多进程导致的同一CSV文件被频繁覆盖、数据丢失问题，以下是几个可行的解决办法：

1. 原子文件写入（最简便的跨平台方案） #

核心思路是先将内容写入临时文件，确认写入完成后，再通过原子重命名操作替换目标CSV文件。文件系统的重命名操作在大多数操作系统中是原子性的，不会出现多个进程同时覆盖导致的文件损坏或数据丢失，且能保证下游读到的要么是完整的旧文件，要么是完整的新文件。

修改后的代码示例：

import os
import tempfile
from uuid import uuid4

def write_csv_files(self, destination: str) -> None:
    for filename in self.dataDict:
        filepath = os.path.join(destination, f"{filename}.csv")
        # 在目标目录创建临时文件，保证重命名的原子性
        temp_file = tempfile.NamedTemporaryFile(
            mode='w', 
            delete=False, 
            dir=destination,
            prefix=f"{filename}_tmp_{uuid4().hex}_",
            suffix=".csv"
        )
        try:
            # 写入内容到临时文件并强制刷入磁盘
            temp_file.write(gen_row_helper(self.dataDict[filename]))
            temp_file.flush()
            os.fsync(temp_file.fileno())
        finally:
            temp_file.close()
        
        # 原子重命名替换目标文件，失败则清理临时文件
        try:
            os.replace(temp_file.name, filepath)
        except OSError:
            os.remove(temp_file.name)
            raise

2. 文件锁机制（控制并发写入） #

通过给目标文件加排他锁，确保同一时间只有一个进程能执行写入操作，其他进程需等待锁释放后再尝试。可以使用跨平台的portalocker库实现：

先安装依赖：

pip install portalocker

修改后的代码：

import os
import portalocker

def write_csv_files(self, destination: str) -> None:
    for filename in self.dataDict:
        filepath = os.path.join(destination, f"{filename}.csv")
        # 打开文件并加排他锁，写入完成后自动释放
        with open(filepath, "w") as fobj:
            try:
                portalocker.lock(fobj, portalocker.LOCK_EX)
                fobj.write(gen_row_helper(self.dataDict[filename]))
                fobj.flush()
                os.fsync(fobj.fileno())
            finally:
                portalocker.unlock(fobj)

该方案会让多进程排队等待锁，有一定性能损耗，但能保证每个进程的写入都不会被覆盖，适合对数据完整性要求极高的场景。

3. 剥离写入任务到独立进程（彻底避免竞争） #

把CSV定时写入任务从Flask应用进程中分离出来，单独用一个定时进程负责写入操作，Flask进程仅负责收集数据并存入中间存储（如Redis、数据库）。

示例架构：

1. Flask端数据收集逻辑：

import redis
r = redis.Redis(host='localhost', port=6379, db=0)

def update_csv_data():
    # 每2分钟触发的数据更新逻辑
    data_dict = get_new_data()
    # 将CSV内容存入Redis
    r.hmset("csv_data", {k: gen_row_helper(v) for k, v in data_dict.items()})

2. 独立定时写入进程：

from apscheduler.schedulers.blocking import BlockingScheduler
import os
import redis

r = redis.Redis(host='localhost', port=6379, db=0)
destination = "/path/to/csv/directory"

def write_csv_files():
    csv_data = r.hgetall("csv_data")
    for filename, content in csv_data.items():
        filename = filename.decode('utf-8')
        content = content.decode('utf-8')
        filepath = os.path.join(destination, f"{filename}.csv")
        with open(filepath, "w") as fobj:
            fobj.write(content)

if __name__ == "__main__":
    scheduler = BlockingScheduler()
    scheduler.add_job(write_csv_files, 'interval', minutes=2)
    scheduler.start()

这种方案从根源上消除了多进程竞争，架构更清晰，Flask进程可专注处理Web请求。

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