由于服务端升级、服务端重启、网络抖动等原因，服务端和客户端的网络连接可能会断开。本文介绍如何在客户端设置Connection和Topology自动恢复，使Connection和Topology在网络连接断开后自动恢复，避免断连对您的业务造成影响。
触发原因
Connection的I/O抛出异常。
Socket读取操作超时。
检测到服务端心跳丢失。
恢复方法
Java
重要
4.0.0及以上版本Java客户端默认开启Connection和Topology自动恢复，您无需在代码中设置。
在客户端开启Connection和Topology（Queue、Exchange、Binding、Consumer）自动恢复的方法如下：
factory.setAutomaticRecoveryEnabled(boolean)：用于开启或关闭Connection自动恢复。factory.setNetworkRecoveryInterval(long)：用于设置重试时间间隔。如果Connection自动恢复异常，设置了Connection自动恢复的客户端将在一段固定时间间隔（默认为5秒）后重试。factory.setTopologyRecoveryEnabled(boolean)：用于开启Topology自动恢复。Topology包括Queue、Exchange、Binding、Consumer。
Python
Pika是开源RabbitMQ官方推荐的Python客户端库。与Java客户端提供的连接自动恢复功能不同，Pika库本身不直接支持通过配置来实现自动的连接恢复。因此，要在Python中实现这一功能，您需要通过编写回调函数来手动处理连接的恢复过程。
示例代码
Java
开启Connection和Topology自动恢复的客户端示例代码如下：
ConnectionFactory factory = new ConnectionFactory();
// 设置接入点，在云消息队列 RabbitMQ 版控制台实例详情页面获取。
factory.setHost("xxx.xxx.aliyuncs.com");
// ${instanceId}为实例ID，从云消息队列 RabbitMQ 版控制台实例详情页面获取。
factory.setCredentialsProvider(new AliyunCredentialsProvider("${instanceId}"));
// 设置Vhost名称，请确保已在消息队列AMQP版控制台上创建。
factory.setVirtualHost("${VhostName}");
// 默认端口，非加密端口5672，加密端口5671。
factory.setPort(5672);
// 基于网络环境设置合理的超时时间。
factory.setConnectionTimeout(30 * 1000);
factory.setHandshakeTimeout(30 * 1000);
factory.setShutdownTimeout(0);
// 开启Connection自动恢复。
factory.setAutomaticRecoveryEnabled(true);
// 设置Connection重试时间间隔为10秒。
factory.setNetworkRecoveryInterval(10000);
// 开启Topology自动恢复。
factory.setTopologyRecoveryEnabled(true);
Connection connection = factory.newConnection(); Python
Python Pika连接自动恢复的消费者客户端示例代码如下：
# -*- coding: utf-8 -*-
import logging
import time
import pika
LOG_FORMAT = ('%(levelname) -10s %(asctime)s %(name) -30s %(funcName) '
'-35s %(lineno) -5d: %(message)s')
LOGGER = logging.getLogger(__name__)
logging.basicConfig(level=logging.INFO, format=LOG_FORMAT)
class Consumer(object):
def __init__(self, amqp_url, queue):
self.should_reconnect = False
self._connection = None
self._channel = None
self._closing = False
self._url = amqp_url
self._queue = queue
def connect(self):
'''
创建connection，并设置以下回调：
on_open_callback: 连接成功回调
on_open_error_callback: 创建连接失败回调
on_close_callback: 连接关闭回调
'''
return pika.SelectConnection(
parameters=pika.URLParameters(self._url),
on_open_callback=self.on_connection_open,
on_open_error_callback=self.on_connection_open_error,
on_close_callback=self.on_connection_closed)
def on_connection_open(self, _unused_connection):
'''
连接成功回调
创建channel，并设置回调：
on_channel_open: channel创建成功回调
'''
self._connection.channel(on_open_callback=self.on_channel_open)
def on_connection_open_error(self, _unused_connection, err):
"""
创建连接失败回调
打印错误信息，并尝试重新连接
"""
LOGGER.error('Connection open failed: %s', err)
self.reconnect()
def on_connection_closed(self, _unused_connection, reason):
"""
连接关闭回调
分以下两种情况：
1. 正常关闭，直接退出
2. 连接异常断开，尝试重新连接
"""
self._channel = None
if self._closing:
self._connection.ioloop.stop()
else:
LOGGER.warning('Connection closed, reconnect necessary: %s', reason)
self.reconnect()
def close_connection(self):
"""
关闭连接
"""
if self._connection.is_closing or self._connection.is_closed:
LOGGER.info('Connection is closing or already closed')
else:
LOGGER.info('Closing connection')
self._connection.close()
def reconnect(self):
"""
修改should_reconnect为True，并停止io_loop
"""
self.should_reconnect = True
self.stop()
def on_channel_open(self, channel):
"""
channel创建成功回调
设置回调：
on_channel_closed: channel关闭回调
开始队列消费
"""
self._channel = channel
self._channel.add_on_close_callback(self.on_channel_closed)
self.start_consuming()
def on_channel_closed(self, channel, reason):
"""
channel关闭回调
打印channel关闭信息，并关闭连接
"""
LOGGER.warning('Channel %i was closed: %s', channel, reason)
self.close_connection()
def start_consuming(self):
"""
开始队列消费
"""
LOGGER.info('start consuming...')
self._channel.basic_consume(
self._queue, self.on_message)
def on_message(self, _unused_channel, basic_deliver, properties, body):
"""
消费消息并上传ack
"""
LOGGER.info('Received message: %s', body.decode())
# 处理业务逻辑
self._channel.basic_ack(basic_deliver.delivery_tag)
def run(self):
"""
创建connection，并启动io_loop
"""
self._connection = self.connect()
self._connection.ioloop.start()
def stop(self):
"""
停止io_loop
"""
if not self._closing:
self._closing = True
self._connection.ioloop.stop()
LOGGER.info('Stopped')
class AutoRecoveryConsumer(object):
def __init__(self, amqp_url, queue):
self._amqp_url = amqp_url
self._queue = queue
self._consumer = Consumer(self._amqp_url, queue)
def run(self):
"""
while True 循环，直到KeyboardInterrupt异常
在run方法中，会启动io_loop监听队列并处理消息，通过循环确保消费者持续运行并且能够自动重连
"""
while True:
try:
self._consumer.run()
except KeyboardInterrupt:
self._consumer.stop()
break
self._maybe_reconnect()
def _maybe_reconnect(self):
"""
判断是否需要重连，每次重连间隔1s
"""
if self._consumer.should_reconnect:
self._consumer.stop()
time.sleep(1)
self._consumer = Consumer(self._amqp_url, self._queue)
def main():
username = 'MjoxODgwNzcwODY5MD****'
password = 'NDAxREVDQzI2MjA0OT****'
host = '1880770****.mq-amqp.cn-hangzhou-a.aliyuncs.com'
port = 5672
vhost = 'vhost_test'
# amqp_url： amqp://<username>:<password>@<host>:<port>/<vhost>
amqp_url = 'amqp://%s:%s@%s:%i/%s' % (username, password, host, port, vhost)
consumer = AutoRecoveryConsumer(amqp_url, 'QueueTest')
consumer.run()
if __name__ == '__main__':
main()恢复限制
Connection断开需要一定的时间检测。要确保这段时间内发送的消息不丢失，需使用Publisher Confirms实现可靠发送。
Channel异常导致Connection断开时，不会触发Connection自动恢复。Channel异常通常为应用级别的问题，需要使用方自行处理。
Connection自动恢复不会使Channel也自动恢复。