FastAPI异步任务处理与轮询机制实战-AI智能范式网

FastAPI异步任务处理与轮询机制实战

我行我素12334·2026-11-05·阅读 23 GEO检测（免费）

FastAPI异步任务Celery

|难度：普通|阅读约 23 分钟

1. 异步任务处理的必要性

在Web开发中，我们经常会遇到需要处理耗时操作的情况。比如用户上传一个大文件需要后台处理，或者需要调用第三方API获取数据但响应时间较长。传统同步处理方式会让用户一直等待，严重影响用户体验。

FastAPI作为现代Python异步框架，提供了强大的后台任务处理能力。通过结合异步任务和轮询机制，我们可以实现这样的工作流程：立即响应用户请求 -> 后台处理任务 -> 允许用户随时查询进度 -> 最终返回完整结果。

这种模式特别适合以下场景：

文件处理（转码、分析等）
复杂计算或机器学习推理
需要调用多个外部服务的聚合操作
任何执行时间超过3秒的操作

需要模型API调用？免费领10W Token，多模型网关一键接入 Claude、DeepSeek 等主流模型。

2. FastAPI后台任务基础实现

2.1 使用BackgroundTasks

FastAPI内置了BackgroundTasks类，这是实现后台任务最简单的方式：

python复制from fastapi import BackgroundTasks, FastAPI

app = FastAPI()

def process_file(filename: str):
    # 模拟耗时操作
    import time
    time.sleep(5)
    print(f"处理完成文件: {filename}")

@app.post("/upload/")
async def upload_file(background_tasks: BackgroundTasks, filename: str):
    background_tasks.add_task(process_file, filename)
    return {"message": "文件已接收，正在后台处理"}

这种方式的优点是简单直接，但缺点是无法获取任务状态或结果。任务一旦提交就会在后台运行，与原始请求完全分离。

2.2 更强大的Celery集成

对于需要状态跟踪和结果获取的复杂场景，Celery是更好的选择。以下是基本配置：

python复制# celery_app.py
from celery import Celery

celery = Celery(
    'worker',
    broker='redis://localhost:6379/0',
    backend='redis://localhost:6379/1'
)

@celery.task
def process_file_task(filename):
    # 耗时处理逻辑
    return {"status": "completed", "filename": filename}

然后在FastAPI中调用：

python复制from celery_app import process_file_task

@app.post("/celery-upload/")
async def celery_upload(filename: str):
    task = process_file_task.delay(filename)
    return {"task_id": task.id}

3. 任务状态轮询实现

3.1 基础轮询接口设计

实现轮询需要两个关键接口：

提交任务并返回任务ID
通过任务ID查询状态

python复制from fastapi import FastAPI
from celery.result import AsyncResult
from celery_app import celery

@app.get("/task-status/{task_id}")
async def get_status(task_id: str):
    task_result = AsyncResult(task_id, app=celery)
    return {
        "task_id": task_id,
        "status": task_result.status,
        "result": task_result.result if task_result.ready() else None
    }

3.2 状态缓存优化

频繁查询Celery后端可能造成性能压力。可以添加Redis缓存层：

python复制from fastapi_cache import FastAPICache
from fastapi_cache.backends.redis import RedisBackend
from fastapi_cache.decorator import cache

@app.get("/cached-task-status/{task_id}")
@cache(expire=30)  # 缓存30秒
async def get_cached_status(task_id: str):
    task_result = AsyncResult(task_id, app=celery)
    return {
        "status": task_result.status,
        "ready": task_result.ready()
    }

4. 前端轮询策略实现

4.1 JavaScript轮询示例

前端可以使用简单的setInterval实现轮询：

javascript复制async function checkStatus(taskId) {
  const response = await fetch(`/task-status/${taskId}`);
  const data = await response.json();
  
  if (data.status === 'SUCCESS') {
    clearInterval(pollingInterval);
    // 处理完成结果
  } else if (data.status === 'FAILURE') {
    clearInterval(pollingInterval);
    // 处理错误
  }
}

const pollingInterval = setInterval(() => checkStatus(taskId), 2000);

4.2 更优雅的Exponential Backoff

指数退避算法可以减少不必要的请求：

javascript复制let delay = 1000; // 初始1秒
const maxDelay = 10000; // 最大10秒

async function pollWithBackoff(taskId) {
  try {
    const response = await fetch(`/task-status/${taskId}`);
    const data = await response.json();
    
    if (data.status === 'SUCCESS' || data.status === 'FAILURE') {
      // 处理最终状态
      return;
    }
    
    // 未完成，继续轮询
    delay = Math.min(delay * 1.5, maxDelay);
    setTimeout(() => pollWithBackoff(taskId), delay);
  } catch (error) {
    console.error('轮询错误:', error);
    delay = Math.min(delay * 2, maxDelay);
    setTimeout(() => pollWithBackoff(taskId), delay);
  }
}

5. 进阶优化与最佳实践

5.1 任务超时与重试机制

为Celery任务添加可靠的超时和重试：

python复制from celery.exceptions import SoftTimeLimitExceeded

@celery.task(bind=True, max_retries=3, soft_time_limit=60)
def reliable_task(self, filename):
    try:
        # 任务逻辑
    except SoftTimeLimitExceeded:
        self.retry(countdown=10)
    except Exception as exc:
        self.retry(exc=exc, countdown=30)

5.2 进度报告实现

对于长时间任务，可以实时报告进度：

python复制@celery.task(bind=True)
def long_running_task(self, data):
    total = len(data)
    for i, item in enumerate(data):
        process_item(item)
        self.update_state(
            state='PROGRESS',
            meta={'current': i + 1, 'total': total}
        )

前端可以显示进度条：

javascript复制// 在轮询响应处理中
if (data.status === 'PROGRESS') {
  const progress = (data.result.current / data.result.total) * 100;
  updateProgressBar(progress);
}

5.3 WebSocket实时更新

替代轮询的更高效方案：

python复制from fastapi import WebSocket

@app.websocket("/ws/task-status/{task_id}")
async def websocket_task_status(
    websocket: WebSocket,
    task_id: str
):
    await websocket.accept()
    while True:
        task_result = AsyncResult(task_id)
        await websocket.send_json({
            "status": task_result.status,
            "result": task_result.result if task_result.ready() else None
        })
        if task_result.ready():
            break
        await asyncio.sleep(1)  # 控制检查频率

6. 性能监控与错误处理

6.1 添加Prometheus监控

监控后台任务的关键指标：

python复制from prometheus_client import Counter, Histogram

TASK_STARTED = Counter('task_started', '任务启动次数', ['task_name'])
TASK_TIME = Histogram('task_seconds', '任务执行时间', ['task_name'])

@celery.task
def monitored_task():
    TASK_STARTED.labels(task_name='monitored_task').inc()
    start_time = time.time()
    
    try:
        # 任务逻辑
    finally:
        TASK_TIME.labels(task_name='monitored_task').observe(time.time() - start_time)

6.2 结构化错误处理

统一错误响应格式：

python复制class TaskError(Exception):
    def __init__(self, message, code):
        self.message = message
        self.code = code

@app.exception_handler(TaskError)
async def task_error_handler(request, exc):
    return JSONResponse(
        status_code=exc.code,
        content={
            "error": {
                "message": exc.message,
                "type": "TASK_ERROR"
            }
        }
    )

7. 安全考虑与防护

7.1 任务ID安全性

使用UUID替代自增ID防止猜测：

python复制import uuid

def create_task_id():
    return str(uuid.uuid4())

@app.post("/secure-task/")
async def create_secure_task():
    task_id = create_task_id()
    # 存储task_id与用户关联
    return {"task_id": task_id}

7.2 速率限制

防止轮询接口被滥用：

python复制from fastapi import Request
from fastapi.middleware import Middleware
from slowapi import Limiter
from slowapi.util import get_remote_address

limiter = Limiter(key_func=get_remote_address)

@app.get("/rate-limited-status/{task_id}")
@limiter.limit("10/minute")
async def get_limited_status(request: Request, task_id: str):
    # 状态检查逻辑

8. 测试策略

8.1 单元测试示例

测试后台任务提交：

python复制from fastapi.testclient import TestClient

def test_background_task():
    client = TestClient(app)
    response = client.post("/upload/", json={"filename": "test.txt"})
    assert response.status_code == 200
    assert response.json()["message"] == "文件已接收，正在后台处理"

8.2 集成测试示例

测试完整轮询流程：

python复制def test_polling_flow():
    client = TestClient(app)
    
    # 提交任务
    submit_response = client.post("/celery-upload/", json={"filename": "test.txt"})
    task_id = submit_response.json()["task_id"]
    
    # 轮询直到完成
    while True:
        status_response = client.get(f"/task-status/{task_id}")
        status = status_response.json()["status"]
        
        if status == "SUCCESS":
            break
        elif status == "FAILURE":
            pytest.fail("任务执行失败")
        
        time.sleep(0.5)  # 短暂等待
    
    # 验证最终结果
    final_response = client.get(f"/task-status/{task_id}")
    assert final_response.json()["result"]["filename"] == "test.txt"

9. 部署注意事项

9.1 Celery Worker配置

生产环境推荐使用gevent提高并发：

bash复制celery -A celery_app worker --loglevel=info --pool=gevent --concurrency=100

9.2 任务结果过期设置

防止Redis被旧结果填满：

python复制celery = Celery(
    'worker',
    broker='redis://localhost:6379/0',
    backend='redis://localhost:6379/1',
    result_expires=3600  # 1小时后过期
)

10. 实际项目经验分享

在真实项目中，我们发现几个关键点：

任务粒度控制：不要把任务拆得过细，每个任务应该有足够的工作量来抵消调度开销。经验值是每个任务至少执行100ms以上。
结果序列化：Celery任务返回的结果需要可序列化。对于复杂对象，建议转换为字典或使用专门的序列化方法。
内存泄漏排查：长时间运行的Worker可能出现内存泄漏。定期重启Worker（如使用--max-tasks-per-child参数）可以缓解这个问题。
任务去重：对于幂等操作，可以在任务开始时检查是否已经有相同参数的正在运行的任务，避免重复计算。
本地开发配置：开发环境可以使用如下配置加速调试：

python复制app.conf.update(
    task_always_eager=True,  # 本地直接执行不排队
    task_eager_propagates=True,
)