Flask集成Celery遇RuntimeError：请求上下文外工作问题排查

问题描述 #

开发Flask+Celery项目时，接收大量POST请求并分发异步任务，持续触发RuntimeError: Working outside of request context错误，报错点为Celery任务中调用Flask的request对象。

核心原因 #

Celery Worker是独立于Flask服务的进程，执行任务时完全脱离HTTP请求上下文。你在process_live任务中直接调用request.is_json、request.get_json()，但request对象仅绑定于当前活跃的HTTP请求，Worker进程中不存在该上下文，因此触发错误。

解决方案 #

1. 传递请求数据到Celery任务，而非依赖request对象 #

在Flask视图函数中先提取请求数据，将数据作为参数传递给Celery任务，彻底让任务与请求上下文解耦。

修改all.py视图函数

from flask import Blueprint, make_response, jsonify, request
import os
from .task import process_live

bp = Blueprint("all", __name__)

@bp.route("/")
def index():
    return "Hello!"

@bp.route("/api/live", methods=["POST"])
def live():
    # 在视图层完成请求数据提取
    if request.is_json:
        data = request.get_json()
        # 将数据作为参数传入Celery任务
        process_live.delay(data)
        return make_response(jsonify({"message":"JSON received"}), 200)
    else:
        return make_response(jsonify({"message": "Invalid JSON format"}), 400)

修改task.py Celery任务

from liveapp import celery
from datetime import datetime
import re
from numpy import array, concatenate
from .utils import series_to_supervised, model_retrain, model_repredict, model_predict
import pyodbc
import gc

@celery.task()
def process_live(data):  # 接收传入的请求数据
    # 直接使用传入的data执行业务逻辑，不再依赖request对象
    # ,,,,,,,,,,,do something with data,,,,

2. 确保Celery Worker加载正确的Flask应用配置 #

检查celery_worker.py（你未提供该文件内容），需确保它正确初始化Flask应用并关联Celery实例：

from liveapp import celery
from liveapp.factory import create_app

# 创建Flask应用，完成Celery与应用的绑定
app = create_app(celery=celery)

3. 重启服务与Worker #

• 重启Flask服务
• 重启Celery Worker，加载最新代码：

D:\new test>celery -A celery_worker.celery worker -l info -P solo

关键说明 #

• Celery任务与原HTTP请求完全解耦，绝对不能依赖请求上下文对象（如request、session），所有所需数据必须通过参数传递。
• 你的celery_utils.py中ContextTask已确保任务运行在Flask应用上下文（提供数据库连接、应用配置等），但应用上下文≠请求上下文，无法替代HTTP请求级别的对象。

内容的提问来源于stack exchange，提问作者Amanda Choy Siew Wen