在芹菜中如何获得队列中的任务位置？

Question

我正在使用Celery和Redis作为代理,我可以看到队列实际上是一个redis列表,其中序列化任务作为项目.

我的问题是,如果我有一个AsyncResult对象作为调用的结果<task>.delay(),有没有办法确定项目在队列中的位置？

更新:

我终于可以使用以下方式获得该职位:

from celery.task.control import inspect
i = inspect()
i.reserved()

但它有点慢,因为它需要与所有工人沟通.

Answer 1

在inspect.reserved()/scheduled()你提到的可能的工作,但并不总是准确的,因为它只考虑到工人已预取的任务.

Celery不允许队列中的带外操作,例如从队列中删除消息或重新排序它们,因为它不会在分布式系统中扩展.消息可能尚未到达队列,这可能导致竞争条件,并且实际上它不是具有事务操作的顺序队列,而是源自多个位置的消息流.也就是说,Celery API基于严格的消息传递语义.

可以直接在Celery支持的某些代理(如Redis或Database)上访问队列,但这不是公共API的一部分,并且不鼓励您这样做,但当然如果您不打算支持大规模的操作你应该做最方便的事情并放弃我的建议.

如果这只是为了让用户知道他的工作何时完成,那么我确信你可以想出一个算法来预测任务何时执行,如果你只有队列的长度和时间插入每个任务的位置.

第一个只是一个redis.len("celery"),后者你可以通过听取task_sent信号来添加自己:

from celery.signals import task_sent

@task_sent.connect
def record_insertion_time(id, **kwargs):
   redis.zadd("celery.insertion_times", id)

在这里使用排序集:http://redis.io/commands/zadd

对于纯消息传递解决方案,您可以使用专用监视器来使用Celery事件流并预测任务何时完成. http://docs.celeryproject.org/en/latest/userguide/monitoring.html#event-reference

(只是注意到任务发送缺少文档中的时间戳字段,但是随着该事件发送时间戳,所以我将修复它).

事件还包含一个"时钟"字段,它是一个逻辑时钟(请参阅http://en.wikipedia.org/wiki/Lamport_timestamps),这可用于检测分布式系统中事件的顺序,而不依赖于系统每台机器上的时间要同步(这是不可能实现的).