我有一个简单的 AMQP/RabbitMQ 异步消费者，使用 Pika 库用 Python 编写，并基于异步消费者示例 http://pika.readthedocs.org/en/latest/examples/asynchronous_consumer_example.html来自皮卡文档。主要区别是我想在线程中运行我的线程，并且希望它正确关闭连接，然后在一定时间间隔后退出（即终止线程）。以下是我打开连接和设置超时的方法。我还打开一个通道，创建一个交换并绑定一个队列......所有这些都工作正常。
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_close() 永远不会被调用。此外，连接未关闭。我可以在 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() 应该可以工作：皮卡文档 http://pika.readthedocs.org/en/latest/modules/adapters/select.html:
close(reply_code=200,reply_text='正常关闭')
断开与 RabbitMQ 的连接。如果有任何打开的通道，它将在完全断开连接之前尝试关闭它们。具有活动消费者的通道将尝试向 RabbitMQ 发送 Basic.Cancel，以在关闭通道之前彻底停止消息的传递。
