芹菜 - 在链子里面的小组

Question

我想在链中使用一个组(或块),例如:

chain(getRange.s(3),  GROUP() , xsum.s() )

GROUP()一组double()任务在哪里,即group(double(0),double(1),double(2)).在如何链接一个将列表返回到组中的Celery任务中发布了类似的问题？但是没有解释如何将组中的输出传递给链中的下一个任务.

@task
def getRange(x):
    return range(x)

@task
def double(nr):
    return nr*2

@task
def xsum(list):
    return sum(list)

Answer 1

我不相信有一种方法可以用单链中的当前基元来做到这一点.在你提到的问题中传递回调将不允许你在组任务完成时收听.你能得到的最接近的是:

@task
def get_range(x):
  return range(x)

@task
def mapper(nr):
  return nr * 2

@task
def reducer(nrs):
  return sum(nrs)

@task
def double_then_sum(nrs):
  return (
    group([mapper.s(nr) for nr in nrs]) |
    reducer.s()
  )()

ar = (get_range.s(3) | double_then_sum.s())() # call the procedure
ar.result.result # get the result

否则,您可以尝试使用动态链接,这将导致更简单的解决方案,或者只是map在不需要分组任务并行运行时使用.