FastAPI请求中多CPU核处理列表CPU密集任务的方法及multiprocessing模块可用性

当然可以用Python标准库的multiprocessing模块来解决这个问题——毕竟Python的GIL（全局解释器锁）限制了多线程在CPU密集场景下的并行能力，多进程是绕开GIL实现真正并行的核心方案。不过在FastAPI框架里用的时候，得注意不要重复创建进程池、配合FastAPI的生命周期管理，还要考虑生产环境的运行模式（比如uvicorn的多worker设置）。

核心实现方案：全局进程池+生命周期管理 #

最稳妥的方式是在FastAPI应用启动时初始化一个全局的进程池，所有请求复用这个池，避免每次请求创建进程的高昂开销。这里可以用FastAPI的lifespan上下文管理器来完成进程池的创建和销毁：

from fastapi import FastAPI
import multiprocessing
from contextlib import asynccontextmanager

# 定义你的CPU密集型任务（必须是可序列化的函数，不能依赖FastAPI请求对象等不可序列化的资源）
def cpu_intensive_task(item):
    # 示例：模拟耗时计算，替换成你的实际业务逻辑
    total = 0
    for i in range(10**7):
        total += i * item
    return total

# 用lifespan管理进程池的生命周期
@asynccontextmanager
async def lifespan(app: FastAPI):
    # 启动时创建进程池，进程数建议设为CPU核心数（或根据实际情况调整）
    app.state.pool = multiprocessing.Pool(processes=multiprocessing.cpu_count())
    yield  # 应用运行期间保持进程池存活
    # 关闭进程池：先停止接受新任务，再等待所有任务完成
    app.state.pool.close()
    app.state.pool.join()

# 初始化FastAPI应用，绑定lifespan
app = FastAPI(lifespan=lifespan)

@app.post("/process-list")
async def process_items(items: list[int]):
    # 用进程池的map方法并行处理列表中的每个元素
    results = app.state.pool.map(cpu_intensive_task, items)
    return {"processed_results": results}

这段代码的关键细节： #

• 全局进程池：通过app.state存储进程池，确保所有请求复用同一个池，避免重复创建进程的开销。
• 任务函数的可序列化性：cpu_intensive_task必须是能被pickle序列化的函数，不能引用FastAPI的Request对象、数据库连接等不可序列化的资源——如果需要外部资源，建议在任务函数内部初始化（比如每次进程启动时创建数据库连接）。
• 进程数设置：multiprocessing.cpu_count()会自动获取当前机器的CPU核心数，这是比较合理的默认值，也可以根据业务压力调整（比如设为核心数的1.5倍）。

生产环境的注意事项 #

1. 避免uvicorn多worker+进程池的叠加：
   如果用uvicorn的--workers参数启动多个FastAPI worker（比如uvicorn main:app --workers 4），每个worker都会创建自己的进程池，这会导致总进程数飙升（4个worker × 8核心 = 32个进程），反而会因为CPU上下文切换频繁降低性能。
   建议生产环境要么用单worker+进程池，要么如果要用多worker，就把每个worker的进程池大小设为1，或者直接用Celery这类分布式任务队列。

2. 进程间通信的开销：
   多进程之间的数据传递依赖pickle序列化，如果你的任务输入/输出数据量很大，序列化的开销会很明显。尽量精简传递给任务函数的数据，只传必要的参数。

3. 替代方案：用Celery做异步任务队列
   如果你的CPU密集型任务不需要实时返回结果（或者可以接受异步返回），更推荐用Celery配合Redis/RabbitMQ做任务队列：把任务丢给Celery worker进程处理，FastAPI只负责接收请求、提交任务，然后返回任务ID，客户端后续通过ID查询结果。这种方式更适合生产环境的高并发场景，也能更好地隔离FastAPI进程和计算进程。

为什么不能用多线程？ #

因为Python的GIL存在，多线程在CPU密集型任务中无法真正并行执行——同一时刻只有一个线程能执行Python字节码，多线程只能用来处理I/O密集型任务（比如网络请求、数据库查询），而CPU密集型任务必须用多进程绕开GIL。

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