Python多进程池：如何为低优先级任务分配子池（m<n）？

针对你的CPU密集型异步任务场景（20物理核心，task1可占满全部资源，task2最多用10个核心），以下是最优实现方案及对应设计模式：

一、推荐实现：全局进程池 + 信号量限流 #

复用单个全局进程池（大小=20），通过信号量精准限制task2的并发数，既保证全局总并发不超核心数，又能让task1在需要时占满全部资源。

代码示例 #

import multiprocessing
from multiprocessing import Semaphore

# 全局进程池，匹配物理核心数
global_pool = multiprocessing.get_context("fork").Pool(20)
# 限制task2最多同时执行10个任务
task2_concurrency_limit = Semaphore(10)

def cpu_bound_task1(data):
    # task1的CPU密集型逻辑示例
    return sum(i*i for i in range(data))

def cpu_bound_task2(data):
    # 信号量控制：达到10个并发时，新任务会阻塞等待
    with task2_concurrency_limit:
        # task2的CPU密集型逻辑示例
        return sum(i*3 for i in range(data))

def submit_task1_batch(task_list):
    # 异步提交task1，无并发限制（会用满全局池）
    for data in task_list:
        global_pool.apply_async(
            cpu_bound_task1,
            args=(data,),
            callback=lambda res: print(f"Task1 result: {res}")
        )

def submit_task2_batch(task_list):
    # 异步提交task2，受信号量限制最多10个并发
    for data in task_list:
        global_pool.apply_async(
            cpu_bound_task2,
            args=(data,),
            callback=lambda res: print(f"Task2 result: {res}")
        )

if __name__ == "__main__":
    # 模拟大量异步任务
    task1_data = list(range(500, 1500))
    task2_data = list(range(100, 1100))
    
    submit_task1_batch(task1_data)
    submit_task2_batch(task2_data)
    
    # 等待所有任务完成
    global_pool.close()
    global_pool.join()

设计模式说明 #

• 资源池模式（Resource Pool Pattern）：通过全局进程池统一管理20个worker资源，避免频繁创建销毁进程的开销。
• 限流模式（Rate Limiting Pattern）：利用Semaphore对低优先级的task2做并发数限制，确保其资源占用不超过总核心数的一半，同时不影响task1的资源使用。

二、备选方案：双进程池 + 全局资源锁 #

若需要更严格的任务隔离，可创建两个进程池，但需通过全局信号量限制总并发数（避免超过20核心）：

• task1池大小设为20，task2池大小设为10
• 全局Semaphore(20)限制两个池的总并发执行数

但此方案会增加代码复杂度，且资源利用率略低于第一种方案，因此仅推荐在任务逻辑需要完全隔离时使用。

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