Python并行化：ProcessPoolExecutor+AsyncIO实现遇阻求助

完全可行，这种多进程（利用多核）+ 进程内AsyncIO（处理IO密集型校验）的混合架构，正好适配你当前CPU利用率低但请求量压垮单线程/多线程的场景。下面针对你的问题逐一解决：

核心问题分析 #

1. submit()执行main.py：Windows系统下多进程无fork机制，会重新导入主模块，若主模块在if __name__ == "__main__"外有执行代码，就会被重复触发。
2. 循环导入：校验逻辑和主程序耦合在同一模块导致，抽离逻辑到独立模块即可解决。
3. 隔离进程的AsyncIO启动：每个进程需要单独启动事件循环，执行带TaskGroup和超时的异步校验逻辑。

分步实现方案 #

1. 抽离异步校验逻辑到独立模块 #

创建worker_logic.py，把所有异步校验、runner逻辑放在这里，避免和主程序循环依赖：

import asyncio

# 你的API校验函数，替换成实际逻辑
async def validate_api_request(request_data):
    # 模拟API调用/校验的IO操作
    await asyncio.sleep(0.1)
    return {"status": "success", "request_id": request_data["id"]}

async def process_event_runner(requests_batch, timeout=3):
    results = []
    try:
        async with asyncio.TaskGroup() as tg:
            # 批量创建异步校验任务
            tasks = [tg.create_task(validate_api_request(req)) for req in requests_batch]
            # 等待任务完成或触发超时
            done, pending = await asyncio.wait(tasks, timeout=timeout)
            
            # 处理超时任务
            for task in pending:
                task.cancel()
                results.append({"status": "timeout", "request_id": None})
            
            # 处理完成/失败任务
            for task in done:
                try:
                    results.append(task.result())
                except Exception as e:
                    results.append({"status": "error", "error": str(e), "request_id": None})
    except asyncio.TimeoutError:
        results.append({"status": "batch_timeout", "details": "Whole batch exceeded timeout"})
    return results

# 进程入口函数：负责在子进程中启动AsyncIO循环
def worker_entry(batch_data):
    return asyncio.run(process_event_runner(batch_data))

2. 主程序配置ProcessPoolExecutor #

修改main.py，把核心逻辑放在if __name__ == "__main__"内，避免多进程重复执行主程序代码：

from concurrent.futures import ProcessPoolExecutor
import os
from worker_logic import worker_entry

def main():
    # 模拟待处理的API请求队列
    api_requests = [{"id": i, "payload": f"request_{i}"} for i in range(100)]
    # 按批次打包请求，减少进程间通信开销
    batch_size = 10
    request_batches = [api_requests[i:i+batch_size] for i in range(0, len(api_requests), batch_size)]

    # 配置进程池：max_workers设为CPU核心数（或1.5倍），拉满CPU利用率
    # max_tasks_per_child：每个进程处理N个批次后重启，防止内存泄漏
    with ProcessPoolExecutor(
        max_workers=os.cpu_count(),
        max_tasks_per_child=10
    ) as executor:
        # 提交所有批次任务到进程池
        futures = [executor.submit(worker_entry, batch) for batch in request_batches]
        
        # 收集并处理结果
        for idx, future in enumerate(futures):
            try:
                batch_result = future.result()
                print(f"Batch {idx+1} result: {batch_result}")
            except Exception as e:
                print(f"Batch {idx+1} failed: {str(e)}")

if __name__ == "__main__":
    # 仅主进程执行此逻辑，避免Windows多进程重复启动
    main()

关键细节说明 #

• CPU利用率优化：max_workers设为os.cpu_count()（或CPU核心数的1-1.5倍），能让CPU利用率稳定在70-90%区间，避免进程过多导致调度开销。
• 进程隔离与AsyncIO启动：每个子进程通过worker_entry函数启动独立的AsyncIO事件循环，完全隔离主进程的循环，满足你对隔离进程的要求。
• 超时与TaskGroup：process_event_runner中用asyncio.wait结合timeout实现全局超时，同时用TaskGroup管理异步任务的生命周期，确保任务正确取消和清理。
• 避免循环导入：把校验逻辑抽离到worker_logic.py，主程序只导入worker_entry函数，彻底解决循环依赖问题。
• 解决submit()执行main.py：所有主程序逻辑都放在if __name__ == "__main__"内，Windows下子进程导入主模块时不会执行这些代码，只会导入函数定义。

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