【发布时间】：2015-01-28 10:28:19
【问题描述】：
我有一个简单的 AMQP/RabbitMQ 异步使用者，使用 Pika 库用 Python 编写，并基于 Pika 文档中的 Asynchronous consumer example。主要区别是我想在一个线程中运行我的，我希望它正确关闭连接，然后在一定时间间隔后退出（即终止线程）。这是我打开连接并设置超时的方法。我还打开了一个频道，创建了一个交换并绑定了一个队列……一切都很好。
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 回调都没有被调用，无论我是从关闭通道开始，还是只是关闭连接。唯一有效的是用“basic_cancel”停止消费者，这会导致我的“on_cancel_callback”被调用。
我想在主程序中使用一个循环来创建和销毁消费者线程，但是目前，每次我运行一个循环时，我都会留下一个孤立的连接，因此我的连接数会无限增加。当程序关闭时，连接会消失。
使用 connection.close() 应该可以工作：来自Pika Docs：
close(reply_code=200, reply_text='正常关机')
断开与 RabbitMQ 的连接。如果有任何打开的通道，它将在完全断开之前尝试关闭它们。拥有活跃消费者的通道将尝试在关闭通道之前向 RabbitMQ 发送 Basic.Cancel 以彻底停止消息的传递。
【问题讨论】：
-
pika 不是线程安全的，不同线程不应使用连接。
-
悲哀！可能就是这样。我在文档中没有看到任何关于线程的信息，但是当我搜索时，我发现它是 first thing in the FAQ 也许您应该发表您的评论作为答案？
-
我做了，但不确定您是否正在共享连接。从您的代码中看不出来。
-
这还没有修复。我的代码确实执行了“self._channel.close()”方法，但在所有情况下，鼠兔都不会关闭连接。有时连接会保持活动状态。我也使用了“add_on_close_callback”机制，有时驱动程序不会执行它。这样 pika 连接线程永远保持活动状态，python 无法定期退出程序。我的代码中的线程之间没有连接共享。