如何使用Celery内置功能获取任务的执行时长？

核心思路：借助Celery任务事件系统 #

Celery会自动触发task-started、task-succeeded等关键事件，这些事件自带时间戳，我们只需监听事件并计算时间差，就能得到任务运行时长，无需在任务代码里手动写计时逻辑。

具体实现步骤 #

1. 开启Celery事件功能 #

启动worker时加上--events参数，临时启用事件：

celery -A 你的应用名 worker --loglevel=info --events

或者在Celery配置文件中永久开启：

# celeryconfig.py
worker_send_task_events = True
task_send_sent_event = True

2. 编写事件监听器 #

写一个独立脚本监听事件流，提取时间戳计算时长：

from celery import Celery
from celery.events import EventReceiver

# 初始化Celery应用
app = Celery('你的应用名')
app.config_from_object('celeryconfig')

# 存储任务开始时间的临时字典
task_start_times = {}

def process_event(event):
    event_type = event['type']
    task_id = event['uuid']
    
    if event_type == 'task-started':
        # 记录任务启动的Unix时间戳
        task_start_times[task_id] = event['timestamp']
    elif event_type == 'task-succeeded':
        # 计算并输出时长
        start_time = task_start_times.get(task_id)
        if start_time:
            run_duration = event['timestamp'] - start_time
            print(f"任务 {task_id} 运行时长: {run_duration:.2f} 秒")
            # 这里可以把时长写入日志或数据库，按需扩展
            del task_start_times[task_id]  # 清理已处理任务的缓存

# 建立连接并开始监听
with app.connection() as conn:
    receiver = EventReceiver(conn, handlers={
        'task-started': process_event,
        'task-succeeded': process_event,
    })
    print("开始监听Celery任务事件...")
    receiver.capture(limit=None, timeout=None)

3. 运行监听器 #

直接执行上述脚本，它会持续监听Celery事件，任务完成时自动输出运行时长。

额外提示 #

• 若要统计失败任务时长，可添加监听task-failed事件
• 事件中的timestamp是UTC时间戳，如需转为本地时间，可使用datetime.fromtimestamp(event['timestamp'])处理
• 若只是查看时长，Celery官方监控工具Flower也能直接展示，但上述方法更适合自定义统计需求

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