python - celery :中止或撤销和弦中的所有任务

Question

我将以下设置与 Redis 代理和后端结合使用:

和弦([A, A, A, ...])(B)

• 任务 A 进行一些检查。它使用 AbortableTask 作为基础并定期检查 task.is_aborted() 标志。
• 任务B通知用户计算结果

用户可以中止 A 任务。不幸的是，当在所有任务 A 实例上调用 AbortableAsyncResult(task_a_id).abort() 时，只有事件的实例被中止。工作人员尚未收到的任务的状态更改为 ABORTED，但它们仍在处理中并且 is_aborted() 标志返回 False。

我当然可以 revoke() 挂起的任务而不是 abort() -ing 它们，但问题是在那种情况下和弦主体(任务 B ) 不再执行。

如何停止所有挂起和正在运行的任务 A 实例，同时仍确保任务 B 运行？

Answer 1

只需获取 A 的所有实例的 ID 列表并停止它们。

考虑这个简单的和弦

from celery import chord 

my_chord = chord(a.si() for i in range(300))(b.si())

现在您可以从 my_chord 实例中获取子任务列表(a 任务的所有实例)，使用

for taks in my_chord.parent.subtasks:
    print(task.id)

现在您可以对这些任务实例做任何您想做的事情。例如，您可以撤销所有的，而不管它们的当前状态如何。

from celery.task.control import revoke

for task in my_chord.parent.subtasks:
    revoke(task.id, terminate=True)

revoke 默认情况下只会终止挂起的任务。但是，如果您将 terminate=True 传递给它，它也会终止正在执行的任务。

另外，chord的回调函数会在它的所有子任务执行成功后被调用。由于取消chord子任务，回调函数不会被调用，chord任务失败。因此，您必须重试回调任务。