fastapi中websocket场景必须用asyncio.create_task()而非backgroundtasks，因后者依赖http请求生命周期而websocket无此语义；需手动异常处理并确保变量作用域有效。
FastAPI 本身不“实现”异步任务，它只是提供了一套运行异步代码的环境；真正决定任务是否异步、是否解耦、是否可靠，取决于你选的是 BackgroundTasks、asyncio.create_task() 还是外部任务队列（如 Celery + Redis/RabbitMQ）。选错方案，轻则响应变慢，重则任务静默丢失。
BackgroundTasks 只适合短时、轻量、同进程内清理类操作
BackgroundTasks 是 FastAPI 内置的轻量机制，本质是把协程或同步函数扔进事件循环或线程池里“顺手执行”，但它和请求生命周期强绑定——请求一结束，任务才开始跑，且全程无持久、无重试、无状态跟踪。
- 常见错误现象：
time.sleep(10)或调用sklearn.predict()等 CPU 密集型操作，导致后续 HTTP 请求排队、Uvicorn 超时断连 - 适用场景：发一封确认邮件、写一条审计日志、触发一次缓存失效，耗时建议控制在 200ms 以内
- 不能跨请求共享状态：每个
BackgroundTasks实例只对单个请求有效，WebSocket 中完全不可用 - 别把它当 Celery 用——它没有 broker、没有 worker、没有任务队列，挂掉就没了
WebSocket 场景下必须用 asyncio.create_task()
WebSocket 是长连接，没有“请求完成”这个时机，BackgroundTasks 依赖的 Request 对象根本不存在，所以 add_task() 调用会静默失败。
python-script-generator
下载
快速生成专业的 Python 脚本和应用代码。一键创建完整项目结构，支持CLI、API、爬虫、Bot、Django等多种项目类型，包含完整的项目结构、配置文件、依赖管理、测试、README和文档。
- 正确做法：在
@app.websocket路由内直接调用asyncio.create_task(my_coroutine()) - 必须手动处理异常：未捕获的异常会让后台任务无声崩溃，建议包裹
try/except并打日志 - 注意作用域：任务中引用的变量（如 WebSocket 实例）需确保生命周期足够长，避免
RuntimeError: cannot reuse already awaited coroutine - 不要用
loop.run_in_executor包同步函数——除非你明确需要线程隔离，否则纯 async 协程更轻量、更可控
Celery + Redis 是生产级异步任务的下限
只要任务有以下任一特征，就必须上 Celery：需要重试、要求消息持久化、要跨服务调度、预期执行时间 > 5 秒、需监控任务状态、依赖 ACK 确认防止丢失。
- 配置关键点：
broker='redis://localhost:6379/0'中的 DB 编号别用 0（易被 flush），推荐用 1–15；result_backend建议也设为 Redis，方便查结果 - 启动命令要分开：
celery -A tasks worker --loglevel=info和uvicorn main:app必须是两个独立进程 - FastAPI 中调用不是
add()就完事：要用task.delay()或task.apply_async()，前者走默认序列化，后者可传countdown、expires等参数 - 别在 Celery task 函数里直接操作 FastAPI 的
Depends对象——它们只在请求上下文里有效；DB 连接、HTTP 客户端等要单独初始化
别踩这些隐性坑
异步任务最麻烦的地方不在写法，而在边界模糊处：比如你以为在用 async，其实底层库是同步阻塞的；你以为任务已发出，其实消息根本没进队列；你以为状态可查，其实 result_backend 没配对。
-
httpx.AsyncClient是真异步，requests.get再包run_in_executor也是假异步——只是挪了个线程，没减少 I/O 等待 - RabbitMQ 场景下，
delivery_mode=2必须显式设置，否则消息不持久，Broker 重启即丢 - Celery 的
task_id默认是 UUID4，但如果你用 Redis 做 result_backend，查结果得用AsyncResult(task_id).get()，别直接去 Redis key 里硬读 - 所有外部连接（Redis、RabbitMQ、数据库）都应做成单例 + 生命周期管理（
@app.on_event("startup")/"shutdown"），而不是每次任务都新建
大量免费API接口：立即使用
涵盖生活服务API、金融科技API、企业工商API、等相关的API接口服务。免费API接口可安全、合规地连接上下游，为数据API应用能力赋能！
