芹菜 - 链条内的组

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。