最近线上服务出现rabbitMq队列不消费的情况，我们最终定位到可能是rabbitMq服务异常，而其他服务没有建立重连机制导致的。
首先我们需要了解RabbitMq，RabbitMq 是实现了高级消息队列协议（AMQP）的开源消息代理中间件。消息队列是一种应用程序对应用程序的通行方式，应用程序通过写消息，将消息传递于队列，由另一应用程序读取完成通信。而作为中间件的 RabbitMq 无疑是目前最流行的消息队列之一。
RabbitMq 应用场景广泛：
--系统的高可用：日常生活当中各种商城秒杀，高流量，高并发的场景。当服务器接收到如此大量请求处理业务时，有宕机的风险。某些业务可能极其复杂，但这部分不是高时效性，不需要立即反馈给用户，我们可以将这部分处理请求抛给队列，让程序后置去处理，减轻服务器在高并发场景下的压力。
--分布式系统，集成系统，子系统之间的对接，以及架构设计中常常需要考虑消息队列的应用。
python连接操作rabbitMQ 主要是使用pika库。
pika连接rabbitMQ的连接参数主要是在使用ConnectionParameters和URLParameters。
URLParameters可官方自行了解，这里主要简述下ConnectionParameters。
connectionParameters定义简化为：
class ConnectionParameters(Parameters):
def __init__(self,
host=_DEFAULT,
port=_DEFAULT,
virtual_host=_DEFAULT,
credentials=_DEFAULT,
channel_max=_DEFAULT,
frame_max=_DEFAULT,
heartbeat=_DEFAULT,
ssl_options=_DEFAULT,
connection_attempts=_DEFAULT,
retry_delay=_DEFAULT,
socket_timeout=_DEFAULT,
stack_timeout=_DEFAULT,
locale=_DEFAULT,
blocked_connection_timeout=_DEFAULT,
client_properties=_DEFAULT,
tcp_options=_DEFAULT,
**kwargs)
参数默认值都是一个_DEFAULT的类， 这个将映射对应的默认值到对应的参数
参数说明：
1.host
DEFAULT_HOST = 'localhost'
2.port
DEFAULT_PORT = 5672
3.virtual_host
DEFAULT_VIRTUAL_HOST = ‘/’
4.credentials认证参数：默认值：DEFAULT_CREDENTIALS = pika.credentials.PlainCredentials(DEFAULT_USERNAME, DEFAULT_PASSWORD)
DEFAULT_USERNAME = 'guest'
DEFAULT_PASSWORD = 'guest'
5.channel_max最大通道数
DEFAULT_CHANNEL_MAX = pika.channel.MAX_CHANNELS
6.frame_max要使用的所需最大AMQP帧大小
DEFAULT_FRAME_MAX = spec.FRAME_MAX_SIZE
7.heartbeat心跳， 0 为关闭。连接调整期间协商的AMQP连接心跳超时值或连接调整期间调用的可调用值
DEFAULT_HEARTBEAT_TIMEOUT = None # None accepts server’s proposal
8.ssl_options传入值pika.SSLOptions
DEFAULT_SSL_OPTIONS = None
9.connection_attempts套接字连接尝试次数
DEFAULT_CONNECTION_ATTEMPTS = 1
10.retry_delay套接字连接尝试重连间隔
DEFAULT_RETRY_DELAY = 2.0
11.socket_timeout
DEFAULT_SOCKET_TIMEOUT = 10.0 # socket.connect() timeout
12.stack_timeout套接字连接尝试间隔 ， None为禁用
DEFAULT_STACK_TIMEOUT = 15.0 # full-stack TCP/[SSl]/AMQP bring-up timeout
13.locale
DEFAULT_LOCALE = ‘en_US’
14.blocked_connection_timeout阻塞的超时时间，默认不超时
DEFAULT_BLOCKED_CONNECTION_TIMEOUT = None
15.client_properties客户端属性，用于覆盖通过Connection.StartOk 方法向RabbitMQ报告的默认客户端属性中的字段，字典类型/None
DEFAULT_CLIENT_PROPERTIES = None
16.tcp_options
DEFAULT_TCP_OPTIONS = None
其他：
DEFAULT_SSL = False
DEFAULT_SSL_PORT = 5671
回到这个话题的出发点，当时排除代码发现，python使用pika连接rabbitMq并没有做任何处理，导致rabbitMq异常，相应的python服务会直接假死或者直接挂掉，这就导致了rabbitMQ队列不消费。
#无重试配置
@property
def connection(self):
if self._connection is not None:
return self._connection
credentials = pika.credentials.PlainCredentials('root','root')
connectionParameters = pika.ConnectionParameters(
host='localhost',
virtual_host=5672,
credentials=credentials,
socket_timeout=10
)
self._connection = pika.BlockingConnection(connectionParameters)
return self._connection
#增加重试配置及设置心跳为0(不主动关闭连接)
@property
def connection(self):
if self._connection is not None:
return self._connection
credentials = pika.credentials.PlainCredentials('root','root')
connectionParameters = pika.ConnectionParameters(
host='localhost',
virtual_host=5672,
credentials=credentials,
socket_timeout=10,
heartbeat=0,
retry_delay=10,
connection_attempts=10
)
self._connection = pika.BlockingConnection(connectionParameters)
return self._connection
最后对于Bunny, Java, .NET, Objective-C, Swift 的rabbitmq客户端拥有自动重连机制， 但是对于python 客户端 目前还没有提供自动重连机制，这就需要自行实现。
1.while实现：
import pika
while True:
try:
connection = pika.BlockingConnection()
channel = connection.channel()
channel.basic_consume('test', on_message_callback)
channel.start_consuming()
# Don't recover if connection was closed by broker
except pika.exceptions.ConnectionClosedByBroker:
break
# Don't recover on channel errors
except pika.exceptions.AMQPChannelError:
break
# Recover on all other connection errors
except pika.exceptions.AMQPConnectionError:
continue
这种方式简单，但不够优雅， 因为异常后，会不停地进行重试。
2.retry实现
from retry import retry
@retry(pika.exceptions.AMQPConnectionError, delay=5, jitter=(1, 3))
def consume(self, callback):
"""Start consuming AMQP messages in the current process"""
if self.connection.is_closed:
self.reconnect()
try:
channel, queue = self.setListener(callback)
channel.start_consuming()
except (KeyboardInterrupt, SystemExit):
raise
except:
log.exception('Failed to prepare AMQP consumer')
raise
#当时忘记发布也进行连接重连设置了，导致问题没有彻底解决。
@retry(pika.exceptions.AMQPConnectionError, delay=5, jitter=(1, 3))
def publish(self, body, block=True, timeout=None, properties=None):
retry是一个用于错误处理的模块，功能类似try-except，但更加快捷方便，这里就将简单地介绍一下retry的基本用法。
retry-作为装饰器进行使用，不传入参数时功能如下例所示：
from retry import retry
@retry()
def make_trouble():
'''Retry until succeed'''
print ('retrying...')
raise
if __name__ == '__main__':
make_trouble()
# 输出: 一直重试，直到运行成功
retrying...
retrying...
retrying...
retrying...
retrying...
retrying...
retry 参数介绍
def retry(exceptions=Exception, tries=-1, delay=0, max_delay=None, backoff=1, jitter=0, logger=logging_logger):
"""Return a retry decorator.
:param exceptions: an exception or a tuple of exceptions to catch. default: Exception.
:param tries: the maximum number of attempts. default: -1 (infinite).
:param delay: initial delay between attempts. default: 0.
:param max_delay: the maximum value of delay. default: None (no limit).
:param backoff: multiplier applied to delay between attempts. default: 1 (no backoff).
:param jitter: extra seconds added to delay between attempts. default: 0.
fixed if a number, random if a range tuple (min, max)
:param logger: logger.warning(fmt, error, delay) will be called on failed attempts.
default: retry.logging_logger. if None, logging is disabled.
"""
retry()在这里的功能，是在其装饰的函数运行报错后重新运行该函数，在上例中的效果就是反复运行make_trouble()，这也是retry()的基本用法，下面介绍其几个主要参数：
exceptions：传入指定的错误类型，默认为Exception，即捕获所有类型的错误，也可传入元组形式的多种指定错误类型。
tries：定义捕获错误之后重复运行次数，默认为-1，即为无数次。
delay：定义每次重复运行之间的停顿时长，单位秒，默认为0，即无停顿。
backoff：呈指数增长的每次重复运行之间的停顿时长，需要配合delay来使用，譬如delay设置为3，backoff设置为2，则第一次间隔为3*2^0=3秒，第二次3*2^1=6秒，第三次3*2^2=12秒，以此类推，默认为1。
max_delay：定义backoff和delay配合下出现的等待时间上限，当delay*backoff**n大于max_delay时，等待间隔固定为该值而不再增长。
网友评论