芹菜功能“reply_to”无法按预期工作

Question

我需要配置 celery 应该将任务执行结果放入哪个队列，我使用文档中描述的方式（项目“reply_to”）：

@app.task(reply_to='export_task')  # <= configured right way
def test_func():
    return "here is result of task"

预期行为

任务结果应位于名称为“export_task”的队列中（如装饰器中配置的）

实际行为

任务结果位于名称如下的队列中：d5587446-0149-3133-a3ed-d9a297d52a96

---

芹菜报告：

python -m celery -A my_worker report

software -> celery:3.1.24 (Cipater) kombu:3.0.37 py:3.5.1
            billiard:3.3.0.23 py-amqp:1.4.9
platform -> system:Windows arch:64bit, WindowsPE imp:CPython
loader   -> celery.loaders.app.AppLoader
settings -> transport:amqp results:rpc:///

CELERY_ACCEPT_CONTENT: ['json']
CELERY_RESULT_BACKEND: 'rpc:///'
CELERY_QUEUES:
    (<unbound Queue main_check -> <unbound Exchange main_check(direct)> -> main_check>,)
CELERYD_CONCURRENCY: 10
CELERY_TASK_SERIALIZER: 'json'
CELERY_RESULT_PERSISTENT: True
CELERY_ROUTES: {
 'my_worker.test_func': {'queue': 'main_check'}}
BROKER_TRANSPORT: 'amqp'
CELERYD_MAX_TASKS_PER_CHILD: 3
CELERY_RESULT_SERIALIZER: 'json'

---

重现步骤

请创建项目文件。

celery_app.py：

from celery import Celery
from kombu import Exchange, Queue

app = Celery('worker')

app.conf.update(
    CELERY_ROUTES={
        'my_worker.test_func': {'queue': 'main_check'},
    },
    BROKER_TRANSPORT='amqp',
    CELERY_RESULT_BACKEND='rpc://',
    CELERY_RESULT_PERSISTENT=True,
    # CELERY_DEFAULT_DELIVERY_MODE='persistent',
    # CELERY_RESULT_EXCHANGE='export_task',
    CELERYD_CONCURRENCY=10,
    CELERYD_MAX_TASKS_PER_CHILD=3,
    CELERY_TASK_SERIALIZER='json',
    CELERY_RESULT_SERIALIZER='json',
    CELERY_ACCEPT_CONTENT=['json'],

    CELERY_QUEUES=(
        Queue('main_check', Exchange('main_check', type='direct'), routing_key='main_check'),
    ),
)

my_worker.py：

from celery_app import app

@app.task(reply_to='export_task')
def test_func():
    return "here is result of task"

然后启动芹菜：

python -m celery -A my_worker worker --loglevel=info

然后在 python 调试控制台中添加新任务：

from my_worker import *
result = test_func.delay()

我请求在官方问题跟踪器上提供帮助，但没人关心。

Answer 1

我在您的代码中没有看到该队列（export_task）已被声明。