您的位置: 首页  >  文章  >  深入学习RabbitMQ(二)：AMQP事务机制

深入学习RabbitMQ(二)：AMQP事务机制

分类: 文章 • 2025-02-27 14:04:10

  在使用RabbitMQ的时候，我们可以通过消息持久化操作来解决因为服务器的异常奔溃导致的消息丢失，除此之外我们还会遇到一个问题，当消息的发布者在将消息发送出去之后，消息到底有没有正确到达broker代理服务器呢？如果不进行特殊配置的话，默认情况下发布操作是不会返回任何信息给生产者的，也就是默认情况下我们的生产者是不知道消息有没有正确到达broker的，如果在消息到达broker之前已经丢失的话，持久化操作也解决不了这个问题，因为消息根本就没到达代理服务器，你怎么进行持久化，那么这个问题该怎么解决呢？

        RabbitMQ为我们提供了两种方式：

        方式一：通过AMQP事务机制实现，这也是从AMQP协议层面提供的解决方案；

        方式二：通过将channel设置成confirm模式来实现；

        这篇博客我们讲解AMQP事务机制，下一篇再探讨channel的confirm模式

        首先，我们通过实例来看看AMQP的事务模式是怎么使用的：

        RabbitMQ中与事务机制有关的方法有三个，分别是Channel里面的txSelect()，txCommit()以及txRollback()，txSelect用于将当前Channel设置成是transaction模式，txCommit用于提交事务，txRollback用于回滚事务，在通过txSelect开启事务之后，我们便可以发布消息给broker代理服务器了，如果txCommit提交成功了，则消息一定是到达broker了，如果在txCommit执行之前broker异常奔溃或者由于其他原因抛出异常，这个时候我们便可以捕获异常通过txRollback回滚事务了；

       具体实例：

[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 = 3;  
  8.           
  9.         ConnectionFactory factory = new ConnectionFactory();  
  10.         factory.setHost("172.16.151.74");  
  11.         factory.setUsername("test");  
  12.         factory.setPassword("test");  
  13.         factory.setPort(5672);  
  14.           
  15.         //创建生产者  
  16.         Sender producer = new Sender(factory, count, exchangeName, queueName,routingKey,bindingKey);  
  17.         producer.run();  
  18.     }  
  19. }  
  20.   
  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;  
  29.       
  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.     }  
  38.       
  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.             //发送持久化消息  
  51.             for(int i = 0;i < count;i++)  
  52.             {  
  53.                 //第一个参数是exchangeName(默认情况下代理服务器端是存在一个""名字的exchange的,  
  54.                 //因此如果不创建exchange的话我们可以直接将该参数设置成"",如果创建了exchange的话  
  55.                 //我们需要将该参数设置成创建的exchange的名字),第二个参数是路由键  
  56.                 //开启事务  
  57.                 channel.txSelect();  
  58.                 channel.basicPublish(exchangeName, routingKey, true, MessageProperties.PERSISTENT_BASIC, ("第"+(i+1)+"条消息").getBytes());  
  59.                 if(i == 1)  
  60.                 {  
  61.                     int result = 1/0;  
  62.                 }  
  63.                 //提交事务  
  64.                 channel.txCommit();  
  65.             }  
  66.         } catch (Exception e) {  
  67.             try {  
  68.                 //回滚操作  
  69.                 channel.txRollback();  
  70.             } catch (IOException e1) {  
  71.                 e1.printStackTrace();  
  72.             }  
  73.             e.printStackTrace();  
  74.         }  
  75.     }  
  76. }  
       在第57行通过channel.txSelect方法开启事务，第64行通过channel.txCommit提交事务，为了模拟broker代理服务器异常奔溃或者发布过程中抛出异常，我们通过第61行除以0的操作来模拟（实际中第58行的basicPublish方法是有可能会抛出IOException异常），在捕获到异常之后，第69行调用了channel.txRollback进行事务回滚操作，运行整个程序你会发现在"confirmQueue"这个队列中只存储了一条消息，因为在59行i等于1的时候，抛出了异常，调用了第69行进行了事务回滚操作；在实际应用中，可以在回滚操作之后进行消息重发操作；
       我们来通过抓包看看程序执行过程中发出了哪些请求：

   深入学习RabbitMQ(二)：AMQP事务机制

       1：第一条消息调用channel.txSelect开启事务

       2：第一条消息调用channel.txCommit提交事务

       3：第二条消息调用channel.txSelect开启事务

       4：因为除以0的操作程序抛出异常，执行catch语句中的channel.txRollback回滚事务

       从上面的分析中，我们知道使用事务确实能够解决发布者与broker代理服务器之间的消息确认，只有消息成功被broker接收事务提交才能成功，否则我们便可以在捕获异常进行事务回滚操作同时进行消息重发，但是使用事务机制的话会降低RabbitMQ的性能，就拿上面的程序发送1000条消息，使用事务的话需要58244毫秒，而不使用事务的话仅仅需要89毫秒，因此在实际中使用事务会带来很大的性能损失，那么有没有更好的方法既能保证发布者知道消息已经正确到达，又能基本上不带来性能上的损失呢？从AMQP协议的层面看是没有更好的方法的，但是RabbitMQ提供了一个更好的方案，即将channel信道设置成confirm模式，关于confirm的注意点将在下一篇博客介绍；




http://blog.****.net/hzw19920329/article/details/54315940

相关推荐