撤销具有相同 args/kwargs 的 celery 任务

Question

想象一下有一个长时间运行的任务，其中包含一组特定的 args 和 kwargs。在开始新任务之前是否有机会撤销所有正在运行和挂起的具有相同 args/kwargs 的任务，因为我只对最后添加的任务的结果感兴趣。 （底层数据在两次调用之间发生变化）

我尝试迭代 inspect.active()、inspect.registered() 和 inspect.scheduled() 的结果以获取所有任务，然后 filter/revoke 那些与我的 args 和 kwargs 相关的任务。

但这并不可靠，因为检查所有工人和寻找任务的时间太长了。

谁能帮我找到正确的方向？

Answer 1

如果检查工人的速度很慢，最好通过辅助数据存储进行检查，例如像 redis 这样的键值存储。你为一个函数添加了一个“锁”，所以你知道你已经启动了它。

排队任务时：

• 检查您的商店中是否已存在
  • 是：引发异常
  • 否：将相关任务信息作为商品添加到您的商店。
• 将任务添加到 celery 任务队列

一个工人将执行它

• 执行它
• 从存储中移除

---

有一个基于redis的现成实现：

celery-once (permalink to current commit)。

您必须将其指定为任务的base

from celery import Celery
from celery_once import QueueOnce

@celery.task(base=QueueOnce)
def sum(a, b):
    ...

您还可以指定要考虑的参数和更多设置，请阅读那里的自述文件了解更多详细信息。