ProcessPoolExecutor内嵌ThreadPoolExecutor报错及解决方案咨询

错误原因解析 #

• RuntimeError: There is no current event loop in thread：磁盘读写属于同步IO任务，不需要绑定asyncio事件循环，你强行添加事件循环的操作完全多余，反而导致线程中无可用事件循环报错。
• BrokenProcessPool：重写_process_worker时破坏了原进程池的通信逻辑（比如未正确处理任务/结果队列、子进程异常未捕获导致崩溃），引发进程池通信中断。

正确的_process_worker修改方案 #

核心思路是在子进程内初始化ThreadPoolExecutor，让磁盘读写任务通过线程池并发执行，同时完全保留原进程池的通信机制，避免破坏原有流程。

代码实现 #

import concurrent.futures.process
from concurrent.futures import ThreadPoolExecutor
import multiprocessing as mp

def custom_process_worker(call_queue, result_queue, initializer=None, initargs=()):
    # 根据磁盘IO并发需求设置线程数，建议4-8（磁盘寻址开销限制了过高并发的收益）
    thread_pool = ThreadPoolExecutor(max_workers=6)
    
    # 保留原进程初始化逻辑
    if initializer is not None:
        try:
            initializer(*initargs)
        except BaseException as e:
            result_queue.put((None, None, None, e))
            return
    
    try:
        while True:
            call_item = call_queue.get(block=True)
            if call_item is None:
                # 收到进程池退出信号，关闭线程池后退出子进程
                thread_pool.shutdown(wait=True)
                return
            
            # 解析原进程池传递的任务参数
            call_id, fn, args, kwargs = call_item
            
            # 包装任务，捕获异常避免单个任务崩溃导致子进程退出
            def task_wrapper():
                try:
                    result = fn(*args, **kwargs)
                    return (call_id, result, None, None)
                except BaseException as e:
                    return (call_id, None, e, None)
            
            # 提交任务到子进程线程池，将结果放回进程池结果队列
            future = thread_pool.submit(task_wrapper)
            result_queue.put(future.result())
    except BaseException as e:
        # 捕获子进程全局异常，避免进程池通信中断
        result_queue.put((None, None, e, None))
        thread_pool.shutdown(wait=False)

# 替换默认的_process_worker函数
concurrent.futures.process._process_worker = custom_process_worker

关键注意点 #

1. 线程数合理设置：磁盘IO的并发能力受限于磁盘硬件，过高的线程数会增加磁盘寻址开销，反而降低性能，建议设置为4-8。
2. 保留原通信逻辑：严格遵循原_process_worker的任务队列、结果队列处理流程，包括退出信号（收到None时关闭线程池并退出）。
3. 异常捕获：通过task_wrapper捕获单个任务的异常，避免任务崩溃导致子进程退出，引发BrokenProcessPool。
4. 移除asyncio依赖：同步磁盘读写任务不需要asyncio事件循环，直接删除相关代码即可解决事件循环报错。
5. 可序列化要求：任务的参数和返回值必须是可pickle的，否则会导致进程间通信失败。

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