如何在Flask中使用多进程？请求级创建Process是否合适？

首先明确：为每个请求创建独立Process不是最优解——进程的创建、销毁存在系统开销，高请求量下会浪费CPU调度资源，反而拉低整体性能。下面是几种更合理的实现方式：

1. 推荐：用multiprocessing.Pool进程池复用进程 #

进程池会预先创建固定数量的进程，复用这些进程处理后续请求，避免重复创建销毁的开销。代码实现简单，且不需要手动管理队列：

from flask import Flask, request
from multiprocessing import Pool

app = Flask(__name__)
# 根据CPU核心数或任务数量设置进程数，这里3个进程对应3个独立任务
pool = Pool(processes=3)

# 模拟子任务（保持你原有的业务逻辑即可）
def do_task_a(data):
    return {"result_a": "a"}

def do_task_b(data):
    return {"result_b": "b"}

def do_task_c(data):
    return {"result_c": "c"}

@app.route("/foo", methods=["POST"])
def foo():
    data = request.get_json()
    
    # 异步提交任务到进程池
    task_a = pool.apply_async(do_task_a, args=(data,))
    task_b = pool.apply_async(do_task_b, args=(data,))
    task_c = pool.apply_async(do_task_c, args=(data,))
    
    # 阻塞等待所有任务完成并获取结果
    result_a = task_a.get()
    result_b = task_b.get()
    result_c = task_c.get()
    
    return {
        "result_a": result_a["result_a"],
        "result_b": result_b["result_b"],
        "result_c": result_c["result_c"]
    }

if __name__ == "__main__":
    # 建议开启threaded模式，允许同时处理多个请求
    app.run(host='0.0.0.0', port=4000, threaded=True)

2. 手动用Queue实现多进程（不推荐，仅作参考） #

如果一定要手动管理进程和队列，可参考以下实现，但缺点是每个请求都创建新进程，开销大：

from flask import Flask, request
from multiprocessing import Process, Queue

app = Flask(__name__)

# 通用worker函数，负责执行任务并把结果放入队列
def worker(task_func, data, queue):
    result = task_func(data)
    queue.put(result)

# 子任务函数同前...

@app.route("/foo", methods=["POST"])
def foo():
    data = request.get_json()
    queue = Queue()
    
    # 为每个任务创建进程
    p1 = Process(target=worker, args=(do_task_a, data, queue))
    p2 = Process(target=worker, args=(do_task_b, data, queue))
    p3 = Process(target=worker, args=(do_task_c, data, queue))
    
    p1.start()
    p2.start()
    p3.start()
    
    # 等待所有进程完成
    p1.join()
    p2.join()
    p3.join()
    
    # 从队列取出结果（注意：队列取出顺序不一定等于任务提交顺序）
    final_result = {}
    while not queue.empty():
        res = queue.get()
        final_result.update(res)
    
    return final_result

3. 进阶：异步Flask + ProcessPoolExecutor #

如果用Flask 2.0+的异步视图，结合concurrent.futures.ProcessPoolExecutor可以更好地处理高并发请求：

from flask import Flask, request
from concurrent.futures import ProcessPoolExecutor

app = Flask(__name__)
executor = ProcessPoolExecutor(max_workers=3)

# 子任务函数同前...

@app.route("/foo", methods=["POST"])
async def foo():
    data = request.get_json()
    
    # 提交任务到进程池
    future_a = executor.submit(do_task_a, data)
    future_b = executor.submit(do_task_b, data)
    future_c = executor.submit(do_task_c, data)
    
    # 获取结果
    result_a = future_a.result()
    result_b = future_b.result()
    result_c = future_c.result()
    
    return {
        "result_a": result_a["result_a"],
        "result_b": result_b["result_b"],
        "result_c": result_c["result_c"]
    }

if __name__ == "__main__":
    app.run(host='0.0.0.0', port=4000, threaded=True)

部署注意事项 #

• 如果用Gunicorn等WSGI服务器部署，不要在Flask app初始化时创建进程池，否则Gunicorn的多worker会导致多个进程池实例，浪费资源。应在每个worker启动后创建进程池（比如用Gunicorn的post_fork钩子）。
• 务必关闭threaded=False，否则同一时间只能处理一个请求，即使进程池并行处理任务，也无法提升整体吞吐量。

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