芹菜 - 在另一项任务结束时安排定期任务

Question

我想在另一组任务结束时动态安排Celery的周期性任务.

我知道如何使用Celery创建(静态)周期性任务:

CELERYBEAT_SCHEDULE = {
      'poll_actions': {
          'task': 'tasks.poll_actions',
          'schedule': timedelta(seconds=5)
      }
}

但我想从我的任务中动态创建周期性作业(并且可能有一种方法可以在达到某些条件时停止这些周期性作业(所有任务都已完成).

就像是:

@celery.task
def run(ids):
    group(prepare.s(id) for id in ids) | execute.s(ids) | poll.s(ids, schedule=timedelta(seconds=5))

@celery.task
def prepare(id):
    ...

@celery.task
def execute(id):
    ...

@celery.task
def poll(ids):
    # This task has to be schedulable on demand
    ...

Answer 1

对此的简单解决方案要求您能够动态添加/删除节拍调度程序条目。从回答这个问题开始...

如何向Celery动态添加/删除周期性任务（celerybeat）

这是不可能的。我怀疑它是否在过渡期间可用，因为......

您在这里混淆了两个概念。“事件驱动工作”的概念和“批计划驱动工作”的概念（这实际上只是事件按计划发生的第一种情况）。如果您真正考虑一下您在这里所做的事情，您会发现存在一组相当复杂的边缘情况。消息本质上是分布式的，当从不同消息生成的组开始创建冲突条目时会发生什么？当你发现自己被堆积如山的预定克鲁夫特淹没时，你会怎么做？

当使用消息传递系统时，您确实希望构建递归树。工作的主轴可以做一些事情并产生更多的消息来做更多的事情。除了这些循环（有意或无意）之外，这些循环最终实现其基本情况并终止。

无论您实际想要实现什么目标，答案都在于在消息传递系统和异步工作框架的限制内重新编码您的问题。