Rabbitmq heartbea心跳检测机制原理解析
前言
使用rabbitmq的时候，当你客户端与rabbitmq服务器之间一段时间没有流量，服务器将会断开与客户端之间tcp连接。
而你将在服务器上看这样的日志：
missedheartbeatsfromclient,timeout:xxs
这个间隔时间就是心跳间隔。
heartbeat通常用来检测通信的对端是否存活（未正常关闭socket连接而异常crash）。其基本原理是检测对应的socket连接上数据的收发是否正常，如果一段时间内没有收发数据，则向对端发送一个心跳检测包，如果一段时间内没有回应则认为心跳超时，即认为对端可能异常crash了。
rabbitmq也不例外，heatbeat在客户端和服务端之间用于检测对端是否正常，即客户端与服务端之间的tcp链接是否正常。
关于rabbitmq心跳
1.heartbeat检测时间间隔可在配置文件rabbitmq.config中增加配置项{heartbeat,Timeout}进行配置，其中Timeout指定时间间隔，单位为秒，另外客户端也可以配置heartbeat时间。
如果服务端没有配置
默认代理心跳时间：
RabbitMQ3.2.2：580秒
RabbitMQ3.5.5：60秒
2.官方建议不要禁用心跳，且建议心跳时间为60秒。
3.心跳每heartbeattimeout/2秒发送一次，服务器两次没有接收到则断开tcp连接，以前的连接将失效，客户端需要重新连接。
4.如果你使用Java,.NETandErlangclients，服务器与客户端会协商heartbeat时间
如果其中一个值为0，则使用两者中较大的一个
否则，使用两者中较小的一个
两个值都为0，则表示要禁用心跳，则服务端与客户端维持此tcp连接，不会断开。
注意：在python客户端上直接设置为0，则禁用心跳。
禁用心跳在python客户端该如何设置：
在py3:ConnectionParameters设置heartbeat_interval=0即可。
在py2:ConnectionParameters设置heartbeat=0即可。
5.连接上的任何流量（传输的有效数据、确认等）都将被计入有效心跳，当然也包括心跳帧。
6.我在网上看到有人问到这个问题：
为什么服务端宕机，在心跳检测机制下，服务器侧断开连接，而客户端这边不能检测到tcp断开，我测试过，客户端确实不能检测到tcp连接断开，只有当客户端在这个tcp有操作后，才能检测到，当然在一个断开的tcp连接上做操作会报错（如发送消息）。
importpika
importtime
credit=pika.PlainCredentials(username='cloud',password='cloud')
connection=pika.BlockingConnection(pika.ConnectionParameters(
host='10.32.1.12',credentials=credit))
channel=connection.channel()
whileTrue:
connect_close=connection.is_closed
connect_open=connection.is_open
channel_close=channel.is_closed
channel_open=channel.is_open
print("connectionis_closed",connect_close)
print("connectionis_open",connect_open)
print("channelis_closed",channel_close)
print("channelis_open",channel_open)
print("")
time.sleep(5)
7.一些RabbitMQ客户端（Bunny，Java，.NET，Objective-C，Swift）提供了一种在网络故障后自动恢复连接的机制，而pika只能通过检测连接异常后再重新创建连接的方式。
示例代码：通过检测连接异常，重新创建连接：
importpika
whileTrue:
try:
connection=pika.BlockingConnection()
channel=connection.channel()
channel.basic_consume('test',on_message_callback)
channel.start_consuming()
#Don'trecoverifconnectionwasclosedbybroker
exceptpika.exceptions.ConnectionClosedByBroker:
break
#Don'trecoveronchannelerrors
exceptpika.exceptions.AMQPChannelError:
break
#Recoveronallotherconnectionerrors
exceptpika.exceptions.AMQPConnectionError:
continue
你也可以使用操作重试库，例如retry。
fromretryimportretry
@retry(pika.exceptions.AMQPConnectionError,delay=5,jitter=(1,3))
defconsume():
connection=pika.BlockingConnection()
channel=connection.channel()
channel.basic_consume('test',on_message_callback)
try:
channel.start_consuming()
#Don'trecoverconnectionsclosedbyserver
exceptpika.exceptions.ConnectionClosedByBroker:
pass
consume()
heartbeat的实现
rabbitmq在收到来自客户端的connection.tune-ok信令后，启用心跳检测，rabbitmq会为每个tcp连接创建两个进程用于心跳检测，一个进程定时检测tcp连接上是否有数据发送（这里的发送是指rabbitmq发送数据给客户端），如果一段时间内没有数据发送给客户端，则发送一个心跳包给客户端，然后循环进行下一次检测；另一个进程定时检测tcp连接上是否有数据的接收，如果一段时间内没有收到任何数据，则判定为心跳超时，最终会关闭tcp连接。另外，rabbitmq的流量控制机制可能会暂停heartbeat检测，这里不展开描述。
涉及的源码：
start(SupPid,Sock,SendTimeoutSec,
SendFun,ReceiveTimeoutSec,ReceiveFun)->
%%数据发送检测进程
{ok,Sender}=start_heartbeater(SendTimeoutSec,SupPid,Sock,
SendFun,heartbeat_sender,
start_heartbeat_sender),
%%数据接收检测进程
{ok,Receiver}=start_heartbeater(ReceiveTimeoutSec,SupPid,
Sock,ReceiveFun,
heartbeat_receiver,
start_heartbeat_receiver),
{Sender,Receiver}.
start_heartbeat_sender(Sock,TimeoutSec,SendFun)->
%%the'div2'istheresothatwedon'tendupwaitingfor
%%nearly2*TimeoutSecbeforesendingaheartbeatinthe
%%boundarycase
heartbeater({Sock,TimeoutSec*1000div2,send_oct,0,
fun()->SendFun(),continueend}).
start_heartbeat_receiver(Sock,TimeoutSec,ReceiveFun)->
%%wecheckforincomingdataeveryinterval,andtimeoutafter
%%twocheckswithnochange.Asaresultwewilltimeout
%%between2and3intervalsafterthelastdatahasbeen
%%received
heartbeater({Sock,TimeoutSec*1000,recv_oct,1,
fun()->ReceiveFun(),stopend}).
heartbeater({Sock,TimeoutMillisec,
StatName,Threshold,Handler}=Params,
Deb,
{StatVal,SameCount}=State)->
Recurse=fun(State1)->heartbeater(Params,Deb,State1)end,
receive
...
%%定时检测
afterTimeoutMillisec->
caserabbit_net:getstat(Sock,[StatName])of
{ok,[{StatName,NewStatVal}]}->
%%收发数据有变化
ifNewStatVal=/=StatVal->
%%重新开始检测
Recurse({NewStatVal,0});
%%未达到指定次数,发送为0,接收为1
SameCount
%%计数加1,再次检测
Recurse({NewStatVal,SameCount+1});
%%heartbeat超时
true->
%%对于发送检测超时,向客户端发送heartbeat包
%%对于接收检测超时,向父进程发送超时通知
%%由父进程触发tcp关闭等操作
caseHandler()of
%%接收检测超时
stop->ok;
%%发送检测超时
continue->Recurse({NewStatVal,0})
end;
...
收发检测的时候利用了inet模块的getstat，查看socket的统计信息
recv_oct:查看socket上接收的字节数
send_oct:查看socket上发送的字节数
inet详细见这里: http://www.erlang.org/doc/man/inet.html
以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持毛票票。