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

[转]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)
​