从 Java 调度 Celery 任务：“接收并删除未知消息。错误的目的地”答案

【问题标题】：Scheduling Celery task from Java: "Received and deleted unknown message. Wrong destination"从 Java 调度 Celery 任务：“接收并删除未知消息。错误的目的地”
【发布时间】：2016-03-02 16:37:40
【问题描述】：

我正在尝试从 Java 安排 Celery 任务。

我正在像这样向 RabbitMQ 发送任务：

ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
connection = factory.newConnection();
channel = connection.createChannel();

boolean durable = true;
channel.queueDeclarePassive("celery");
channel.exchangeDeclarePassive("celery");

String MESSAGE_FORMAT = "{\"id\": \"%s\", " + "\"task\": \"%s\", " + "\"args\": [\"%s\"]}";
message = String.format(MESSAGE_FORMAT, UUID.randomUUID().toString(), "celery.tasks.add", "2501");
channel.basicPublish("celery", "celery", null, message.getBytes("UTF-8"));

消息已发送：

| routing_key | exchange | message_count |                                                  payload                                                  | payload_bytes | payload_encoding | properties | redelivered |
+-------------+----------+---------------+-----------------------------------------------------------------------------------------------------------+---------------+------------------+------------+-------------+
| celery      |          | 0             | {"id": "7421864e-aff3-4f2f-b274-9d5eacfc8941", "task": "celery.tasks.add", "args": ["2501"]} | 105           | string           |            | False

但是当我开始工作时：

celery -A tasks worker --loglevel=info

我收到一条消息：

[2016-03-02 16:23:08,457: WARNING/MainProcess] Received and deleted unknown message. Wrong destination?!?

The full contents of the message body was: body: '{"id": "76e0fc3b-3ae0-4603-9099-3cd6bb15ffa9", "task": "celery.tasks.add", "args": ["2501"]}' (105b)
{content_type:None content_encoding:None
  delivery_info:{'redelivered': False, 'delivery_tag': 2, 'routing_key': 'celery', 'exchange': '', 'consumer_tag': 'None4'} headers={}}

所以任务没有运行，因为格式有些不对。我跟着： http://docs.celeryproject.org/en/latest/internals/protocol.html

任务在tasks.py 中声明如下：

from celery import Celery

app = Celery('tasks')
app.config_from_object('celeryconfig')


@app.task
def add(materialId):
    f = open('logg','w')
    f.write('processing task: ' + materialId)
    f.close()

所以我不确定应该纠正什么。你能帮我解决这个问题吗？

如果不可能，我将退回到调用 python 来自java的进程，但这会使部署过程复杂化所以我想避免它。

编辑：

为了测试我的方法，我也尝试从 python 以这种方式安排任务，但错误是一样的：

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
channel.queue_declare(queue='celery', durable=True)
channel.basic_publish(exchange='celery', routing_key='celery', body='{"id": "7421864e-aff3-4f2f-b274-9d5eacfc8941", "task": "celery.tasks.add", "args": ["2501"]}')

编辑2：

这不是 Interoperating with Django/Celery From Java 的副本因为答案没有解决我的问题：调整路由密钥和交换没有解决问题。

【问题讨论】：

Interoperating with Django/Celery From Java的可能重复

标签： java json rabbitmq celery

【解决方案1】：

您需要使用 BasicProperties 构建器将消息的内容类型指定为 JSON。

channel.basicPublish("celery", "celery", null, 
    new AMQP.BasicProperties().builder().contentType("application/json").build(),
    message.getBytes("UTF-8"));

关于如何使用它以及 Rabbit 文档中的更多示例：https://www.rabbitmq.com/api-guide.html#publishing

对于 celery 支持的内容类型标头格式：http://docs.celeryproject.org/en/latest/internals/protocol.html#serialization

【讨论】：

【解决方案2】：

您将需要获取从 python 客户端代码发送的原始任务请求，并将其与您在 Java 中生成的内容进行比较。

【讨论】：

谢谢，我还没有考虑过。它告诉我我也应该声明交换（我将编辑我的问题）。但是，即使我完全复制了从 python 客户端发送的消息（我只更改了 UUID），我仍然遇到同样的错误。

【解决方案3】：

或者，您可以尝试使用 celery 源中 examples 目录中的 HTTP gateway 提交任务 - 我不确定代码的这个特定部分是如何维护的。

【讨论】：