如果尚未使用celery安排任务,则允许执行任务

Question

我正在使用Celery来处理我正在开发的Django应用程序中的任务调度,我正在使用Django数据库进行测试.

我只是想几件事情要处理,只有当它不是已经计划任务的执行,或者在这样的提议进步文章,到目前为止,但没有工作.

像这样的东西:

task.py

@task()
def add(x, y):
   return x + y

然后当你按照以下方式调用它两次时:

import myapp.tasks.add

myapp.tasks.add.apply_async((2,2), task_id=1, countdown=15)
myapp.tasks.add.apply_async((2,2), task_id=2, countdown=15)

它应该允许一个基于的实例countdown=15.如果有另一个正在运行或等待的话,我怎么能完成第二个调用从不执行它？

Answer 1

接受的答案的一个问题是它很慢.检查任务是否已在运行涉及调用代理,然后迭代运行和活动任务.如果您想快速排队任务,这将无法正常工作.此外,当前解决方案具有较小的竞争条件,因为2个进程可以检查任务是否已经排队等同(发现它不是),这将排队2个任务.

更好的解决方案是我称之为去抖动的任务.基本上,每次排队任务时都会增加一个计数器.当任务开始时,你减少它.使用redis然后它都是原子的.

例如

排队任务:

conn = get_redis()
conn.incr(key)
task.apply_async(args=args, kwargs=kwargs, countdown=countdown)

然后在任务中,您有2个选项,是否要在第一个排队(节流)后15秒执行任务,或者在最后一个排队后15秒执行任务(去抖动).也就是说,如果我们继续尝试运行相同的任务,我们是否延长了计时器,或者我们只是等待第一个计时器15并忽略排队的其他任务.

容易支持两者,这里是debounce,我们等到任务停止排队:

conn = get_redis()
counter = conn.decr(key)
if counter > 0:
    # task is queued
    return
# continue on to rest of task

油门版本:

counter = conn.getset(key, '0')
if counter == '0':
    # we already ran so ignore all the tasks that were queued since
    return
# continue on to task

该解决方案相对于接受的另一个好处是,密钥完全在您的控制之下.因此,如果您希望执行相同的任务,但仅针对不同的ID /对象执行一次,则将其合并到您的密钥中.

更新

考虑到这一点,你可以更轻松地完成油门版本,而无需排队任务.

节流v2(排队任务时)

conn = get_redis()
counter = conn.incr(key)
if counter == 1:
    # queue up the task only the first time
    task.apply_async(args=args, kwargs=kwargs, countdown=countdown)

然后在任务中将计数器设置回0.

您甚至不必使用计数器,如果您有一个集合,您可以将密钥添加到集合中.如果你回到1,那么密钥不在集合中,你应该排队任务.如果返回0,则密钥已经在集合中,因此不要对任务进行排队.