FastAPI中替代Flask-SocketIO start_background_task()的方案有哪些？

在FastAPI中，你可以通过两种常用方式实现类似Flask-SocketIO中start_background_task的后台任务功能，适配WebSocket场景：

1. 使用FastAPI内置的BackgroundTasks #

BackgroundTasks是FastAPI提供的轻量后台任务管理工具，支持同步和异步任务，会在WebSocket会话的处理流程中自动调度任务执行。

代码示例 #

from fastapi import FastAPI, WebSocket, BackgroundTasks

app = FastAPI()

# 同步后台任务
def sync_bg_task():
    # 这里写你的同步任务逻辑，比如数据处理、文件操作等
    print("同步后台任务执行中")

# 异步后台任务
async def async_bg_task():
    # 这里写你的异步任务逻辑，比如异步IO操作
    await some_async_operation()
    print("异步后台任务执行中")

@app.websocket("/ws")
async def websocket_handler(websocket: WebSocket, background_tasks: BackgroundTasks):
    await websocket.accept()
    while True:
        data = await websocket.receive_text()
        
        # 添加同步任务到后台
        background_tasks.add_task(sync_bg_task)
        # 添加异步任务到后台
        background_tasks.add_task(async_bg_task)
        
        await websocket.send_text(f"已接收消息: {data}")

2. 直接使用asyncio.create_task #

如果你的后台任务是异步的，或者希望任务完全独立于WebSocket会话生命周期，可以直接使用Python标准库asyncio的create_task方法，将任务交给事件循环管理。

代码示例 #

import asyncio
from fastapi import FastAPI, WebSocket

app = FastAPI()

async def async_bg_task():
    # 异步任务逻辑，比如调用异步API、数据库操作
    await asyncio.sleep(3)
    print("独立异步后台任务完成")

@app.websocket("/ws")
async def websocket_handler(websocket: WebSocket):
    await websocket.accept()
    while True:
        data = await websocket.receive_text()
        
        # 启动异步后台任务
        asyncio.create_task(async_bg_task())
        
        await websocket.send_text(f"已接收消息: {data}")

同步任务适配 #

如果需要执行同步任务，可通过asyncio.to_thread将同步函数包装为异步任务，再用create_task启动：

def sync_bg_task():
    print("同步后台任务执行中")

# 在WebSocket处理函数内
asyncio.create_task(asyncio.to_thread(sync_bg_task))

两种方案的区别 #

• BackgroundTasks：任务与WebSocket会话的生命周期绑定，FastAPI会确保在会话处理的合适时机执行任务，适合需要和请求/会话流程联动的场景。
• asyncio.create_task：任务直接由事件循环管理，独立于WebSocket会话，适合不需要和会话绑定的长期异步任务，但需注意如果任务需要访问WebSocket连接，要确保连接仍处于活跃状态，避免报错。

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