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

Question

我有一个简单的AMQP / RabbitMQ异步消费者，使用Pika库用Python编写，并基于Pika文档中的异步消费者示例。主要区别在于我想在线程中运行我的线程，我希望它正确关闭连接，然后在一定时间间隔后退出（即终止线程）。以下是我打开连接并设置超时的方法。我还打开一个频道，创建一个交易所并绑定一个队列......一切正常。

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.

“Closing connection（200）:Normal shutdown”来自Pika，但我的on_close或on_cancel回调都没有被调用，无论我是从关闭通道开始，还是只是关闭连接。唯一有效的是使用“basic_cancel”停止消费者，这会导致调用我的“on_cancel_callback”。

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

使用connection.close（）应该有效：来自Pika文档：

关闭（reply_code=200， reply_text='正常关机'）

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

Answer 1

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

常见问题解答的第一部分:

问题：

鼠兔线安全吗？

一个：

Pika在代码中没有任何线程的概念。如果您想将Pika与线程一起使用，请确保在该线程中为每个线程创建了一个Pika连接。跨线程共享一个Pika连接是不安全的。