python - 去抖 celery 任务？

Question

是否有用于消除 Celery 任务的标准方法？

例如，一个任务可以多次“启动”，但在一些延迟后只会运行一次:

def debounce_task(task):
    if task_is_queued(task):
        return
    task.apply_async(countdown=30)

Answer 1

这是我们使用 Redis 计数器的方法。所有这些可能都可以在装饰器中概括，但我们只将它用于特定任务(webhooks)

您的面向公众的任务是您从其他职能部门调用的任务。它需要在 Redis 中增加一个键。键由函数的参数组成，无论它们是什么(这确保计数器在各个任务中是唯一的)

@task
def your_public_task(*args, **kwargs):
    cache_key = make_public_task_cache_key(*args, **kwargs)
    get_redis().incr(cache_key)
    _your_task(*args, **kwargs, countdown=settings.QUEUE_DELAY)

注意缓存键函数是共享的(你希望每个函数中有相同的缓存键)，countdown设置。

然后，执行代码的实际任务执行以下操作:

@task
def _your_task(*args, **kwargs):
    cache_key = make_public_task_cache_key(*args, **kwargs)
    counter = get_redis().getset(cache_key, 0)
    # redis makes the zero a string.
    if counter == '0':
       return

    ... execute your actual task code.

这让您可以点击 your_public_task.delay(..)在你的 QUEUE_DELAY 内，你想要多少次都行。 , 它只会触发一次。