Flask中使用joblib实现多CPU并行计算失败求助

这个问题我碰到过不少次，核心原因是Flask默认服务器的多线程模式和joblib多进程的嵌套冲突。Werkzeug（Flask默认的开发服务器）默认是单进程多线程运行的，当你在请求处理的线程里尝试启动joblib的多进程并行时，joblib会检测到这种“线程嵌套多进程”的不安全场景，自动将n_jobs降级为1（也就是串行执行），这就是你看到警告的原因。

下面给你几个可行的解决思路，按场景优先级排序：

1. 调整Flask开发服务器为多进程模式（仅用于开发验证） #

如果你只是在开发阶段验证并行功能，可以修改Flask的启动参数，关闭线程模式，改用多进程运行。这样每个请求会被分配到独立的进程中，joblib的多进程就可以正常工作了：

from flask import Flask
from joblib import Parallel, delayed
from time import sleep

def myfunc(x):
    sleep(5)
    return x

application = Flask(__name__)

@application.route('/', methods = ['GET'])
def getresult():
    # 这里n_jobs=-1可以正常使用了
    out = Parallel(n_jobs=-1, verbose=10)(delayed(myfunc)(i) for i in range(5))
    return str(sum(out))

if __name__ == "__main__":
    application.debug = True
    # 关闭线程，启用4个进程处理请求
    application.run(threaded=False, processes=4)

⚠️ 注意：这种方式只适合开发测试，Werkzeug不是生产级服务器，不要直接部署到EC2生产环境。

2. 使用生产级多进程服务器（推荐用于EC2部署） #

在AWS EC2上部署Flask应用，推荐用Gunicorn这类生产级服务器，它默认采用多进程模式，每个Worker是独立的操作系统进程，完全避免了线程嵌套的问题。

步骤： #

1. 安装Gunicorn：

pip install gunicorn

2. 用多进程模式启动Flask应用（假设你的代码文件名为app.py）：

# --workers 设置为EC2实例的CPU核心数（比如4核就设为4）
# --threads 1 确保每个Worker是单线程，彻底避免线程嵌套风险
gunicorn --workers 4 --threads 1 app:application

启动后，当请求进来时，Gunicorn会把请求分配给某个Worker进程，在这个进程里调用joblib的Parallel(n_jobs=-1)就会正常使用多CPU并行，不会再触发降级警告。

3. 用Celery做异步任务队列（适合耗时较长的任务） #

如果你的myfunc执行时间很长（比如超过10秒），不适合让用户在请求过程中一直等待，那更优的方案是把并行任务放到异步队列里处理：

简单示例： #

1. 安装Celery和Redis（用Redis做消息中间件）：

pip install celery redis

2. 修改代码，将任务异步化：

from flask import Flask, jsonify
from celery import Celery
from joblib import Parallel, delayed
from time import sleep

def myfunc(x):
    sleep(5)
    return x

# 初始化Flask和Celery
application = Flask(__name__)
application.config['CELERY_BROKER_URL'] = 'redis://localhost:6379/0'
application.config['CELERY_RESULT_BACKEND'] = 'redis://localhost:6379/0'

celery = Celery(application.name, broker=application.config['CELERY_BROKER_URL'])
celery.conf.update(application.config)

# 定义Celery异步任务
@celery.task
def run_parallel_task():
    out = Parallel(n_jobs=-1, verbose=10)(delayed(myfunc)(i) for i in range(5))
    return sum(out)

@application.route('/', methods=['GET'])
def getresult():
    # 触发异步任务，立即返回任务ID
    task = run_parallel_task.delay()
    return jsonify({'task_id': task.id, 'status': '任务已启动'})

@application.route('/result/<task_id>', methods=['GET'])
def get_task_result(task_id):
    task = run_parallel_task.AsyncResult(task_id)
    if task.state == 'SUCCESS':
        return jsonify({'result': task.result, 'status': task.state})
    else:
        return jsonify({'status': task.state})

if __name__ == "__main__":
    application.run(threaded=True)

3. 启动Redis服务（EC2上可以用apt install redis-server或yum install redis安装），然后启动Celery Worker：

celery -A app.celery worker --loglevel=info --concurrency=4

4. 最后启动Flask应用或用Gunicorn部署，用户请求时会立即收到任务ID，之后可以通过/result/<task_id>查询并行任务的结果。

额外注意事项 #

• 在AWS EC2上，要确保实例的CPU核心数足够，比如你用n_jobs=-1，joblib会自动使用所有可用核心，所以要选择对应配置的EC2实例（比如t2.large是2核，c5.xlarge是4核）。
• 用joblib多进程时，确保myfunc和它依赖的对象是可序列化的（能被Python的pickle模块序列化），否则会抛出序列化错误。

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