Flask应用中异步方法未在后台执行问题排查

我之前也碰到过类似的场景——在Python 3.5.x的同步Flask环境里，处理需要好几分钟的GraphQL mutation，直接执行会阻塞请求，用户体验极差。你用asyncio没生效是因为Flask本身是同步框架，默认不会帮你维护异步事件循环，直接调用asyncio的方法其实还是会让主线程等待协程完成。下面给你几个适合Python 3.5.4的可行方案：

方案一：用threading启动后台线程（轻量首选） #

如果你的任务是IO密集型（比如调用外部API、读写数据库/文件），用线程是最简单的方式，不需要额外依赖，Python 3.5原生支持。

示例代码： #

首先定义一个要后台执行的任务函数：

import threading
import time
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def long_running_task(task_param):
    try:
        logger.info(f"开始执行后台任务，参数：{task_param}")
        # 模拟2-5分钟的耗时操作
        time.sleep(180)
        logger.info("后台任务执行完成")
    except Exception as e:
        logger.error(f"后台任务失败：{str(e)}")

然后在你的Graphene Mutation里启动线程，立即返回响应：

from graphene import Mutation, String, ObjectType
from flask_graphql import GraphQLView

class LongTaskMutation(Mutation):
    class Arguments:
        param = String(required=True)
    
    message = String()

    def mutate(root, info, param):
        # 启动后台线程，不等待执行完成
        threading.Thread(target=long_running_task, args=(param,), daemon=True).start()
        # 立即返回给用户的消息
        return LongTaskMutation(message="任务已提交到后台执行，请稍后查看结果")

class Mutation(ObjectType):
    long_task = LongTaskMutation.Field()

# 注册GraphQL视图的代码和之前一样

注意事项： #

• 用daemon=True可以让线程在主进程退出时自动终止，避免僵尸线程
• 后台任务里不要直接使用Flask的请求上下文（比如request对象），因为请求结束后上下文会被销毁
• 建议添加日志记录，方便排查任务执行中的问题

方案二：用multiprocessing启动子进程（CPU密集型首选） #

如果你的任务是CPU密集型（比如大量计算），因为Python的GIL限制，线程没法充分利用多核CPU，这时候用进程更合适。Python 3.5也原生支持multiprocessing模块。

示例代码： #

from multiprocessing import Process

# 同样先定义长耗时任务函数（和threading示例一样）
def long_running_task(task_param):
    # ... 任务逻辑 ...

class LongTaskMutation(Mutation):
    # ... 参数和返回定义和之前一样 ...

    def mutate(root, info, param):
        # 启动子进程
        Process(target=long_running_task, args=(param,), daemon=True).start()
        return LongTaskMutation(message="任务已提交到后台执行，请稍后查看结果")

注意事项： #

• 进程间的数据传递需要序列化（比如用pickle），所以传给任务的参数要能被序列化
• 进程的资源开销比线程大，适合真正需要多核计算的场景

方案三：用任务队列（生产环境首选） #

如果你的场景需要更可靠的任务管理（比如任务持久化、重试机制、状态查询），推荐用Celery这类任务队列工具。Python 3.5可以兼容Celery 4.x版本。

步骤： #

1. 安装依赖（注意版本适配）：

pip install celery==4.4.7 redis==3.5.3

这里用Redis作为消息中间件，你也可以用RabbitMQ。

2. 定义Celery实例和任务：

# celery_config.py
from celery import Celery

# 初始化Celery，用Redis作为 broker 和 backend
app = Celery('tasks', broker='redis://localhost:6379/0', backend='redis://localhost:6379/0')

@app.task
def long_running_task(task_param):
    # ... 你的长耗时任务逻辑 ...
    return f"任务完成，参数：{task_param}"

3. 在Mutation里调用任务：

from celery_config import long_running_task

class LongTaskMutation(Mutation):
    class Arguments:
        param = String(required=True)
    
    message = String()
    task_id = String()

    def mutate(root, info, param):
        # 发送任务到队列，立即返回任务ID
        task = long_running_task.delay(param)
        return LongTaskMutation(
            message="任务已提交到后台执行",
            task_id=task.id
        )

4. 启动Celery worker：

celery -A celery_config worker --loglevel=info

优势： #

• 任务持久化，即使服务重启，未完成的任务也能继续执行
• 支持任务重试、定时任务、任务状态查询（通过task_id可以用long_running_task.AsyncResult(task_id)查询状态）
• 适合大规模、高可靠性的生产环境

为什么asyncio没生效？ #

Flask是同步Web框架，默认的请求处理是同步的。你直接在mutation里调用asyncio的方法，比如loop.run_until_complete(coroutine())，会让主线程等待协程执行完成，本质还是阻塞的。如果要在Flask里用异步，需要用flask-async这类扩展，但Python 3.5的异步语法（比如async/await）刚引入，支持不够完善，而且Graphene在3.5版本里也没有原生的异步支持，所以不如上面的方案直接有效。

额外建议 #

• 不管用哪种方案，都建议给用户提供查询任务状态的接口（比如新增一个Query，通过任务ID查询执行状态/结果）
• 后台任务里的异常一定要捕获并记录，避免任务静默失败
• 如果是数据库操作，要确保线程/进程的数据库连接是独立的，避免连接共享导致的问题

内容的提问来源于stack exchange，提问作者Alexis.Rolland