为什么我的链返回的参数没有部分应用于下一个任务？

Question

我正在编写一个链，它根据内容将一些JSON转换为URI，然后将数据发布到该URI。我想用芹菜异步地做这件事，并且知道对链进行分组可以让我轻松地做这件事。你知道吗

我写了以下任务：

import time    

from celery import group, chain
from celery.utils.log import get_task_logger

from app import celery

logger = get_task_logger(__name__)

@celery.task
def create_uri(obj_json, endpoint):
    uri = "{0}:{1}/{2}".format(
        obj_json["host"],
        obj_json["port"],
        endpoint
    )
    logger.debug("Created host {0} from {1}".format(uri, obj_json))
    return uri

@celery.task
def send_post(uri, data):
    logger.debug("Posting {0} to {1}...".format(data, uri))
    return uri

def send_messages(objs, endpoint, data):
    chains = [
         # The next line is causing problems.
        (create_uri.s(obj, endpoint) | send_post.s(data))
        for obj in objs
    ]
    g = group(*chains)
    res = g.apply_async(queue="default")
    while not res.ready():
       time.sleep(1)
    uris = res.get()
    print("Posted to {0}".format(uris))
    return uris

然而，我发现，当我尝试使用它时，链的create_uri位结束，但是send_post从未在我的链中被调用。这很奇怪，因为我following the docs有点关于链，实际上，我几乎是在遵循here所示的关于避免同步作业的示例。你知道吗

我在管理我的工人

celery worker -A celery_worker.celery -l debug -c 5 -Q default

其中celery_worker只是推送应用程序上下文并导入app.celery。你知道吗

我的配置如下：

CELERY_BROKER_URL = 'redis://localhost:6379/0'
CELERY_RESULT_BACKEND = 'redis://localhost:6379/0'
CELERY_ACCEPT_CONTENT = ['application/json']
CELERY_TASK_SERIALIZER = 'json'
CELERY_RESULT_SERIALIZER = 'json'

我的日志中的一行是：

[2017-01-09 12:35:59,298: DEBUG/MainProcess] TaskPool: Apply (args:('app.tasks.create_uri', '197f4836-1cd8-4f7f-adaf-b8cebdb304ef', {'timelimit': [None, None], 'group': None, 'parent_id': None, 'retries': 0, 'argsrepr': "({'port': 8079, 'host': 'localhost'}, 'start')", 'lang': 'py', 'eta': None, 'expires': None, 'delivery_info': {'routing_key': 'default', 'priority': 0, 'redelivered': None, 'exchange': ''}, 'kwargsrepr': '{}', 'task': 'app.tasks.create_uri', 'root_id': '197f4836-1cd8-4f7f-adaf-b8cebdb304ef', 'correlation_id': '197f4836-1cd8-4f7f-adaf-b8cebdb304ef', 'origin': 'foobar', 'reply_to': '6559d43e-6cae-3b6f-89be-7b80e2a43098', 'id': '197f4836-1cd8-4f7f-adaf-b8cebdb304ef'}, b'[[{"port": 8079, "host": "localhost"}, "start"], {}, {"chord": null, "callbacks": null, "errbacks": null, "chain": [{"task": "app.tasks.send_post", "subtask_type": null, "options": {"group_id": "60c2c9b2-eb51-457d-b248-b8e5552e0fd8", "task_id":... kwargs:{})

当我打印chains[0].tasks时，我看到：

(app.tasks.create_uri({'host': 'localhost', 'port': 8079}, 'start'), 
 app.tasks.send_post({'hello': 'world'}))

它认识到send_post是链中的下一个任务，但该任务从未被接受。你知道吗

为什么我的group在完成链中的第一个任务后挂起？你知道吗

Answer 1

您正在正确创建链和组。但是，发送到无效队列的任务不会被工作人员识别。当您对它们执行.get()操作时，它们将永远挂起，因为它永远不会返回结果。你知道吗

因此，可以使用默认的celery队列

res = g.apply_async().get()

# explicit
res = g.apply_async(queue="celery").get()

或者正确配置路由，然后使用自定义队列。你知道吗

res = g.apply_async(queue='foo').get()