python - 在 celery 中如何获得队列中的任务位置？

Question

我使用 Celery 和 Redis 作为代理，我可以看到队列实际上是一个以序列化任务作为项目的 Redis 列表。

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

更新:

我终于能够通过以下方式获得职位:

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

但它有点慢，因为它需要与所有 worker 沟通。

Answer 1

您提到的 inspect.reserved()/scheduled() 可能有效，但不是总是准确的，因为它只考虑任务 worker 已经预取。

Celery 不允许对队列进行带外操作，比如删除消息从队列中，或重新排序它们，因为它不会在分布式系统中扩展。消息可能还没有到达队列，这可能会导致在竞争条件下，实际上它不是具有事务性的顺序队列操作，而是来自多个位置的消息流。也就是说，Celery API 基于严格的消息传递语义。

可以直接在某些代理上访问队列Celery 支持(如 Redis 或数据库)，但这不是公共(public) 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)，这可用于检测分布式事件的顺序系统不依赖于每台机器上的系统时间同步(这是不可能实现的)。