Python非阻塞写入CSV文件：如何优化计算与写入循环？

我正在编写Python代码执行计算并将结果写入文件，当前代码如下：

for name, group in data.groupby('Date'):
    df = lot_of_numpy_calculations(group)
    with open('result.csv', 'a') as f:
        df.to_csv(f, header=False, index=False)

计算与写入操作均耗时，我查阅过Python异步相关文章，但不知如何实现。是否有简便方法优化该循环，使其无需等待写入完成即可启动下一次迭代？

首先，你提到的场景里，计算是CPU密集型（numpy运算），写入是IO密集型，直接用asyncio可能不是最优解——因为asyncio对CPU密集型任务的优化有限，反而用多进程+多线程的组合或者线程池会更简便。下面给你几个实用的方案：

方案1：用concurrent.futures.ThreadPoolExecutor异步处理写入 #

因为文件写入是IO操作，适合用线程来异步执行，这样计算任务可以不用等写入完成就继续下一轮：

from concurrent.futures import ThreadPoolExecutor
import pandas as pd

# 先定义写入函数
def write_to_csv(df):
    with open('result.csv', 'a') as f:
        df.to_csv(f, header=False, index=False)

# 初始化线程池，根据你的IO并发需求设置线程数，比如4
with ThreadPoolExecutor(max_workers=4) as executor:
    for name, group in data.groupby('Date'):
        df = lot_of_numpy_calculations(group)
        # 提交写入任务到线程池，不用等待完成
        executor.submit(write_to_csv, df)

这个方案的好处是改动极小，不需要大改现有代码，线程池会帮你管理异步写入的任务，主线程可以继续专注于计算。

方案2：计算与写入分离，用队列解耦 #

如果计算也想并行（因为numpy计算是CPU密集，单进程的话只能用一个核），可以用多进程处理计算，线程处理写入，通过队列传递结果：

from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor
import queue
import threading

# 定义写入线程的工作函数
def writer_worker(q):
    while True:
        df = q.get()
        if df is None:  # 结束信号
            break
        with open('result.csv', 'a') as f:
            df.to_csv(f, header=False, index=False)
        q.task_done()

# 创建队列
q = queue.Queue(maxsize=10)  # 设置队列大小，防止内存溢出
# 启动写入线程
writer_thread = threading.Thread(target=writer_worker, args=(q,))
writer_thread.start()

# 用进程池处理计算（CPU密集型适合多进程）
with ProcessPoolExecutor() as executor:
    # 提交所有计算任务
    futures = [executor.submit(lot_of_numpy_calculations, group) for _, group in data.groupby('Date')]
    # 遍历结果，放入队列
    for future in futures:
        df = future.result()
        q.put(df)

# 发送结束信号，等待写入线程完成
q.put(None)
writer_thread.join()

这个方案适合计算任务非常耗时的场景，多进程可以利用多核CPU加速计算，同时写入线程异步处理IO，最大化资源利用率。

关键注意事项 #

• 如果你的lot_of_numpy_calculations已经依赖numpy的底层并行优化（比如MKL/OpenBLAS），那多进程可能不会有明显提升，甚至因为进程间开销变慢，这时候方案1就足够了。
• 异步写入时，Python的open('a')模式在大部分操作系统下会自动保证追加操作的原子性，不用担心多个线程同时写入导致内容混乱；但如果是复杂的写入逻辑，建议手动加锁（比如用threading.Lock）。
• 队列的maxsize要根据你的内存情况合理设置，避免计算出的DataFrame过多堆积，占满内存。

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