Python从异步模型迁移到多进程模型的性能优化求助

一、针对异步模型的阻塞问题解决办法 #

1. 用线程池隔离同步阻塞代码 #

把任务中涉及同步阻塞的部分（比如第三方库的同步I/O、轻量CPU阻塞操作），用concurrent.futures.ThreadPoolExecutor包裹，通过事件循环的run_in_executor方法异步执行，避免阻塞整个事件循环：

import asyncio
from concurrent.futures import ThreadPoolExecutor

# 根据阻塞任务数量调整线程数
executor = ThreadPoolExecutor(max_workers=20)

async def async_task(task_data):
    # 异步非阻塞逻辑
    # ...
    # 同步阻塞部分交给线程池执行
    blocking_result = await asyncio.get_running_loop().run_in_executor(
        executor, blocking_function, task_data
    )
    # 后续异步处理逻辑
    # ...

线程池开销远低于进程池，适配I/O密集型阻塞场景，不会影响其他异步任务调度。

2. 给异步任务添加超时控制 #

用asyncio.wait_for为每个任务设置超时时间，防止任务无限挂起导致事件循环僵死：

async def run_task_with_timeout(task_data, timeout=300):  # 设为单任务最长执行时间
    try:
        await asyncio.wait_for(async_task(task_data), timeout=timeout)
    except asyncio.TimeoutError:
        # 处理超时：记录日志、标记任务失败或重试
        print(f"Task {task_data} timed out")

3. 重度阻塞任务单独进程隔离 #

对确认会长期阻塞甚至死锁的任务，单独用concurrent.futures.ProcessPoolExecutor执行，仅针对这类任务做进程隔离，其余任务仍用异步模型，平衡性能与可靠性：

from concurrent.futures import ProcessPoolExecutor

# 少量进程处理重度阻塞任务
process_executor = ProcessPoolExecutor(max_workers=5)

async def handle_heavy_task(task_data):
    loop = asyncio.get_running_loop()
    result = await loop.run_in_executor(process_executor, heavy_blocking_task, task_data)
    # 处理结果并放入对应队列
    # ...

二、多进程模型的性能优化方案 #

1. 替换multiprocessing.Manager队列为原生队列 #

Manager队列依赖中间进程转发数据，开销极大，改用multiprocessing.Queue（原生跨进程队列），直接通过内存共享传递数据，大幅降低通信开销：

import multiprocessing

def worker(task_queue, queue_a, queue_b, queue_c, queue_d):
    while True:
        task_data = task_queue.get()
        if task_data is None:  # 进程终止信号
            break
        # 执行任务并输出到对应队列
        # ...

if __name__ == "__main__":
    task_queue = multiprocessing.Queue(maxsize=100)
    queue_a = multiprocessing.Queue(maxsize=500)
    queue_b = multiprocessing.Queue(maxsize=500)
    queue_c = multiprocessing.Queue()
    queue_d = multiprocessing.Queue()

    # 根据并发限制启动对应数量的进程
    workers = [
        multiprocessing.Process(
            target=worker,
            args=(task_queue, queue_a, queue_b, queue_c, queue_d)
        ) for _ in range(10)
    ]
    for p in workers:
        p.start()

2. 减少序列化/反序列化开销 #

• 用更快的序列化库：替换默认pickle为msgpack（体积小、速度快），手动控制序列化逻辑：

  import msgpack
  
  # 子进程输出时序列化
  queue_a.put(msgpack.packb(output_a))
  
  # 主进程消费时反序列化
  output_a = msgpack.unpackb(queue_a.get())
  

• 批量输出大体积数据：A/B类输出积累到一定数量（比如100个）再批量放入队列，减少队列操作次数与总序列化开销：

  def worker(...):
      batch_a = []
      for item in task_output_a:
          batch_a.append(item)
          if len(batch_a) >= 100:
              queue_a.put(msgpack.packb(batch_a))
              batch_a = []
      if batch_a:
          queue_a.put(msgpack.packb(batch_a))
  

• 用共享内存传递大对象：对于5MB级的A/B输出，用Python3.8+的multiprocessing.SharedMemory直接共享内存，避免完整序列化：

  from multiprocessing import SharedMemory
  
  # 子进程发送大数据
  def send_large_data(queue, data):
      shm = SharedMemory(create=True, size=len(data))
      shm.buf[:len(data)] = data
      # 将共享内存名称和数据长度传入队列
      queue.put((shm.name, len(data)))
      return shm  # 保持引用直到主进程读取
  
  # 主进程读取共享内存
  def receive_large_data(queue):
      shm_name, data_len = queue.get()
      shm = SharedMemory(name=shm_name)
      data = bytes(shm.buf[:data_len])
      shm.close()
      shm.unlink()  # 释放内存
      return data
  

3. 复用进程池避免进程创建开销 #

用concurrent.futures.ProcessPoolExecutor代替手动创建Process，自动复用进程，避免任务数量多时频繁创建销毁进程的开销：

from concurrent.futures import ProcessPoolExecutor

def process_task(task_data):
    # 执行任务，返回四类输出的批量结果
    outputs_a, outputs_b, outputs_c, outputs_d = run_task_logic(task_data)
    return (outputs_a, outputs_b, outputs_c, outputs_d)

if __name__ == "__main__":
    # 匹配允许的最大并发数
    with ProcessPoolExecutor(max_workers=50) as executor:
        futures = [executor.submit(process_task, task_data) for task_data in all_tasks]
        # 处理结果并放入对应队列
        for future in futures:
            a, b, c, d = future.result()
            queue_a.put(a)
            queue_b.put(b)
            queue_c.put(c)
            queue_d.put(d)

4. 优化信号量控制逻辑 #

把A/B类输出的信号量控制放在子进程内部，减少跨进程信号量操作开销。比如子进程内部检查队列长度，超过阈值则暂停输出：

import time

def worker(task_queue, queue_a, queue_b, ...):
    while True:
        task_data = task_queue.get()
        # 生成任务输出
        for item_a in outputs_a:
            # 控制队列A长度，避免内存溢出
            while queue_a.qsize() >= 500:  # 阈值根据内存情况调整
                time.sleep(0.1)
            queue_a.put(item_a)
        # 同理处理outputs_b
        # ...

三、版本升级建议 #

升级到Python3.11，它对异步事件循环、多进程通信有显著性能优化：

• 异步任务调度效率提升
• multiprocessing模块底层优化
• 新增asyncio.TaskGroup简化任务管理
  这些优化能直接降低异步/多进程模型的运行开销。

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