python - Celery - 在另一个任务结束时安排周期性任务

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

对此的直接解决方案要求您能够即时添加/删除节拍调度程序条目。在回答这个问题时...

How to dynamically add / remove periodic tasks to Celery (celerybeat)

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

您在这里混淆了两个概念。 “事件驱动工作”的概念和“批量计划驱动工作”的想法(这实际上只是事件按计划发生的第一种情况)。如果你真的考虑你在这里做什么，你会发现有一组相当复杂的边缘情况。消息本质上是分布的 当从不同消息产生的组开始创建冲突的条目时会发生什么？当您发现自己处于先前预定的垃圾山下时，您会怎么做？

在使用消息系统时，您实际上是在寻求构建递归树。做某事并产生更多消息以做更多事情的工作轴。除了这些循环(有意或无意)最终实现其基本情况并终止。

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