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

Question

我正在编写一个链，它根据其内容将一些 JSON 转换为 URI，然后将数据 POST 到该 URI。我想使用 Celery 异步执行此操作，并且知道对链进行分组将使我能够轻松地执行此操作。

我写了以下任务:

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关于避免同步作业。

我正在运行我的 worker

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 是链中的下一个任务，但该任务从未被接受。

为什么我的组在完成链中的第一个任务后挂起？

Answer 1

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

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

res = g.apply_async().get()

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

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

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