Celery多任务运行时任务进度元信息显示异常问题

核心原因 #

你遇到的问题本质是gevent协程模型下，Celery任务状态更新与读取的竞态条件，或是当前使用的Celery Backend不支持高并发协程读写，导致部分任务的meta信息未能及时同步到Backend中。

具体解决方案 #

1. 更换为支持协程并发的Celery Backend #

默认的rpc:// Backend在gevent环境下容易出现状态同步延迟或丢失，建议替换为Redis或Memcached这类高并发友好的Backend：

# celery_app.py 配置示例
from celery import Celery

app = Celery(
    'tasks',
    broker='amqp://user:pass@localhost:5672//',
    backend='redis://localhost:6379/0'  # 替换为Redis Backend
)

启动Redis服务后，重启Celery Worker测试多任务场景。

2. 强制刷新任务状态 #

在FastAPI查询进度时，调用AsyncResult.refresh()强制从Backend拉取最新状态，避免依赖本地缓存的过期数据：

# FastAPI 单个任务进度查询接口示例
from celery.result import AsyncResult
from fastapi import FastAPI

app = FastAPI()

@app.get("/curr_progress/{task_id}")
def get_progress(task_id: str):
    result = AsyncResult(task_id)
    result.refresh()  # 强制刷新，获取最新状态和meta
    return {
        "task_id": task_id,
        "status": result.status,
        "meta": result.info.get('meta') if result.info else None
    }

3. 调整Celery Worker的gevent配置 #

修改启动命令，优化协程并发数并关闭预取机制，避免任务调度导致的状态更新延迟：

celery -A celery_app worker -l info -P gevent --concurrency=10 --prefetch-multiplier=1

• --concurrency=10：根据服务器配置设置合适的协程数量
• --prefetch-multiplier=1：禁止Worker预取任务，确保状态更新及时提交

4. 规范任务进度更新逻辑 #

检查Celery任务中update_state的调用，确保每次更新都正确传入meta参数，且在任务执行过程中定期调用：

# Celery任务示例
@app.task(bind=True)
def long_running_task(self, total):
    for done in range(total):
        # 执行任务核心逻辑
        # 更新进度状态
        self.update_state(
            state='PROGRESS',
            meta={'done': done + 1, 'total': total}
        )
        # 协程环境下主动让出CPU，避免阻塞状态更新
        import gevent
        gevent.sleep(0.1)
    return {"result": "success"}

5. 优化FastAPI服务的并发能力 #

如果FastAPI使用默认单线程模式，多任务查询可能出现阻塞，建议启动时使用多线程/多进程模式：

uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4 --threads 2

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