python - 芹菜 - 链内组

Question

我想在链中使用一个组（或块），例如：

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

GROUP()一组任务在哪里double()，即group(double(0),double(1),double(2))。How to chain a Celery task that return a list into a group?中发布了一个类似的问题？但没有解释如何将输出从组传递到链中的下一个任务。

@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在您不需要分组任务并行运行时使用。