如果参数和任务名称已经在服务器中排队，是否可以跳过委派芹菜任务？

Question

说我有这个任务：

def do_stuff_for_some_time(some_id):
    e = Model.objects.get(id=some_id)
    e.domanystuff()

我就是这样使用它的：

do_stuff_for_some_time.apply_async(args=[some_id], queue='some_queue')

我面临的问题是有很多具有相同 arg 参数的重复性任务，而且它正在排队等待。

是否只有在相同的参数和相同的任务不在队列中时才可以应用异步？

Answer 1

celery-singleton 解决了这个需求

警告：需要 redis 代理（用于分布式锁）

pip install celery-singleton

使用Singleton 任务基类：

from celery_singleton import Singleton

@celery_app.task(base=Singleton)
def do_stuff_for_some_time(some_id):
    e = Model.objects.get(id=some_id)
    e.domanystuff()

来自文档：

对 do_stuff.delay() 的调用将排队一个新任务 或为当前排队/运行的实例返回 AsyncResult 任务

Answer 2

我会尝试混合使用 cache lock 和 task result backend 来存储每个任务的结果：

• 缓存锁将阻止具有相同参数的任务多次添加到队列中。 Celery 文档包含一个很好的缓存锁实现示例here，但如果您不想自己创建它，可以使用celery-once 模块。

• 对于任务结果后端，我们将使用推荐的django-celery-results，它会创建一个TaskResult 表，我们将在该表中查询任务结果。

示例：

• 安装和配置django-celery-results:

  settings.py:

  INSTALLED_APPS = (
      ...,
      'django_celery_results',
  )
  CELERY_RESULT_BACKEND = 'django-db'  # You can also use 'django-cache'
  

  ./manage.py migrate django_celery_results

• 安装和配置celery-once 模块：

  tasks.py:

  from celery import Celery
  from celery_once import QueueOnce
  from time import sleep
  
  celery = Celery('tasks', broker='amqp://guest@localhost//')
  celery.conf.ONCE = {
      'backend': 'celery_once.backends.Redis',
      'settings': {
          'url': 'redis://localhost:6379/0',
          'default_timeout': 60 * 60
       }
  }
  
  @celery.task(base=QueueOnce)
  def do_stuff_for_some_time(some_id):
      e = Model.objects.get(id=some_id)
      e.domanystuff()
  

  此时，如果要执行具有相同参数的任务，
  将引发 AlreadyQueued 异常。

• 让我们使用上面的：

  from django_celery_results.models import TaskResult
  
  try:
      result = do_stuff_for_some_time(some_id)
  except AlreadyQueued:
      result = TaskResult.objects.get(task_args=some_id)
  

注意事项：

• 请注意，在出现AlreadyQueued 异常时，参数=some_id 的初始任务可能不会被执行，因此它不会在TaskResult 表中产生结果。

  李>

• 注意代码中可能出错的所有内容并挂起上述任何进程（因为它会这样做！）。

补充阅读：

• 另一个Task with Lock DIY implementation
• django-celery-result 的 TaskResult 模型。

Answer 3

我不确定 celery 是否有这样的选择。不过，我想建议一种解决方法。

1) 为所有排队的 celery 任务创建一个模型。在该模型中，保存 task_name、queue_name 以及参数

2) 在该模型上为每个准备排队的 celery 任务使用 get_or_create。

3) 如果步骤 2 中 created = True，则允许将任务添加到队列中，否则不要将任务添加到队列中