python - Pika SelectConnection 适配器的 close() 方法不会关闭连接

Question

我有一个用于 AMQP/RabbitMQ 的简单异步使用者，使用 Pika 库用 Python 编写并基于 Asynchronous consumer example来自鼠兔文档。主要区别在于我想在一个线程中运行我的，我希望它正确关闭连接然后在一定时间间隔后退出(即终止线程)。这是我打开连接和设置超时的方法。我还打开了一个 channel ，创建了一个交换器并绑定(bind)了一个队列……一切正常。

def connect(self):
  LOGGER.info('OPEN connection...')
  return pika.SelectConnection(self._parameters, self.on_connection_open, stop_ioloop_on_close=False)

def on_connection_open(self, unused_connection):
  LOGGER.info('Connection opened')
  self.add_on_connection_close_callback()
  self._connection.add_timeout(5, self.timer_tick)
  self.open_recv_channel()

这是超时回调:

def timer_tick(self):
  LOGGER.info('---TICK---')
  self._stop()

这是 _stop 方法:

def _stop(self):
  LOGGER.info('Stopping...')
  self._connection.close()
  LOGGER.info('Stopped')
  time.sleep(5)
  self._connection.ioloop.stop()

这是启动线程的 run 方法:

def run(self):
  print "-Run Started-"
  self._connection = self.connect()
  self._connection.ioloop.start()
  print "-Run Finished-"

这是 main() 的主要部分:

client = TestClient()
client.start()
client.join()
LOGGER.info('Returned.')
time.sleep(30)

我的问题是“self._connection.close()”无法正常工作。我添加了一个 on_close 回调:

self._connection.add_on_close_callback(self.on_connection_closed)

但是永远不会调用 on_connection_closed()。此外，连接未关闭。我可以在 RabbitMQ 管理 Web 界面中看到它，它甚至在线程结束后仍然存在。这是输出:

-Run Started-
2015-01-28 14:39:28,431: OPEN connection...
2015-01-28 14:39:28,491: Queue bound
(...[snipped] various other messages here...)
2015-01-28 14:39:28,491: Issuing consumer related RPC commands
2015-01-28 14:39:28,491: Adding consumer cancellation callback
(Pause here waiting for timeout callback)
2015-01-28 14:39:33,505: ---TICK---
2015-01-28 14:39:33,505: Stopping...
2015-01-28 14:39:33,505: Closing connection (200): Normal shutdown
2015-01-28 14:39:33,505: Stopped
-Run Finished-
2015-01-28 14:39:39,507: Returned.

“关闭连接(200):正常关闭”来自 Pika，但我的 on_close 或 on_cancel 回调都没有被调用，无论我是从关闭 channel 开始，还是只是关闭连接。唯一可行的是使用“basic_cancel”停止消费者，这会导致调用我的“on_cancel_callback”。

我想在主程序中使用一个循环来创建和销毁消费者线程，但目前，每次我运行一个线程时，我都会留下一个孤立的连接，因此我的连接数会无限增加。当程序关闭时，连接会消失。

使用 connection.close() 应该有效:来自 Pika Docs :

close(reply_code=200, reply_text='正常关机')

断开与 RabbitMQ 的连接。如果有任何打开的 channel ，它会在完全断开连接之前尝试关闭它们。具有活跃消费者的 channel 将尝试向 RabbitMQ 发送 Basic.Cancel 以在关闭 channel 之前彻底停止消息传递。

Answer 1

如果您在线程之间共享连接，这可能会导致问题。 pika 不是线程安全的，不同线程不应使用连接。

First bit常见问题解答:

问:

Is Pika thread safe?

答:

Pika does not have any notion of threading in the code. If you want to use Pika with threading, make sure you have a Pika connection per thread, created in that thread. It is not safe to share one Pika connection across threads.