FastAPI中使用asyncio运行后台长时异步任务随机终止的问题及最佳实践

核心问题分析 #

你遇到的随机停止问题，主要源于以下几个常见原因：

• 后台任务抛出未捕获的异常，导致任务静默终止
• 请求结束后，请求级资源（如数据库连接）被回收，后台任务调用时出错
• 局部创建的ThreadPoolExecutor被垃圾回收，提前关闭线程池
• 事件循环对未跟踪的任务进行了自动清理

具体解决方案 #

1. 强制捕获任务异常，避免静默失败 #

asyncio.create_task创建的任务如果抛出异常，不会主动触发报错，只有在await或调用result()时才会暴露问题。必须给后台任务添加异常捕获逻辑：

async def run_heavy_task(...):
    try:
        # 原任务执行逻辑
    except Exception as e:
        # 用logging记录详细错误（建议替代print）
        print(f"后台任务执行失败: {str(e)}")
        # 可选：添加告警或错误补偿逻辑

也可以给任务绑定完成回调，统一处理异常：

def handle_task_exception(task):
    try:
        task.result()
    except Exception as e:
        print(f"后台任务异常: {str(e)}")

task = asyncio.create_task(run_heavy_task(...))
task.add_done_callback(handle_task_exception)

2. 使用FastAPI官方BackgroundTasks #

FastAPI提供的BackgroundTasks会自动管理任务生命周期，避免请求结束后任务被意外清理，是处理轻量后台任务的首选方案：

from fastapi import BackgroundTasks

async def from_template_to_conv(
    transaction_id: str,
    db,
    background_tasks: BackgroundTasks,  # 注入BackgroundTasks
):
    # ... 原有前置逻辑 ...

    # 定义后台任务（无需再用asyncio.create_task）
    async def run_heavy_task(...):
        # 带异常捕获的任务逻辑

    # 将任务加入后台队列
    background_tasks.add_task(run_heavy_task, transaction_id, question_dics, db, executor)

    return response

3. 避免使用请求级资源 #

代码中的db如果是请求依赖注入的实例，请求结束后连接可能被回收，导致后台任务操作数据库时失败。解决方法：

• 在后台任务中重新创建独立的数据库连接
• 调整数据库连接池配置，允许后台任务复用连接（如增大连接池容量、延长超时时间）

4. 全局管理线程池 #

不要在函数内部创建局部ThreadPoolExecutor，函数返回后该对象会被垃圾回收，线程池提前关闭导致任务中断。改为使用全局线程池：

# 模块级别定义全局线程池
global_executor = ThreadPoolExecutor(max_workers=1)

async def from_template_to_conv(...):
    # ... 逻辑 ...
    # 使用全局线程池
    background_tasks.add_task(run_heavy_task, transaction_id, question_dics, db, global_executor)

5. 跟踪后台任务（可选） #

如果需要监控任务状态，可以维护全局任务字典：

import asyncio
from typing import Dict

active_tasks: Dict[str, asyncio.Task] = {}

async def from_template_to_conv(...):
    # ... 逻辑 ...
    task = asyncio.create_task(run_heavy_task(...))
    active_tasks[transaction_id] = task
    # 任务完成后从字典移除
    task.add_done_callback(lambda t: active_tasks.pop(transaction_id, None))

代码修改示例 #

结合以上建议，调整你的代码片段：

from fastapi import BackgroundTasks, HTTPException
import asyncio
from concurrent.futures import ThreadPoolExecutor

# 全局线程池
global_executor = ThreadPoolExecutor(max_workers=1)

async def from_template_to_conv(
    transaction_id: str,
    db,
    background_tasks: BackgroundTasks,
):
    print("Getting templates")
    transaction = await fetch_transaction_with_workstreams(db, transaction_id)
    if len(transaction.workstreams) > 0:
        print("Transaction already has workstreams. Aborting.")
        return

    print("Creating workstreams")
    response = await process_transaction_based_on_template(db, transaction=transaction)
    print("Workstream created successfully.")

    questions = await fetch_transaction_questions(db, transaction_id)
    question_dics = [
        {
            "id": q.id,
            "content": q.content,
        }
        for q in questions
    ]

    async def run_heavy_task(transaction_id, questions, db, executor):
        try:
            for question in questions:
                # 注意：若db为请求级实例，建议在此重新创建独立连接
                transaction = await get_transaction(db, transaction_id)
                print(f"Fetched transaction {transaction_id}.")

                users = await get_users_by_organization(db, transaction.organization_id)
                print(f"Fetched users for transaction {transaction_id}.")
                user_ids = [u.id for u in users]
                print(user_ids)
                print(f"Automating questionnaire for question {question['id']}.")
                
                conversation = await create_automation_conversation(
                    db, user_ids, "deals", transaction.organization_id, transaction
                )
                print(f"Created conversation for transaction {transaction_id}.")

                response = await process_message_conversation(
                    question["id"],
                    conversation,
                    conversation.id,
                    question["content"],
                    user_ids,
                    db,
                )
                print(f"Processed message for question {question['id']}.")

                final_message = None
                async for message in response.body_iterator:
                    final_message = message

                if final_message:
                    print(f"Final message received for question {question['id']}: {schema.Message.parse_raw(final_message)}")
                else:
                    raise HTTPException(status_code=500, detail="Internal server error")
        except Exception as e:
            print(f"后台任务执行失败（transaction_id: {transaction_id}）: {str(e)}")

    background_tasks.add_task(run_heavy_task, transaction_id, question_dics, db, global_executor)

    return response

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