Celery &Rabbitmq:WARNING/MainProcess] 收到并删除未知消息。错误的目的地？！？- GIT 上的实验

Question

最近，我在做一个GIT项目的实验，以了解大数据处理框架。

1、GIT项目：https://github.com/esperdyne/celery-message-processing

我们有以下组件：

1、AMPQ broker(RabbitMQ)：作为消息缓冲区，作为邮箱，为不同的用户交换消息！

2、worker：作为服务服务器，为各种服务客户端提供服务。 3、Queue("celery"：它作为一个多处理容器，用于同时处理各种worker实例。

关键配置如下：

我们使用对象proj/celery.py来定义app，定义如下：

app = Celery('proj',
         broker='amqp://',
         backend='redis://localhost',
         include=['proj.tasks'])

在此处输入代码

当我们启动应用程序时：

1、启动应用时，我们看到了rabbitmq产生的消息，但是celery无法处理这个消息。

Parse.log 长这样：[2017-02-04 14:28:06,909: WARNING/MainProcess] 收到并删除未知消息。目的地错误？！？

我们有以下问题：

4.2.1 AMQP 机制 我们可以看到AMQP作为消息缓冲区工作，那么就会有一个消息发送者和一个消息获取者：

在上图中，谁是消息发送者，谁是消息获取者。

4.2.2 消息定义 在我们的应用程序中，我们找不到代码来定义要发送或从 AMQP 接收的消息。

4.2.3 消息监控 我们如何监控 AMQP 中的消息发送和接收。 希望老师指导我们解决问题，并给我们一些详细的说明

芹菜经纪人机制介绍！

注意：可以在这里查看错误日志

[2017-02-04 14:28:06,909: WARNING/MainProcess] Received and deleted unknown message.  Wrong destination?!?



 The full contents of the message body was: body: [[u'maildir/allen-       p/inbox/1.'], {}, {u'errbacks': None, u'callbacks': None, u'chord': None, u'chain': [{u'chord_size': None, u'task': u'celery.group', u'args': [], u'immutable': False, u'subtask_type': u'group', u'kwargs': {u'tasks': [{u'chord_size': None, u'task': u'proj.tasks.deploy_db', u'args': [], u'options': {u'reply_to': u'3d9de118-f9d0-3bee-9972-b6a4d4482446', u'task_id': u'3cafda16-3e7c-44db-b05e-1327ef97ffc3'}, u'subtask_type': None, u'kwargs': {}, u'immutable': False}, {u'chord_size': None, u'task': u'proj.tasks.deploy_es', u'args': [], u'options': {u'reply_to': u'3d9de118-f9d0-3bee-9972-b6a4d4482446', u'task_id': u'1f4c728b-680d-4dde-98b9-b153d5282780'}, u'subtask_type': None, u'kwargs': {}, u'immutable': False}]}, u'options': {u'parent_id': None, u'task_id': u'f21c911e-f2ac-462e-9662-2efbd27bcf91', u'root_id': None}}]}] (801b)
{content_type:'application/json' content_encoding:'utf-8'
  delivery_info:{'consumer_tag': 'None4', 'redelivered': False, 'routing_key': 'parse', 'delivery_tag': 623422L, 'exchange': ''} headers={'\xe5\xca.\xdb\x00\x00\x00\x00\x00': None, 'P&5\x07\x00': None, 'T\nKB\x00\x00\x00': 'fc8f0bed-665f-4699-89dd-a56fc247ea8b', 'N\xfd\x17=\x00\x00': 'gen17347@centos1', '\xcfb\xddR': 'py', '9*\xa8': None, '\xb7/b\x84\x00\x00\x00': 0, '\xe0\x0b\xfa\x89\x00\x00\x00': None, '\xdfR\xc4x\x00\x00\x00\x00\x00': [None, None], 'T3\x1d ': 'proj.tasks.parse', '\xae\xbf': 'fc8f0bed-665f-4699-89dd-a56fc247ea8b', '\x11s\x1f\xd8\x00\x00\x00\x00': "('maildir/allen-p/inbox/1.',)", 'UL\xa1\xfc\x00\x00\x00\x00\x00\x00': '{}'}}


[2017-02-04 15:47:22,463: INFO/MainProcess] Connected to amqp://guest:**@localhost:5672//
[2017-02-04 15:47:22,473: INFO/MainProcess] mingle: searching for neighbors
[2017-02-04 15:47:23,503: INFO/MainProcess] mingle: sync with 2 nodes
[2017-02-04 15:47:23,504: INFO/MainProcess] mingle: sync complete
[2017-02-04 15:47:23,530: INFO/MainProcess] parse@centos1 ready.
[2017-02-04 15:47:24,890: INFO/MainProcess] sync with es_deploy@centos1
[2017-02-04 15:47:51,017: WARNING/MainProcess] Received and deleted unknown message.  Wrong destination?!?

The full contents of the message body was: body: [[u'maildir/allen-p/inbox/1.'], {}, {u'errbacks': None, u'callbacks': None, u'chord': None, u'chain': [{u'chord_size': None, u'task': u'celery.group', u'args': [], u'immutable': False, u'subtask_type': u'group', u'kwargs': {u'tasks': [{u'chord_size': None, u'task': u'proj.tasks.deploy_db', u'args': [], u'options': {u'reply_to': u'bd66dd5c-516d-3b51-ab40-c8337a33b18e', u'task_id': u'765e5bbe-198f-405c-b10c-023d35e03981'}, u'subtask_type': None, u'kwargs': {}, u'immutable': False}, {u'chord_size': None, u'task': u'proj.tasks.deploy_es', u'args': [], u'options': {u'reply_to': u'bd66dd5c-516d-3b51-ab40-c8337a33b18e', u'task_id': u'7dacb897-d023-40b5-9874-e00b75107bbd'}, u'subtask_type': None, u'kwargs': {}, u'immutable': False}]}, u'options': {u'parent_id': None, u'task_id': u'f0d41289-33e2-4c8c-8d84-9d1d4c5a9c80', u'root_id': None}}]}] (801b)
{content_type:'application/json' content_encoding:'utf-8'
  delivery_info:{'consumer_tag': 'None4', 'redelivered': False, 'routing_key': 'parse', 'delivery_tag': 3L, 'exchange': ''} headers={'\xe5\xca.\xdb\x00\x00\x00\x00\x00': None, 'P&5\x07\x00': None, 'T\nKB\x00\x00\x00': '4d7754ed-0e36-4731-ae99-a84f42b8eba1', 'N\xfd\x17=\x00\x00': 'gen19722@centos1', '\xcfb\xddR': 'py', '9*\xa8': None, '\xb7/b\x84\x00\x00\x00': 0, '\xe0\x0b\xfa\x89\x00\x00\x00': None, '\xdfR\xc4x\x00\x00\x00\x00\x00': [None, None], 'T3\x1d ': 'proj.tasks.parse', '\xae\xbf': '4d7754ed-0e36-4731-ae99-a84f42b8eba1', '\x11s\x1f\xd8\x00\x00\x00\x00': "('maildir/allen-p/inbox/1.',)", 'UL\xa1\xfc\x00\x00\x00\x00\x00\x00': '{}'}}
 
enter code here

Answer 1

只是为了让答案也位于此处。在线程中 Anis 提到 23doors mentions 表示 Celery 4 的新默认协议无法与 librabbitmq 配合使用：

显然 librabbitmq 问题与 celery 4.x 中的新默认协议有关。

他还提到，要解决此问题，您可以通过设置（如果您使用 Django）来使用 Celery 提供的旧协议：

CELERY_TASK_PROTOCOL = 1 

否则您可以在celeryconf.py 文件中设置以下内容

app.conf.task_protocol = 1

所有功劳归于 23doors :)

Answer 2

提供您正在使用的 celery 和 librabbitmq 的版本会很有帮助。由于我有一个非常相似的问题，我猜你使用的是 celery 4.0.2 和 librabbitmq 1.6.1。

在这种情况下，这是一个已知的兼容性问题，您可以参考https://github.com/celery/celery/issues/3675 和https://github.com/celery/librabbitmq/issues/93。

第一个链接为您提供解决问题的建议：

• 卸载 librabbitmq pip uninstall librabbitmq （你可能要多次调用这个命令）

• 将您的 borker 网址中出现的 amqp 更改为 pyamqp。 （但如果您使用的是配置文件，则不在您的配置文件中。这样做对我不起作用）。

为了更准确地回答您的其他问题：您说得对，有一个发送者和一个提取者。

发件人角色由您调用Celery(...) 时创建的应用承担。它的作用之一是充当任务的注册表，如果你在 app/base.py 中查看它的实现，你会发现它实现了一个方法 send_task，该方法直接由方法 apply_async 调用任务类。此方法的作用是通过线路将您的任务的编组版本发送到代理，以便工作人员可以获取它。用于传输消息的应用程序协议是 amqp，其实现是 librabbitmq。

在线路的另一端，有另一个实例，由执行获取工作的工作人员启动。用芹菜的话来说，它被称为Consumer。你可以在 worker/consumer/consumer.py 中找到它的实现。您将看到它实现了一个create_task_handler，它又定义了on_task_received 函数，这些函数会引发您所看到的错误。它是当从工作人员获取新任务并下一个被处理时调用的函数。

因此，建议的解决方案包括更改 amqp 协议的实现，以便在 on_task_received 中不会引发 TypeError（在我看来，这可能是由编码问题引起的）。

我希望它能回答您的所有问题，并让您更清楚地了解芹菜的工作原理。最后我应该说，据我所知，Celery 的“常规”使用永远不会要求您篡改这些内部结构，并且您可以通过实现自定义任务类和自定义后端来实现 99% 的目标.