FastAPI中处理长时间IO密集型任务的最佳方案探讨

Background Task是否适用？ #

完全不适用。FastAPI的Background Task是绑定请求生命周期的实现：

• 一旦服务进程重启、崩溃，或者worker被超时回收（比如Uvicorn的--timeout-keep-alive配置），未完成的任务会直接丢失，这就是你遇到可靠性问题的核心原因。
• 它的设计初衷是处理短时间、非关键的后台操作（比如发送通知、日志上报），完全不适合你这种耗时超1小时、需要可靠执行的任务。

适合你的解决方案 #

结合你每日最多100个IO密集型长任务的场景，推荐以下几种轻量、可靠的方案：

1. 用RQ（Redis Queue）实现异步队列 #

RQ是基于Redis的轻量级任务队列，配置和使用都比Celery简单太多：

• 任务会持久化到Redis，服务重启后未完成的任务可以继续执行；
• 天然支持IO密集型任务，不需要复杂的并发配置；
• 只需安装rq和Redis，写个简单的任务函数，就能把状态检查逻辑丢进队列，后台worker会自动执行。
• 核心逻辑示例：

  # 定义任务函数
  def check_external_task_status(task_id, db_session):
      while True:
          # 调用外部API检查状态
          status = call_external_api(task_id)
          if status in ["completed", "failed"]:
              # 更新数据库状态
              db_session.query(Task).filter(Task.id == task_id).update({"status": status})
              db_session.commit()
              break
          # IO密集型任务用sleep等待，不占用CPU
          time.sleep(60)  # 每分钟检查一次
  
  # FastAPI接口中提交任务
  from rq import Queue
  from redis import Redis
  
  redis_conn = Redis(host="localhost")
  q = Queue(connection=redis_conn)
  
  @app.post("/start-task")
  def start_task(task_id: str, db: Session = Depends(get_db)):
      # 提交任务到队列
      q.enqueue(check_external_task_status, task_id, db)
      return {"message": "Task started"}
  

  启动RQ worker只需执行命令：rq worker。

2. 基于现有数据库实现极简任务队列 #

如果不想引入Redis依赖，可以直接用你当前的数据库做任务持久化：

• 创建一个task_queue表，字段包括：task_id、target_params、status（待执行/执行中/完成/失败）、check_interval、created_at；
• 写一个独立的后台脚本（或者用FastAPI启动一个后台线程），定时轮询这个表，取出status=待执行的任务，标记为执行中后开始状态检查逻辑；
• 完成后更新任务状态到数据库，失败的话可以设置重试次数。
  这种方案完全不需要额外组件，每日100个任务的量级完全够用，可靠性拉满。

3. 关于Celery：是否小题大做？ #

确实有点杀鸡用牛刀。Celery需要配置Broker（Redis/RabbitMQ）、Result Backend，还需要管理worker、beat等进程，对于每日100个任务的场景来说，配置和维护成本过高。除非你未来有大规模任务扩容的明确需求，或者团队已经有Celery的技术栈，否则不推荐。

内容的提问来源于stack exchange，提问作者Python geek