【问题标题】:Is it possible to wait until celery group done?是否可以等到芹菜组完成?
【发布时间】:2019-09-01 18:00:33
【问题描述】:

我正在尝试执行组任务并等到所有组子任务完成然后运行最后一个任务。但是当我调用任务时,它调用组和最后一个任务,但最后一个任务在组完成之前完成。是否可以等到组内所有任务完成?

@shared_task(name="print")
def print_order():
    print("PRINT #1")
    mylist = [(1, 2), (4, 6), (1, 4)]
    group([(add.s(*i) | order_id_print.s()) for i in mylist]).delay()


@shared_task(name="print.add")
def add(x,y):
    print("ADD #2")
    chain(add_task1.s(x, y, 'task id') | add_task2.si(x, y, "task_id")).delay()
    return x+y

@shared_task(name="add_task_1")
def add_task1(order_id, ftype, task_id):
    print("ADD task #2-1")
    print("add tasks task1 order_id {} {} {}".format(order_id, ftype, task_id))

@shared_task(name="add_task_2")
def add_task2(order_id, ftype, task_id):
    print("ADD task #2-2")
    print("add tasks task2 order_id {} {} {}".format(order_id, ftype, task_id))


@shared_task(name="print.order_id_print")
def order_id_print(id):
    print("ORDER #3")
    print("order id is {}".format(id))

【问题讨论】:

  • 不久前,我在 FreeNode 的 Celery IRC 频道上问了同样的问题。他们向我解释说,最好的办法是从您的组中创建一个链并添加一个收集数据(或处理组的结果)的任务。

标签: python django rabbitmq celery messagebroker


【解决方案1】:

您可能想要的是一个和弦而不是一个组。 chord 是仅在组中的所有任务执行完毕后才执行的任务。

查看文档:

https://docs.celeryproject.org/en/latest/userguide/canvas.html#chords

【讨论】:

  • 所以在我的情况下,如果我使用和弦,在所有add 任务完成之后,order_id_print 会调用吗?
  • 正确。 chord([add.s(*i) for i in mylist], order_id_print.s()).delay() 之类的东西将异步运行所有 add 任务。当所有add 任务完成后,它将异步调用您的order_id_print 任务。请注意,Celery 会将所有单独的 add 结果作为列表传递给 order_id_print,因此您必须添加一些任务参数。
  • 我照原样做了,但在 celery 中,它在 order_id_print 之后显示 ADD task #2-2。得到这样的芹菜结果 [2019-04-11 17:14:45,475: WARNING/ForkPoolWorker-3] ORDER #3 [2019-04-11 17:14:45,476: WARNING/ForkPoolWorker-3] 订单 ID 是 [5 , 3, 10] [2019-04-11 17:14:45,476: INFO/ForkPoolWorker-3] 任务 print.order_id_print[77b50397-a2bb-4141-8380-e86aa1c04642] 成功在 0.0006398810000973754s: 无 [2019-04-11 17:14:45,476:INFO/MainProcess] 收到的任务:add_task_2[30fbbf70-cae8-482c-81fd-fdc17a816f48]
  • 这是因为您异步触发了add 任务中的另一个任务(链)。如果这不是您想要的,您需要重新安排您的工作流程。
  • 哦,我明白了。因此,我必须在和弦中调用add_task1add_task2 而不是调用add,对吗?这样就不会有任何第三个异步进程???但它需要在 for 循环内,所以在 chord 内应该有组?喜欢chord([(add_task1(*i), add_task2(*i)) for i in my_list], order_id_print.s()).delay()
猜你喜欢
  • 2020-07-09
  • 1970-01-01
  • 2020-05-19
  • 2020-10-11
  • 2020-12-20
  • 1970-01-01
  • 1970-01-01
  • 1970-01-01
  • 2022-01-27
相关资源
最近更新 更多