在asyncio循环中用ProcessPoolExecutor执行CPU密集型操作是否影响性能？

我尝试使用concurrent.futures.ProcessPoolExecutor配合asyncio.get_running_loop()，以非阻塞方式运行CPU密集型操作。我写了一个底层基于Polars实现CPU密集型逻辑的process_zip_file函数，还有一个异步函数process_zip_file_concurrent，通过asyncio循环在进程池中运行这个函数。请问这样的实现会导致性能不佳，还是函数会在独立进程中正常运行？

原代码 #

def process_zip_file(file: ZipFile):
    try:
        csv_files = unzip_file(file) 
    except Exception as e:
        raise ZipfileProcessingError("Error unzipping file.")

    try:
        dataframes = create_dataframes(csv_files) # polars operations under the hood

        concat_df = concat_dataframes(dataframes, file.filename)
        return concat_df
    except Exception as e:
        module_logging.error(e)
        raise DataframeProcessingErorr("Error processing csv files.")

async def process_zip_file_concurrent(zip_file: ZipFile):
    loop = asyncio.get_running_loop()
    with ProcessPoolExecutor(max_workers=4) as excecutor:
        result = await loop.run_in_executor(excecutor, process_zip_file, zip_file)    
    return result

核心结论 #

这个实现可以让CPU密集型任务在独立进程中正常运行，不会阻塞asyncio事件循环，但存在几个细节问题可能影响性能，需要针对性优化。

1. 正确性验证 #

• loop.run_in_executor会将process_zip_file提交到ProcessPoolExecutor的子进程中执行，完全脱离asyncio的事件循环线程，因此CPU密集型任务不会阻塞异步逻辑，这部分是正常的。
• 注意事项：ZipFile对象需要能被安全序列化（进程间通信依赖pickle传递参数）。如果ZipFile是基于磁盘路径打开的，通常可以正常序列化；但如果是内存中的对象，可能会出现序列化失败的情况。此时建议改为传递文件路径，在子进程中重新打开ZipFile。

2. 性能优化点 #

• 避免重复创建进程池：原代码每次调用process_zip_file_concurrent都新建ProcessPoolExecutor，进程的启动和销毁会带来额外开销。如果这个异步函数被频繁调用，建议复用一个全局的进程池实例。
• 协调Polars多线程与进程池：Polars默认会使用多线程执行操作，若ProcessPoolExecutor的max_workers设置过高（比如4），再加上Polars内部的多线程，会导致CPU过度饱和，反而降低性能。建议根据CPU核心数调整：比如8核CPU，将ProcessPoolExecutor的max_workers设为2-3，同时通过pl.Config.set_thread_pool_size(1)限制Polars在子进程中使用单线程，避免资源竞争。
• 减少进程间数据传输开销：Polars DataFrame的序列化（pickle）成本较高，如果DataFrame体积大，进程间传递会占用大量时间和内存。建议在子进程中完成更多后续处理（比如写入磁盘），只返回处理状态或小结果，减少数据传输量。

优化后的示例代码 #

from concurrent.futures import ProcessPoolExecutor
import asyncio
import polars as pl
from zipfile import ZipFile

# 全局复用进程池，根据CPU核心数调整max_workers
GLOBAL_PROCESS_POOL = ProcessPoolExecutor(max_workers=2)

# 限制Polars在子进程中使用单线程，避免与进程池竞争CPU
pl.Config.set_thread_pool_size(1)

def process_zip_file(file_path: str):
    try:
        with ZipFile(file_path, 'r') as file:
            csv_files = unzip_file(file) 
    except Exception as e:
        raise ZipfileProcessingError("Error unzipping file.")

    try:
        dataframes = create_dataframes(csv_files)
        concat_df = concat_dataframes(dataframes, file_path)
        # 在子进程中直接写入结果文件，减少返回数据量
        concat_df.write_parquet(f"{file_path}_result.parquet")
        return f"Processed {file_path} successfully"
    except Exception as e:
        module_logging.error(e)
        raise DataframeProcessingErorr("Error processing csv files.")

async def process_zip_file_concurrent(file_path: str):
    loop = asyncio.get_running_loop()
    result = await loop.run_in_executor(GLOBAL_PROCESS_POOL, process_zip_file, file_path)    
    return result

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