AMQPConnectionError 使用 Pika 和 RabbitMQ 与 Asyncore - 为什么？

Question

为什么我在使用 Asyncore 时会收到 AMQPConnectionError 而在使用 BlockingConnection 时却没有？

如果只是“Asyncore 在 Windows 中不起作用”，那就这样吧，尽管我还没有找到任何禁止使用它的东西。（这个问题与平台无关。）为了方便我想使用 Python 2.7 和 Python 3.4 上都可用的异步库，并且 Asyncore 应该在这里工作。

我将 RabbitMQ 3.2.4 与 Python 2.7.6 和 pika 0.9.13 一起使用。用户和管理员运行级别没有区别。除了上面更新的警告消息外，代码中记录器的存在与否与错误无关。在 Linux (Ubuntu 14.04) 和 Windows 7 中也出现了相同的错误，因此它不是平台问题。

由于 pika 使用 BlockingConnection 的性能相当差，我想尝试使用 Asyncore 适配器。对于测试台设置来说似乎很简单（我尝试给它提供凭据，尽管这无关紧要，如果没有给出回调，则回调将被剔除......无论哪种方式都失败了。）：​​

按照教程使用 BlockingConnection - 它可以工作，但吞吐量较低：

connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))

使用 AsyncoreConnection - 我尝试过的所有变体都会立即失败：

connection = pika.AsyncoreConnection(pika.ConnectionParameters(host='localhost'))

错误：

WARNING:pika.connection:Could not connect, 0 attempts left
Traceback (most recent call last):
  File "C:\workspace\send.py", line 8, in <module>
    connection = pika.AsyncoreConnection(pika.ConnectionParameters(host='localhost'))
  File "C:\Python27\lib\site-packages\pika\adapters\asyncore_connection.py", line 135, in __init__
    stop_ioloop_on_close)
  File "C:\Python27\lib\site-packages\pika\adapters\base_connection.py", line 62, in __init__
    on_close_callback)
  File "C:\Python27\lib\site-packages\pika\connection.py", line 590, in __init__
    self.connect()
  File "C:\Python27\lib\site-packages\pika\connection.py", line 707, in connect
    self.callbacks.process(0, self.ON_CONNECTION_ERROR, self, self)
  File "C:\Python27\lib\site-packages\pika\callback.py", line 61, in wrapper
    return function(*tuple(args), **kwargs)
  File "C:\Python27\lib\site-packages\pika\callback.py", line 92, in wrapper
    return function(*args, **kwargs)
  File "C:\Python27\lib\site-packages\pika\callback.py", line 232, in process
    callback(*args, **keywords)
  File "C:\Python27\lib\site-packages\pika\connection.py", line 1192, in _on_connection_error
    raise exceptions.AMQPConnectionError(self.params.connection_attempts)
pika.exceptions.AMQPConnectionError: 1

Answer 1

尝试下面提到的步骤。我在我的 centos 机器上遇到了同样的问题。

1. sudo yum install rabbitmq-server
2. sudo service rabbitmq-server 重启

Answer 2

在我看来，它实际上就像鼠兔身上的一个虫子。下面是最终引发异常的 connection.connect() 代码：

def connect(self):
    """Invoke if trying to reconnect to a RabbitMQ server. Constructing the
    Connection object should connect on its own.

    """
    self._set_connection_state(self.CONNECTION_INIT)
    if self._adapter_connect():
        return self._on_connected()
    self.remaining_connection_attempts -= 1
    LOGGER.warning('Could not connect, %i attempts left',
                   self.remaining_connection_attempts)
    if self.remaining_connection_attempts:
        LOGGER.info('Retrying in %i seconds', self.params.retry_delay)
        self.add_timeout(self.params.retry_delay, self.connect)
    else:
        self.callbacks.process(0, self.ON_CONNECTION_ERROR, self, self)
        self.remaining_connection_attempts = self.params.connection_attempts
        self._set_connection_state(self.CONNECTION_CLOSED)

所以，self._adapter_connect() 显然没有返回 True，这表明连接失败。这是AsyncoreConnection._adapter_connect 代码：

def _adapter_connect(self):
    """Connect to our RabbitMQ broker using AsyncoreDispatcher, then setting
    Pika's suggested buffer size for socket reading and writing. We pass
    the handle to self so that the AsyncoreDispatcher object can call back
    into our various state methods.

    """
    if super(AsyncoreConnection, self)._adapter_connect():
        self.socket = PikaDispatcher(self.socket, None, self._handle_events)
        self.ioloop = self.socket
        self._on_connected()

它不返回任何东西！所以connect 中的 if 语句永远不会为真。如果我更改方法以反映所有其他适配器使用的模式：

def _adapter_connect(self):
    """Connect to our RabbitMQ broker using AsyncoreDispatcher, then setting
    Pika's suggested buffer size for socket reading and writing. We pass
    the handle to self so that the AsyncoreDispatcher object can call back
    into our various state methods.

    """
    if super(AsyncoreConnection, self)._adapter_connect():
        self.socket = PikaDispatcher(self.socket, None, self._handle_events)
        self.ioloop = self.socket
        return True
    return False

它工作正常。我肯定会提交那个错误！

编辑：

该错误似乎已在最新版本中得到修复（来自github）：

    def _adapter_connect(self):
        """Connect to our RabbitMQ broker using AsyncoreDispatcher, then setting Pika's suggested buffer size for socket reading and writing. We pass the handle to self so that the AsyncoreDispatcher object can call back into our various state methods.

        """
        error = super(AsyncoreConnection, self)._adapter_connect()
        if not error:
            self.socket = PikaDispatcher(self.socket, None,
                                         self._handle_events)
            self.ioloop = self.socket
            self._on_connected()
        return error

Answer 3

阅读这篇文章：No handlers could be found for logger "pika.adapters.blocking_connection"

通过添加修复：

import logging
logging.basicConfig()

编辑

问题已报告https://github.com/pika/pika/issues/468

Answer 4

你必须通过运行下面的命令来启动rabbitmq-server，然后它就会工作

sudo systemctl start rabbitmq-server