登录 注册

深入学习 RabbitMQ(三)：channel 的 confirm 模式

本贴最后更新于 2491 天前，其中的信息可能已经时移俗易

[转]http://blog.csdn.net/hzw19920329/article/details/54340711

上一篇博客我们介绍了使用 RabbitMQ 可能会遇到的一个问题，即生产者不知道消息是否真正到达 broker 代理服务器，随后通过 AMQP 协议层面为我们提供的事务机制解决了这个问题，但是采用事务机制实现会降低 RabbitMQ 的消息吞吐量，那么有没有更加高效的解决方式呢？RabbitMQ 团队为我们拿出了更好的方案，即采用发送方确认模式；

   生产者确认模式实现原理：

   生产者将信道设置成confirm模式，一旦信道进入confirm模式，所有在该信道上面发布的消息都将会被指派一个唯一的ID(从1开始)，一旦消息被投递到所有匹配的队列之后，broker就会发送一个确认给生产者(包含消息的唯一ID)，这就使得生产者知道消息已经正确到达目的队列了，如果消息和队列是可持久化的，那么确认消息会在将消息写入磁盘之后发出，broker回传给生产者的确认消息中delivery-tag域包含了确认消息的序列号，此外broker也可以设置basic.ack的multiple域，表示到这个序列号之前的所有消息都已经得到了处理；

   confirm模式最大的好处在于他是异步的，一旦发布一条消息，生产者应用程序就可以在等信道返回确认的同时继续发送下一条消息，当消息最终得到确认之后，生产者应用便可以通过回调方法来处理该确认消息，如果RabbitMQ因为自身内部错误导致消息丢失，就会发送一条nack消息，生产者应用程序同样可以在回调方法中处理该nack消息；

   开启confirm模式的方法：

   生产者通过调用channel的confirmSelect方法将channel设置为confirm模式，(注意一点，已经在transaction事务模式的channel是不能再设置成confirm模式的，即这两种模式是不能共存的)，如果没有设置no-wait标志的话，broker会返回confirm.select-ok表示同意发送者将当前channel信道设置为confirm模式(从目前RabbitMQ最新版本3.6来看，如果调用了channel.confirmSelect方法，默认情况下是直接将no-wait设置成false的，也就是默认情况下broker是必须回传confirm.select-ok的，而且我也没找到我们自己能够设置no-wait标志的方法)；

   生产者实现confiem模式有三种编程方式：

   (1)：普通confirm模式，每发送一条消息，调用waitForConfirms()方法等待服务端confirm，这实际上是一种串行的confirm，每publish一条消息之后就等待服务端confirm，如果服务端返回false或者超时时间内未返回，客户端进行消息重传；

   (2)：批量confirm模式，每发送一批消息之后，调用waitForConfirms()方法，等待服务端confirm，这种批量确认的模式极大的提高了confirm效率，但是如果一旦出现confirm返回false或者超时的情况，客户端需要将这一批次的消息全部重发，这会带来明显的重复消息，如果这种情况频繁发生的话，效率也会不升反降；

   

   讲完了基本的原理之后，代码级别我们该怎么设置channel信道为confirm模式呢？以及我们该怎么获取broker返回给我们的确认消息呢？

   测试1：普通confirm模式

   首先从最简单的开始，仅仅将channel设置成confirm模式，并且生产者每发送一条消息就等待broker回应确认消息，至于确认消息是什么我们不去做任何处理，为了测试方便，此处生产者只发送了5条消息，实现代码如下：

[java] view plain copy

  1. public class ProducerTest {

  2.  public static void main(String[] args) {  

  3.      String exchangeName = "confirmExchange";  

  4.      String queueName = "confirmQueue";  

  5.      String routingKey = "confirmRoutingKey";  

  6.      String bindingKey = "confirmRoutingKey";  

  7.      int count = 5;  

  9.      ConnectionFactory factory = new ConnectionFactory();  

  10.      factory.setHost("172.16.151.74");  

  11.      factory.setUsername("test");  

  12.      factory.setPassword("test");  

  13.      factory.setPort(5672);  

  15.      //创建生产者  

  16.      Sender producer = new Sender(factory, count, exchangeName, queueName,routingKey,bindingKey);  

  17.      producer.run();  

  18.  }

  19. }

  21. class Sender

  22. {

  23.  private ConnectionFactory factory;  

  24.  private int count;  

  25.  private String exchangeName;  

  26.  private String  queueName;  

  27.  private String routingKey;  

  28.  private String bindingKey;  

  30.  public Sender(ConnectionFactory factory,int count,String exchangeName,String queueName,String routingKey,String bindingKey) {  

  31.      this.factory = factory;  

  32.      this.count = count;  

  33.      this.exchangeName = exchangeName;  

  34.      this.queueName = queueName;  

  35.      this.routingKey = routingKey;  

  36.      this.bindingKey = bindingKey;  

  37.  }  

  39.  public void run() {  

  40.      Channel channel = null;  

  41.      try {  

  42.          Connection connection = factory.newConnection();  

  43.          channel = connection.createChannel();  

  44.          //创建exchange  

  45.          channel.exchangeDeclare(exchangeName, "direct", true, false, null);  

  46.          //创建队列  

  47.          channel.queueDeclare(queueName, true, false, false, null);  

  48.          //绑定exchange和queue  

  49.          channel.queueBind(queueName, exchangeName, bindingKey);  

  50.          channel.confirmSelect();  

  51.          //发送持久化消息  

  52.          for(int i = 0;i < count;i++)  

  53.          {  

  54.              //第一个参数是exchangeName(默认情况下代理服务器端是存在一个""名字的exchange的,  

  55.              //因此如果不创建exchange的话我们可以直接将该参数设置成"",如果创建了exchange的话  

  56.              //我们需要将该参数设置成创建的exchange的名字),第二个参数是路由键  

  57.              channel.basicPublish(exchangeName, routingKey,MessageProperties.PERSISTENT_BASIC, ("第"+(i+1)+"条消息").getBytes());  

  58.              if(channel.waitForConfirms())  

  59.              {  

  60.                  System.out.println("发送成功");  

  61.              }  

  62.          }  

  63.          final long start = System.currentTimeMillis();  

  64.          System.out.println("执行waitForConfirmsOrDie耗费时间: "+(System.currentTimeMillis()-start)+"ms");  

  65.      } catch (Exception e) {  

  66.          e.printStackTrace();  

  67.      }  

  68.  }

  69. }

    在第 50 行调用 Channel 信道的 confirmSelect 方法将当前信道设置成了 confirm 模式，第 57 行通过 for 循环调用 Channel 的 basicPublish 方法发送了 5 条消息到消息队列中，第 58 行调用 waitForConfirms 方法等待 broker 服务端返回 ack 或者 nack 消息，这种模式每发送一条消息就会等待 broker 代理服务器返回消息，这点我们可以从抓包的角度观察结果：

   可以看到上面生产者通过Confirm.Select将当前Channel信道设置成confirm模式，broker代理服务器收到之后回传Confirm.Select-Ok同一将当前Channel设置成confirm模式，此外看到返回5条Basic.Ack消息；

    测试2：批量confirm模式

    这种模式生产者不是每发送一条就等待broker确认，而是发送一批，实现代码见下：

[java] view plain copy

  1. public class ProducerTest {

  2.  public static void main(String[] args) {  

  3.      String exchangeName = "confirmExchange";  

  4.      String queueName = "confirmQueue";  

  5.      String routingKey = "confirmRoutingKey";  

  6.      String bindingKey = "confirmRoutingKey";  

  7.      int count = 100;  

  9.      ConnectionFactory factory = new ConnectionFactory();  

  10.      factory.setHost("172.16.151.74");  

  11.      factory.setUsername("test");  

  12.      factory.setPassword("test");  

  13.      factory.setPort(5672);  

  15.      //创建生产者  

  16.      Sender producer = new Sender(factory, count, exchangeName, queueName,routingKey,bindingKey);  

  17.      producer.run();  

  18.  }

  19. }

  21. class Sender

  22. {

  23.  private ConnectionFactory factory;  

  24.  private int count;  

  25.  private String exchangeName;  

  26.  private String  queueName;  

  27.  private String routingKey;  

  28.  private String bindingKey;  

  30.  public Sender(ConnectionFactory factory,int count,String exchangeName,String queueName,String routingKey,String bindingKey) {  

  31.      this.factory = factory;  

  32.      this.count = count;  

  33.      this.exchangeName = exchangeName;  

  34.      this.queueName = queueName;  

  35.      this.routingKey = routingKey;  

  36.      this.bindingKey = bindingKey;  

  37.  }  

  39.  public void run() {  

  40.      Channel channel = null;  

  41.      try {  

  42.          Connection connection = factory.newConnection();  

  43.          channel = connection.createChannel();  

  44.          //创建exchange  

  45.          channel.exchangeDeclare(exchangeName, "direct", true, false, null);  

  46.          //创建队列  

  47.          channel.queueDeclare(queueName, true, false, false, null);  

  48.          //绑定exchange和queue  

  49.          channel.queueBind(queueName, exchangeName, bindingKey);  

  50.          channel.confirmSelect();  

  51.          //发送持久化消息  

  52.          for(int i = 0;i < count;i++)  

  53.          {  

  54.              //第一个参数是exchangeName(默认情况下代理服务器端是存在一个""名字的exchange的,  

  55.              //因此如果不创建exchange的话我们可以直接将该参数设置成"",如果创建了exchange的话  

  56.              //我们需要将该参数设置成创建的exchange的名字),第二个参数是路由键  

  57.              channel.basicPublish(exchangeName, routingKey,MessageProperties.PERSISTENT_BASIC, ("第"+(i+1)+"条消息").getBytes());  

  58.          }  

  59.          long start = System.currentTimeMillis();  

  60.          channel.waitForConfirmsOrDie();  

  61.          System.out.println("执行waitForConfirmsOrDie耗费时间: "+(System.currentTimeMillis()-start)+"ms");  

  62.      } catch (Exception e) {  

  63.          e.printStackTrace();  

  64.      }  

  65.  }

  66. }

    第 50 行调用 channel.confirmSelect 将当前 channel 信道设置成 confirm 模式，接着在第 57 行通过 for 循环发送了 100 条消息，第 60 行调用了 channel 的 waitForConfirmsOrDie，从 waitForConfirmsOrDie 方法的注释上可以看出，该方法会等到最后一条消息得到确认或者得到 nack 才会结束，也就是说在 waitForConfirmsOrDie 处会造成当前程序的阻塞，以测试 1 程序发送 100 条消息为例，阻塞时间是 135ms，我们再来看看对测试 1 的抓包情况：

   从红色箭头的标号1出可以看到：首先是24向74发送了Confirm.Select消息表示请求将当前信道设置为confirm模式，接着74向24回送了Confirm.Select-Ok消息表示同意将信道设置成confirm模式，从红色标号2处NoWait字段的值为false也印证了我们如果直接调用Channel信道的confirmSelect()方法的话，实际上默认是开启broker回传Confirm.Select-Ok确认消息的；  

   接下来我们看看broker回传给客户端的确认消息数据包是什么样子的呢？同样通过抓包看看结果：


   你会发现，在上面测试1中我们通过for循环发送了100条消息，但是在抓包的时候我们仅仅看到有两个Basic.Ack确认消息回传回来，原因在于上面截图的标号3处，你会发现Multiple域的值是True的，之前我们已经讲过broker可以设置Multiple域表示broker已经收到当前确认消息的Delivery-Tag域之前标号的消息，以上面截图为例的话表示broker告诉发送者编号4之前的消息已经全部收到了，从这点我们看出broker端默认情况下是进行批量回复的，并不是针对每条消息都发送一条ack消息；

   测试2：

   测试1我们仅仅是测试发送者能够收到broker的确认消息以及知道了broker对消息默认是采用批量回复方式的，那么在程序中我们该怎么获取到broker回传回来的确认消息呢，假如我们有时候需要在收到确认消息之后做一些提示性操作该怎么办呢？测试1中，我们采用的是Channel信道的waitForConfirmsOrDie等待broker端回传回ack确认消息的，但我们没法拿到这个ack消息进行后期操作，要想拿到ack消息的话，我们可以给当前Channel信道绑定监听器，具体来说就是调用Channel信道的addConfirmListener方法进行设置，Channel信道在收到broker的ack消息之后会回调设置在该信道监听器上的handleAck方法，在收到nack消息之后会回调设置在该信道监听器上的handleNack方法。

   实现代码：

[java] view plain copy

  1. public class ProducerTest {

  2.  public static void main(String[] args) {  

  3.      String exchangeName = "confirmExchange";  

  4.      String queueName = "confirmQueue";  

  5.      String routingKey = "confirmRoutingKey";  

  6.      String bindingKey = "confirmRoutingKey";  

  7.      int count = 100;  

  9.      ConnectionFactory factory = new ConnectionFactory();  

  10.      factory.setHost("172.16.151.74");  

  11.      factory.setUsername("test");  

  12.      factory.setPassword("test");  

  13.      factory.setPort(5672);  

  15.      //创建生产者  

  16.      Sender producer = new Sender(factory, count, exchangeName, queueName,routingKey,bindingKey);  

  17.      producer.run();  

  18.  }

  19. }

  21. class Sender

  22. {

  23.  private ConnectionFactory factory;  

  24.  private int count;  

  25.  private String exchangeName;  

  26.  private String  queueName;  

  27.  private String routingKey;  

  28.  private String bindingKey;  

  30.  public Sender(ConnectionFactory factory,int count,String exchangeName,String queueName,String routingKey,String bindingKey) {  

  31.      this.factory = factory;  

  32.      this.count = count;  

  33.      this.exchangeName = exchangeName;  

  34.      this.queueName = queueName;  

  35.      this.routingKey = routingKey;  

  36.      this.bindingKey = bindingKey;  

  37.  }  

  39.  public void run() {  

  40.      Channel channel = null;  

  41.      try {  

  42.          Connection connection = factory.newConnection();  

  43.          channel = connection.createChannel();  

  44.          //创建exchange  

  45.          channel.exchangeDeclare(exchangeName, "direct", true, false, null);  

  46.          //创建队列  

  47.          channel.queueDeclare(queueName, true, false, false, null);  

  48.          //绑定exchange和queue  

  49.          channel.queueBind(queueName, exchangeName, bindingKey);  

  50.          channel.confirmSelect();  

  51.          //发送持久化消息  

  52.          for(int i = 0;i < count;i++)  

  53.          {  

  54.              //第一个参数是exchangeName(默认情况下代理服务器端是存在一个""名字的exchange的,  

  55.              //因此如果不创建exchange的话我们可以直接将该参数设置成"",如果创建了exchange的话  

  56.              //我们需要将该参数设置成创建的exchange的名字),第二个参数是路由键  

  57.              channel.basicPublish(exchangeName, routingKey,MessageProperties.PERSISTENT_BASIC, ("第"+(i+1)+"条消息").getBytes());  

  58.          }  

  59.          long start = System.currentTimeMillis();  

  60.          channel.addConfirmListener(new ConfirmListener() {  

  62.              @Override  

  63.              public void handleNack(long deliveryTag, boolean multiple) throws IOException {  

  64.                  System.out.println("nack: deliveryTag = "+deliveryTag+" multiple: "+multiple);  

  65.              }  

  67.              @Override  

  68.              public void handleAck(long deliveryTag, boolean multiple) throws IOException {  

  69.                  System.out.println("ack: deliveryTag = "+deliveryTag+" multiple: "+multiple);  

  70.              }  

  71.          });  

  72.          System.out.println("执行waitForConfirmsOrDie耗费时间: "+(System.currentTimeMillis()-start)+"ms");  

  73.      } catch (Exception e) {  

  74.          e.printStackTrace();  

  75.      }  

  76.  }

  77. }

    第 60 行我们调用了 Channel 信道的 addConfirmListener 设置了监听器，并且在监听器的 handleAck 和 handleNack 方法中打印了信息，运行程序查看输出：

   可以看到，虽然我们还是发送了100条消息，同样我们并没有收到100个ack消息 ，只收到两个ack消息，并且这两个ack消息的multiple域都为true，这点和测试1是相同的，你多次运行程序会发现每次发送回来的ack消息中的deliveryTag域的值并不是一样的，说明broker端批量回传给发送者的ack消息并不是以固定的批量大小回传的；

   也就是我们通过信道Channel的waitForConfirmsOrDie方法或者为信道设置监听器都可以保证发送者收到broker回传的ack或者nack消息，那么这两种方式有什么区别呢？从测试一的第61行代码以及测试2的第72行代码处你就能找到答案啦，测试1中调用waitForConfirmsOrDie方法发送100条消息并且全部收到确认需要135ms，测试2中通过监听器的方式仅仅需要1ms，说明调用waitForConfirmsOrDie会造成程序的阻塞，通过监听器并不会造成程序的阻塞，下一篇博客我会试着从RabbitMQ的源码层面来分析这两种方式造成这种区别的原因啦啦；

   参考资料：

   [RabbitMQ官网](http://www.rabbitmq.com/confirms.html)

   [RabbitMQ不同Confirm模式下的性能对比](http://ju.outofmemory.cn/entry/177937)
  • RabbitMQ

    RabbitMQ 是一个开源的 AMQP 实现，服务器端用 Erlang 语言编写，支持多种语言客户端，如：Python、Ruby、.NET、Java、C、PHP、ActionScript 等。用于在分布式系统中存储转发消息，在易用性、扩展性、高可用性等方面表现不俗。

    49 引用 • 60 回帖 • 362 关注
613 2 105 150 21 42 6 49 229

相关帖子

回帖
深入学习 RabbitMQ(三)：channel 的 confirm 模式

欢迎来到这里！

我们正在构建一个小众社区，大家在这里相互信任，以平等 • 自由 • 奔放的价值观进行分享交流。最终，希望大家能够找到与自己志同道合的伙伴，共同成长。

注册 关于
请输入回帖内容 ...
honglan
回帖
0
 帖子
28
 积分
621

近期热议

推荐标签 标签

  • FFmpeg

    FFmpeg 是一套可以用来记录、转换数字音频、视频，并能将其转化为流的开源计算机程序。

    23 引用 • 32 回帖
  • Spring

    Spring 是一个开源框架，是于 2003 年兴起的一个轻量级的 Java 开发框架，由 Rod Johnson 在其著作《Expert One-On-One J2EE Development and Design》中阐述的部分理念和原型衍生而来。它是为了解决企业应用开发的复杂性而创建的。框架的主要优势之一就是其分层架构，分层架构允许使用者选择使用哪一个组件，同时为 JavaEE 应用程序开发提供集成的框架。

    944 引用 • 1459 回帖 • 18 关注
  • PHP

    PHP（Hypertext Preprocessor）是一种开源脚本语言。语法吸收了 C 语言、 Java 和 Perl 的特点，主要适用于 Web 开发领域，据说是世界上最好的编程语言。

    179 引用 • 407 回帖 • 487 关注
  • Flutter

    Flutter 是谷歌的移动 UI 框架，可以快速在 iOS 和 Android 上构建高质量的原生用户界面。 Flutter 可以与现有的代码一起工作，它正在被越来越多的开发者和组织使用，并且 Flutter 是完全免费、开源的。

    39 引用 • 92 回帖
  • InfluxDB

    InfluxDB 是一个开源的没有外部依赖的时间序列数据库。适用于记录度量，事件及实时分析。

    2 引用 • 72 关注
  • iOS

    iOS 是由苹果公司开发的移动操作系统，最早于 2007 年 1 月 9 日的 Macworld 大会上公布这个系统，最初是设计给 iPhone 使用的，后来陆续套用到 iPod touch、iPad 以及 Apple TV 等产品上。iOS 与苹果的 Mac OS X 操作系统一样，属于类 Unix 的商业操作系统。

    85 引用 • 139 回帖
  • ReactiveX

    ReactiveX 是一个专注于异步编程与控制可观察数据（或者事件）流的 API。它组合了观察者模式，迭代器模式和函数式编程的优秀思想。

    1 引用 • 2 回帖 • 156 关注
  • Maven

    Maven 是基于项目对象模型（POM）、通过一小段描述信息来管理项目的构建、报告和文档的软件项目管理工具。

    186 引用 • 318 回帖 • 302 关注
  • Gitea

    Gitea 是一个开源社区驱动的轻量级代码托管解决方案，后端采用 Go 编写，采用 MIT 许可证。

    4 引用 • 16 回帖 • 1 关注
  • SEO

    发布对别人有帮助的原创内容是最好的 SEO 方式。

    35 引用 • 200 回帖 • 22 关注
  • JVM

    JVM（Java Virtual Machine）Java 虚拟机是一个微型操作系统，有自己的硬件构架体系，还有相应的指令系统。能够识别 Java 独特的 .class 文件（字节码），能够将这些文件中的信息读取出来，使得 Java 程序只需要生成 Java 虚拟机上的字节码后就能在不同操作系统平台上进行运行。

    180 引用 • 120 回帖
  • Sublime

    Sublime Text 是一款可以用来写代码、写文章的文本编辑器。支持代码高亮、自动完成，还支持通过插件进行扩展。

    10 引用 • 5 回帖
  • OpenStack

    OpenStack 是一个云操作系统，通过数据中心可控制大型的计算、存储、网络等资源池。所有的管理通过前端界面管理员就可以完成，同样也可以通过 Web 接口让最终用户部署资源。

    10 引用 • 3 关注
  • frp

    frp 是一个可用于内网穿透的高性能的反向代理应用，支持 TCP、UDP、 HTTP 和 HTTPS 协议。

    20 引用 • 7 回帖
  • V2Ray
    1 引用 • 15 回帖
  • Logseq

    Logseq 是一个隐私优先、开源的知识库工具。

    Logseq is a joyful, open-source outliner that works on top of local plain-text Markdown and Org-mode files. Use it to write, organize and share your thoughts, keep your to-do list, and build your own digital garden.

    6 引用 • 63 回帖
  • 锤子科技

    锤子科技（Smartisan）成立于 2012 年 5 月，是一家制造移动互联网终端设备的公司，公司的使命是用完美主义的工匠精神，打造用户体验一流的数码消费类产品（智能手机为主），改善人们的生活质量。

    4 引用 • 31 回帖 • 4 关注
  • PostgreSQL

    PostgreSQL 是一款功能强大的企业级数据库系统，在 BSD 开源许可证下发布。

    22 引用 • 22 回帖
  • 生活

    生活是指人类生存过程中的各项活动的总和，范畴较广，一般指为幸福的意义而存在。生活实际上是对人生的一种诠释。生活包括人类在社会中与自己息息相关的日常活动和心理影射。

    230 引用 • 1454 回帖
  • Pipe

    Pipe 是一款小而美的开源博客平台。Pipe 有着非常活跃的社区，可将文章作为帖子推送到社区，来自社区的回帖将作为博客评论进行联动（具体细节请浏览 B3log 构思 - 分布式社区网络）。

    这是一种全新的网络社区体验，让热爱记录和分享的你不再感到孤单！

    132 引用 • 1114 回帖 • 124 关注
  • Solidity

    Solidity 是一种智能合约高级语言，运行在 [以太坊] 虚拟机（EVM）之上。它的语法接近于 JavaScript，是一种面向对象的语言。

    3 引用 • 18 回帖 • 398 关注
  • SSL

    SSL（Secure Sockets Layer 安全套接层），及其继任者传输层安全（Transport Layer Security，TLS）是为网络通信提供安全及数据完整性的一种安全协议。TLS 与 SSL 在传输层对网络连接进行加密。

    70 引用 • 193 回帖 • 431 关注
  • 域名

    域名（Domain Name），简称域名、网域，是由一串用点分隔的名字组成的 Internet 上某一台计算机或计算机组的名称，用于在数据传输时标识计算机的电子方位（有时也指地理位置）。

    43 引用 • 208 回帖 • 1 关注
  • 负能量

    上帝为你关上了一扇门，然后就去睡觉了....努力不一定能成功，但不努力一定很轻松 (° ー °〃)

    88 引用 • 1235 回帖 • 411 关注
  • Facebook

    Facebook 是一个联系朋友的社交工具。大家可以通过它和朋友、同事、同学以及周围的人保持互动交流，分享无限上传的图片，发布链接和视频，更可以增进对朋友的了解。

    4 引用 • 15 回帖 • 455 关注
  • 创业

    你比 99% 的人都优秀么？

    84 引用 • 1399 回帖 • 1 关注
  • Gzip

    gzip （GNU zip）是 GNU 自由软件的文件压缩程序。我们在 Linux 中经常会用到后缀为 .gz 的文件，它们就是 Gzip 格式的。现今已经成为互联网上使用非常普遍的一种数据压缩格式，或者说一种文件格式。

    9 引用 • 12 回帖 • 135 关注

最新标签