当前访客身份:游客 [ 登录  | 注册加入尚学堂]
直播

我来了!

拥有积分:3958
尚学堂雄起!!威武。。。

博客分类

笔记中心

课题中心

提问中心

答题中心

解答题中心

【原创】RabbitMQ 之 no_ack 分析

我来了! 发表于 2年前 (2014-11-07 16:19:54)  |  评论(0)  |  阅读次数(456)| 0 人收藏此文章,   我要收藏   

      熟悉 RabbitMQ 的人肯定知道  no_ack 属性是在调用 Basic.Consume 方法时可以设置的一个重要参数。本文主要针对 no_ack 设置的两种情况,通过抓包分析的形式讲解下实际应用中的异同,并总结一下相关的处理经验。

============ 我是分隔线 =============

      no_ack 的用途:确保 message 被 consumer “成功”处理了。这里“成功”的意思是,(在设置了 no_ack=false 的情况下)只要 consumer 手动应答了 Basic.Ack ,就算其“成功”处理了。

情况一no_ack=true (此时为自动应答
      在这种情况下,consumer 会在接收到 Basic.Deliver + Content-Header + Content-Body 之后,立即回复 Ack 。而这个 Ack 是 TCP 协议中的 Ack 。此 Ack 的回复不关心 consumer 是否对接收到的数据进行了处理,当然也不关心处理数据所需要的耗时。

图1:(Producer+Consumer)


图2:(Consumer)


图3:(Producer)


情况二:no_ack=false (此时为手动应答
      在这种情况下,要求 consumer 在处理完接收到的 Basic.Deliver + Content-Header + Content-Body 之后才回复 Ack 。而这个 Ack 是 AMQP 协议中的 Basic.Ack 。此 Ack 的回复是和业务处理相关的,所以具体的回复时间应该要取决于业务处理的耗时。

图4:(Producer+Consumer)


图5:(Consumer)



图6:(Producer)


总结:
  • Basic.Ack 发回给 RabbitMQ 以告知,可以将相应 message 从 RabbitMQ 的消息缓存中移除。
  • Basic.Ack 未被 consumer 发回给 RabbitMQ 前出现了异常,RabbitMQ 发现与该 consumer 对应的连接被断开,之后将该 message 以轮询方式发送给其他 consumer (假设存在多个 consumer 订阅同一个 queue)。
  • 在 no_ack=true 的情况下,RabbitMQ 认为 message 一旦被 deliver 出去了,就已被确认了,所以会立即将缓存中的 message 删除。所以在 consumer 异常时会导致消息丢失。
  • 来自 consumer 侧的 Basic.Ack 与 发送给 Producer 侧的 Basic.Ack 没有直接关系。


============ 我是分隔线 =============

最后贴上自己改造的、基于 libevent 实现的 rabbitmq-c 的测试打印。

情况一:
[warn] evsignal_init: socketpair: No error
drive_machine: [conn_init]  ---  in TCP 3-way connecting!
drive_machine: [conn_connecting]  ---  connection timeout 1 time on socket(6040)
drive_machine: [conn_connected]  ---  connected on socket(6040)

6040: conn_state change   connected ==> snd_protocol_header
  --> Send Protocol.Header!
6040: conn_state change   snd_protocol_header ==> rcv_connection_start_method
  <-- Recv Connection.Start Method frame!
6040: conn_state change   rcv_connection_start_method ==> snd_connection_start_rsp_method
  --> Send Connection.Start-Ok Method frame!
6040: conn_state change   snd_connection_start_rsp_method ==> rcv_connection_tune_method
  <-- Recv Connection.Tune Method frame!
6040: conn_state change   rcv_connection_tune_method ==> snd_connection_tune_rsp_method
  --> Send Connection.Tune-Ok Method frame!
6040: conn_state change   snd_connection_tune_rsp_method ==> snd_connection_open_method
  --> Send Connection.Open Method frame!
6040: conn_state change   snd_connection_open_method ==> rcv_connection_open_rsp_method
  <-- Recv Connection.Open-Ok Method frame!
6040: conn_state change   rcv_connection_open_rsp_method ==> snd_channel_open_method
  --> Send Channel.Open Method frame!
6040: conn_state change   snd_channel_open_method ==> rcv_channel_open_rsp_method
  <-- Recv Channel.Open-Ok Method frame!
6040: conn_state change   rcv_channel_open_rsp_method ==> idle

drive_machine: [conn_idle]  ---  [CONSUMER]: Queue Declaring!
6040: conn_state change   idle ==> snd_queue_declare_method
  --> Send Queue.Declare Method frame!
6040: conn_state change   snd_queue_declare_method ==> rcv_queue_declare_rsp_method
  <-- Recv Queue.Declare-Ok Method frame!
6040: conn_state change   rcv_queue_declare_rsp_method ==> idle

drive_machine: [conn_idle]  ---  [CONSUMER]: Queue Binding!
6040: conn_state change   idle ==> snd_queue_bind_method
  --> Send Queue.Bind Method frame!
6040: conn_state change   snd_queue_bind_method ==> rcv_queue_bind_rsp_method
  <-- Recv Queue.Bind-Ok Method frame!
6040: conn_state change   rcv_queue_bind_rsp_method ==> idle

drive_machine: [conn_idle]  ---  [CONSUMER]: Basic QoS!
6040: conn_state change   idle ==> snd_basic_qos_method
  --> Send Basic.Qos Method frame!
6040: conn_state change   snd_basic_qos_method ==> rcv_basic_qos_rsp_method
  <-- Recv Basic.Qos-Ok Method frame!
6040: conn_state change   rcv_basic_qos_rsp_method ==> idle

drive_machine: [conn_idle]  ---  [CONSUMER]: Basic Consuming!
6040: conn_state change   idle ==> snd_basic_consume_method
  --> Send Basic.Consume Method frame!
6040: conn_state change   snd_basic_consume_method ==> rcv_basic_consume_rsp_method
  <-- Recv Basic.Consume-Ok Method frame!
6040: conn_state change   rcv_basic_consume_rsp_method ==> idle

drive_machine: [conn_idle]  ---  [CONSUMER]: Start waiting to recv!
6040: conn_state change   idle ==> rcv_basic_deliver_method
drive_machine: wait for Basic.Deliver method another 10 seconds!!
drive_machine: wait for Basic.Deliver method another 10 seconds!!

  <-- Recv Basic.Deliver Method frame!
6040: conn_state change   rcv_basic_deliver_method ==> rcv_basic_content_header
  <-- Recv Content.Header frame!
6040: conn_state change   rcv_basic_content_header ==> rcv_basic_content_body
  <-- Recv Content.Body frame!
Content Body is [Hello World betty].
@@@ CB: body len : [17]    body : [Hello World betty]
6040: conn_state change   rcv_basic_content_body ==> idle

drive_machine: [conn_idle]  ---  [CONSUMER]: Start waiting to recv!
6040: conn_state change   idle ==> rcv_basic_deliver_method
drive_machine: wait for Basic.Deliver method another 10 seconds!!
drive_machine: wait for Basic.Deliver method another 10 seconds!!

情况二:
[warn] evsignal_init: socketpair: No error
drive_machine: [conn_init]  ---  in TCP 3-way connecting!
drive_machine: [conn_connecting]  ---  connection timeout 1 time on socket(6040)
drive_machine: connected on socket(6040)
6040: conn_state change   connected ==> snd_protocol_header
  --> Send Protocol.Header!
6040: conn_state change   snd_protocol_header ==> rcv_connection_start_method
  <-- Recv Connection.Start Method frame!
6040: conn_state change   rcv_connection_start_method ==> snd_connection_start_rsp_method
  --> Send Connection.Start-Ok Method frame!
6040: conn_state change   snd_connection_start_rsp_method ==> rcv_connection_tune_method
  <-- Recv Connection.Tune Method frame!
6040: conn_state change   rcv_connection_tune_method ==> snd_connection_tune_rsp_method
  --> Send Connection.Tune-Ok Method frame!
6040: conn_state change   snd_connection_tune_rsp_method ==> snd_connection_open_method
  --> Send Connection.Open Method frame!
6040: conn_state change   snd_connection_open_method ==> rcv_connection_open_rsp_method
  <-- Recv Connection.Open-Ok Method frame!
6040: conn_state change   rcv_connection_open_rsp_method ==> snd_channel_open_method
  --> Send Channel.Open Method frame!
6040: conn_state change   snd_channel_open_method ==> rcv_channel_open_rsp_method
  <-- Recv Channel.Open-Ok Method frame!
6040: conn_state change   rcv_channel_open_rsp_method ==> idle

drive_machine: [conn_idle]  ---  [CONSUMER]: Queue Declaring!
6040: conn_state change   idle ==> snd_queue_declare_method
  --> Send Queue.Declare Method frame!
6040: conn_state change   snd_queue_declare_method ==> rcv_queue_declare_rsp_method
  <-- Recv Queue.Declare-Ok Method frame!
6040: conn_state change   rcv_queue_declare_rsp_method ==> idle

drive_machine: [conn_idle]  ---  [CONSUMER]: Queue Binding!
6040: conn_state change   idle ==> snd_queue_bind_method
  --> Send Queue.Bind Method frame!
6040: conn_state change   snd_queue_bind_method ==> rcv_queue_bind_rsp_method
  <-- Recv Queue.Bind-Ok Method frame!
6040: conn_state change   rcv_queue_bind_rsp_method ==> idle

drive_machine: [conn_idle]  ---  [CONSUMER]: Basic QoS!
6040: conn_state change   idle ==> snd_basic_qos_method
  --> Send Basic.Qos Method frame!
6040: conn_state change   snd_basic_qos_method ==> rcv_basic_qos_rsp_method
  <-- Recv Basic.Qos-Ok Method frame!
6040: conn_state change   rcv_basic_qos_rsp_method ==> idle

drive_machine: [conn_idle]  ---  [CONSUMER]: Basic Consuming!
6040: conn_state change   idle ==> snd_basic_consume_method
  --> Send Basic.Consume Method frame!
6040: conn_state change   snd_basic_consume_method ==> rcv_basic_consume_rsp_method
  <-- Recv Basic.Consume-Ok Method frame!
6040: conn_state change   rcv_basic_consume_rsp_method ==> idle

drive_machine: [conn_idle]  ---  [CONSUMER]: Start waiting to recv!
6040: conn_state change   idle ==> rcv_basic_deliver_method
drive_machine: wait for Basic.Deliver method another 10 seconds!!
drive_machine: wait for Basic.Deliver method another 10 seconds!!
drive_machine: wait for Basic.Deliver method another 10 seconds!!

  <-- Recv Basic.Deliver Method frame!
6040: conn_state change   rcv_basic_deliver_method ==> rcv_basic_content_header
  <-- Recv Content.Header frame!
6040: conn_state change   rcv_basic_content_header ==> rcv_basic_content_body
  <-- Recv Content.Body frame!
Content Body is [Hello World betty].
@@@ CB: body len : [17]    body : [Hello World betty]
6040: conn_state change   rcv_basic_content_body ==> snd_basic_ack_method
  --> Send Basic.Ack Method frame!
6040: conn_state change   snd_basic_ack_method ==> idle

drive_machine: [conn_idle]  ---  [CONSUMER]: Start waiting to recv!
6040: conn_state change   idle ==> rcv_basic_deliver_method
drive_machine: wait for Basic.Deliver method another 10 seconds!!
drive_machine: wait for Basic.Deliver method another 10 seconds!!





分享到:0
关注微信,跟着我们扩展技术视野。每天推送IT新技术文章,每周聚焦一门新技术。微信二维码如下:
微信公众账号:尚学堂(微信号:bjsxt-java)
北京总部地址:北京市海淀区西三旗桥东建材城西路85号神州科技园B座三层尚学堂 咨询电话:400-009-1906 010-56233821
Copyright 2007-2015 北京尚学堂科技有限公司 京ICP备13018289号-1 京公网安备11010802015183