通过之前的任务将任务结果芹菜。

Question

这是我的代码。我想通过任务 myName 结果将作为签名中的任务反向作为参数。

这是我的代码。我想通过任务 myName 结果将作为签名中的任务反向作为参数。

from app import app
from app import app
from time import sleep
from celery.utils.log import get_task_logger
import os
from celery import signature, chain, group, chord
from celery.result import allow_join_result


MyQUEUE = os.getenv("SCANS_QUEUE")
logger = get_task_logger(__name__)

@app.task(queue=MyQUEUE, ignore_result=True)
def reverse(text):
    logger.info('reverse order '.format(text))
    return {"reversename": str(text[::-1])}

@app.task(queue=MyQUEUE, ignore_result=True)
def add(a,b):
    logger.info('Addition --> a : {0} & b : {1} '.format(a,b))
    return {"addition": str(a+b)}

@app.task(queue=MyQUEUE, ignore_result=True)
def myname(a):
    logger.info('Name --> a : {0}'.format(a))
    return {"name": str(a)}


@app.task(queue=MyQUEUE, ignore_result=True)
def run_pipeline(a,b,n):
    resultchain = chain([
        group([
            signature(
                add,
                args=(a,b),
                queue=MyQUEUE
            ),
            signature(
                myname,
                args=(n),
                queue=MyQUEUE
            )
        ]),
        signature
        (
            reverse,
            args=(-------),
            queue=MyQUEUE
        )
    ]).apply_async()

    with allow_join_result():
        results = resultchain.join()
    return results

Answer 1

首先也是最重要的是，如果您要使用链，group，starmap或其他类型的任务工作流，则将在需要使用ignore_result = false设置未来或省略参数（默认值为false）。至少在myname和添加现在，需要存储该值，

@app.task(queue=MyQUEUE)
def add(a,b):
    logger.info('Addition --> a : {0} & b : {1} '.format(a,b))
    return {"addition": str(a+b)}

@app.task(queue=MyQUEUE)
def myname(a):
    logger.info('Name --> a : {0}'.format(a))
    return {"name": str(a)}

现在，反向以获取group of Add和myName，您需要调整反向来处理组结果（结果列表）。

对于链，任务的结果将用作下一个任务的第一个参数，在这种情况下，组结果将以[{{{{{ '添加'：...}，{'name'：...}]，您可以访问正确的值。

@app.task(queue=MyQUEUE)
def reverse(group_data):
    # group_data value: [{'addition': '3'}, {'name': 'VALUE'}]
    text = group_data[1]['name']
    logger.info('reverse order '.format(text))
    return {"reversename": str(text[::-1])}

最后，如果您只想扭转myName的结果，则必须仅链接myName和recters。

resultchain = chain([
    signature(myname, args=(n,)),
    signature(reverse)
]).apply_async()