芹菜:任务层次结构后的回调

2024-04-27 14:33:52 发布

您现在位置:Python中文网/ 问答频道 /正文

我正在使用一个webapp中的芹菜来启动一个任务层次结构。在

任务

我正在使用以下任务:

  • task_a
  • task_b
  • task_c
  • notify_user

Django视图启动多个task_a实例。它们中的每一个都进行一些处理,然后启动几个task_b实例。每一个都进行一些处理,然后启动几个task_c实例。在

可视化:

Tree

目标

我的目标是执行所有任务,并在整个层次结构完成后立即运行回调函数。此外,我希望能够将数据从最低的任务传递到顶层。

  1. 视图应该只是“启动”任务,然后返回。在
  2. 每个子任务依赖于父任务。父任务不直接依赖于子任务。父任务启动所有子任务后,可以停止它。在
  3. 只要父任务在子任务启动之前运行,所有事情都可以并行化。在
  4. 所有任务完成后,应该调用notify_user回调函数。在
  5. notify_user回调函数需要从task_cs访问数据

所有任务都应该是非阻塞的,因此task_b不应该等待所有task_c子任务完成。在

实现上述目标的正确方法是什么?在


Tags: 数据django实例函数视图目标task层次结构
2条回答

假设您有以下任务:

celery = Celery(
    broker="amqp://test:test@localhost:5672/test"
)
celery.conf.update(
    CELERY_RESULT_BACKEND = "mongodb",
)


@celery.task
def task_a(result):
    print 'task_a:', result
    return result

@celery.task
def task_b(result):
    print 'task_b:', result
    return result

@celery.task
def task_c(result):
    print 'task_c:', result
    return result

@celery.task
def notify_user(result):
    print result
    return result

对于给定的输入数据(绘制时):

^{pr2}$

您可以:

    a_group = []
    for ia, a in enumerate(tree):
        print "A%s:" % ia
        b_group = []
        for ib, b in enumerate(a):
            print " - B%s:" % ib
            for c in b:
                print '   -', c

            c_group = group([task_c.s(c) for c in b])

            b_group.append(c_group | task_b.s())

        a_group.append(group(b_group) | task_a.s())

    final_task = group(a_group) | notify_user.s()

它的表现是(别看,很难看:)

[[[__main__.task_c('C1'), __main__.task_c('C2'), __main__.task_c('C3')] | __main__.task_b(), [__main__.task_c('C4'), __main__.task_c('C5')] | __main__.task_b()] | __main__.task_a(), [[__main__.task_c('C6'), __main__.task_c('C7'), __main__.task_c('C8')] | __main__.task_b(), [__main__.task_c('C9')] | __main__.task_b()] | __main__.task_a()] | __main__.notify_user()

传递给notify_user的数据将是:

[[['C1', 'C2', 'C3'], ['C4', 'C5']], [['C6', 'C7', 'C8'], ['C9']]]

一切都是通过回调(和弦)运行的,因此没有任务等待其他任务。在

解决方案是在这个pull请求中提供的动态任务特性:https://github.com/celery/celery/pull/817。这样,每个任务都可以返回一组子任务,然后这些子任务将替换队列中的原始tak。在

相关问题 更多 >