如何处理基于 wsgi 的服务器启动多个进程写入同一文件导致数据丢失？

Question

我有一个 Flask Web 应用程序，它每两分钟定期写入多个带标题的 CSV 文件，当我使用 Flask 开发服务器运行它时，效果很好。但是，我最近使用 mod-wsgi 生产服务器部署了我的应用程序，并且写入 CSV 文件的代码未按预期工作，因为 mod-wsgi 为我的应用程序启动了多个进程，并且每个进程都尝试写入同一个 CSV 文件它会覆盖之前进程在两分钟间隔内写入的任何数据。

有问题的伪代码是：

# this is called every 2 mins because self.dataDict will have new data for every file every 2mins
def write_csv_files(self, destination: str) -> None:
    for filename in self.dataDict:
        filepath = os.path.join(destination + filename + ".csv")
        with open(filepath, "w") as fobj:
            # gen_row_helper returns a string of comma separated values for a file
            fobj.write(gen_row_helper(self.dataDict[filename]))

单个 CSV 文件示例

1694897668
name,place,number
Kevin, London, 123
Bob, Italy, 987

当有多个进程正在写入带有标头的同一个 csv 文件时如何处理这种情况？

理想的情况是仅在 2 分钟后有新数据可用且下游服务消耗了该数据后才覆盖数据，但现在多个进程最终在两分钟间隔内覆盖数据，从而导致数据丢失。当 Flask 开发服务器中只有一个进程时，上面的代码工作得很好，但这不能在生产中使用。

我已经尝试过的事情：

1. 以追加模式（“a”）打开文件，以便进程可以追加而不是覆盖，但这会导致最终 csv 文件中出现多个重复的标题行，并且文件会不断增长，因为即使在 2 分钟间隔后也不会被覆盖
2. 为每个进程创建一个文件，但这不起作用，因为使用这些 csv 文件的下游服务只需要每个文件名一个文件

Answer 1

您尝试过使用

multiprocessing.Lock

吗？我曾经编写过一个 Flask 应用程序，它必须做类似的事情，锁救了我。 （我使用的是 waitress，所以锁是

threading.Lock

。）只需在需要写入文件时获取锁即可；工作完成后，释放它（也可以使用上下文管理器样式）。这样，当其他进程尝试访问该文件时，它们将必须等待当前作业完成。