Python芹菜任务工作流执行

Question

我对Python芹菜很陌生，并试图将芹菜用于工作流执行。工作流：任务工作流

在附图中，我试着展示了下面的文字说明的工作流程。

1. 客户端调用Task-1。
2. 在任务-1，任务-2和任务-3的成功执行(同时)和任务-4执行，如果任务-1失败。
3. 在成功的任务-2，任务-5和任务-6被执行(同时)和任务-7执行，如果任务-2失败。
4. 在任务-5的成功时，大量的任务-8(同时)被执行(同时)，如果任务-5失败，任务-9将被执行。

当客户端调用Task-1时，它传递一个在成功和失败场景中传递给下一个任务所需的参数。

我知道画布可以用芹菜来设计工作流。但是，我无法弄清楚如何用不同层次的成功和失败来处理这么大的工作流程。请建议我应该如何使用芹菜和帆布来规划这样的工作流程。任何教程或例子的链接也将是一个很大的帮助。

Answer 1

你试过这样的东西吗？(伪码)

workflow = (t1.s().set(link_error=t4.s()) | 

            group((t2.s().set(link_error=t7.s()) |

                   group(t6.s(),
                        (t5.s().set(link_error=t9.s()) |
                         group(t8.s(i) for i in xrange(100)))),

                  t3.s())
            )
).apply_async()

link_error是错误的回调。

http://docs.celeryproject.org/en/latest/userguide/canvas.html

Answer 2

您可以有一个包装器任务来处理调用其他任务的逻辑。

您的代码可能会以这样的方式结束。

@celery.task(name='tasks.wrapper_task')
def wrapper_task(one_arg):
    if task_one(one_arg):
        task_three.apply_async(arg_one)
        if task_two(arg_one):
            task_six.apply_async(arg_one)
            if task_five(arg_one): 
                task_eight.apply_async(1)
                task_eight.apply_async(2)
                # etc...
            else:
                task_nine(arg_one)
        else:
            task_seven(arg_one)
    else:
        task_four(arg_one)

我还没有为你确定所有的任务。正如您所看到的，并非所有的任务都需要是芹菜任务，除非您需要在其他情况下异步调用它们。

我认为“失败”的意思是返回错误，但您可以很容易地将代码修改为使用try / except。