芹菜时间统计每任务名称

Question

我有一些相当繁忙的芹菜队列，但不知道哪些任务是有问题的。有没有一种方法可以对结果进行汇总以确定哪些任务需要很长时间？我有2-4台服务器上有10-20名工人。

使用redis作为代理，并将结果作为后端。我注意到Flower上的繁忙队列，但无法弄清楚如何获得按任务汇总的时间统计信息。

Answer 1

方法1：

如果已经启用了日志记录时，芹菜工人开始，他们记录所采取的每项任务的时间。

$ celery worker -l info -A your_app --logfile celery.log 

这会产生这样的

[2016-06-04 13:21:30,749: INFO/MainProcess] Task sig.add[a8b648eb-9674-44f0-90bd-71cfebe22f2f] succeeded in 0.00979363399983s: 3 
[2016-06-04 13:21:30,973: INFO/MainProcess] Received task: sig.add[7fd422e6-8f48-4dd2-90de-e213afbedc38] 
[2016-06-04 13:21:30,982: WARNING/Worker-2] called by small_task. LOL {'signal': <Signal: Signal>, 'result': 3, 'sender': <@task: sig.add of tasks:0x7fdf33146c50>} 

日志可以筛选具有succeeded in线。使用，[,:作为分隔符分割这些行，打印任务的名称和时间，然后对所有行进行排序。

$ grep ' succeeded in ' celery.log | awk -F'[ :\[]' '{print $9, $13}' | sort 
awk: warning: escape sequence `\[' treated as plain `[' 
sig.add 0.00775764500031s 
sig.add 0.00802627899975s 
sig.foo 12.00813863099938s 
sig.foo 15.00871706100043s 
sig.foo 12.00979363399983s 

正如你可以看到add是非常快的& foo是缓慢的。

方法2：

芹菜具有任务之后之前运行/ task_prerun_handler，task_postrun_handler信号。你可以连接函数来追踪时间，然后记下某处的时间。

from time import time 
from celery.signals import task_prerun, task_postrun 


tasks = {} 
task_avg_time = {} 
Average = namedtuple('Average', 'cum_avg count') 


@task_prerun.connect 
def task_prerun_handler(signal, sender, task_id, task, args, kwargs): 
    tasks[task_id] = time() 


@task_postrun.connect 
def task_postrun_handler(signal, sender, task_id, task, args, kwargs, retval, state): 
    try: 
     cost = time() - tasks.pop(task_id) 
    except KeyError: 
     cost = None 

    if not cost: 
     return 

    try: 
     cum_avg, count = task_avg_time[task.name] 
     new_count = count + 1 
     new_avg = ((cum_avg * count) + cost)/new_count 
     task_avg_time[task.name] = Average(new_avg, new_count) 
    except KeyError: 
     task_avg_time[task.name] = Average(cost, 1) 

    # write to redis: task_avg_time 

参考文献：https://stackoverflow.com/a/31731622/2698552