RabbitMQ事务确认机制（生产者）

- 日理万妓 2022-01-28 13:57 270阅读 0赞

# [RabbitMQ事务确认机制（生产者）][RabbitMQ] #

消息确认机制

生产者

消费者

消费者如何确保消息一定消费成功

队列和消费者建立长连接，推送或者拉取形式。

消费者通过自动应答或者手动应答，队列服务器等待应答结果，如果没有应答结果那么保留给下一个消费者。

![1179709-20181112095626875-43343987.png][]

问题产生背景:  
生产者发送消息出去之后，不知道到底有没有发送到RabbitMQ服务器， 默认是不知道的。而且有的时候我们在发送消息之后，后面的逻辑出问题了，我们不想要发送之前的消息了，需要撤回该怎么做。  
解决方案:  
**1.AMQP 事务机制**  
**2.Confirm 模式**  
事务模式:  
txSelect 将当前channel设置为transaction模式 （开启事务）  
txCommit 提交当前事务   （提交事务）  
txRollback 事务回滚   （回滚事务）

如果RabbitMQ服务器宕机了，消息会丢失吗？

RabbitMQ支持消息持久化机制，把消息持久化到硬盘上。

Producer:

package com.toov5.amqp;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.toov5.utils.MQConnectionUtils;
    
    public class Producer {
        // 队列名称
        private static final String UEUE_NAME = "test_queue";
    
        public static void main(String[] args) throws IOException, TimeoutException {
            // 创建新的连接
            Connection connection = MQConnectionUtils.newConnection();
            // 创建Channel
            Channel channel = connection.createChannel();
            // 创建队列
            channel.queueDeclare(UEUE_NAME, false, false, false, null);
            channel.basicQos(1); // 保证 取一个消费
                try {
                    channel.txSelect();    //开启事务
                    // 创建message
                    String msg = "toov5_message";
                    System.out.println("生产者投递消息" + msg );
                    // 生产者发送消息
                    channel.basicPublish("", UEUE_NAME, null, msg.getBytes());
                    int i = 1/0;
                    channel.txCommit();   //提交事务
                } catch (Exception e) {
                    System.out.println("生产者消息事务已经回滚");
                    channel.txRollback();   //回滚事务
                }finally {
                    // 关闭通道和连接
                    channel.close();
                    connection.close();
    
                }            
                
            }
        
        }

consumer:

package com.toov5.amqp;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    import com.rabbitmq.client.AMQP.BasicProperties;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.DefaultConsumer;
    import com.rabbitmq.client.Envelope;
    import com.toov5.utils.MQConnectionUtils;
    
    public class Consumer1 {
      
         //队列名称
            private static final String QUEUE_NAME = "test_queue";
            
            public static void main(String[] args) throws IOException, TimeoutException {
                System.out.println("消费者启动..........1");
                //创建新的连接
            Connection connection = MQConnectionUtils.newConnection();
               //创建Channel
                final Channel channel = connection.createChannel();
                // 消费者关联队列
                 channel.queueDeclare(QUEUE_NAME, false, false, false, null);
                 channel.basicQos(1);
                  DefaultConsumer defaultConsumerr = new DefaultConsumer(channel) {
                      //监听获取消息
                        @Override
                        public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties,
                                byte[] body) throws IOException {
                            String msg =new String(body,"UTF-8");
                            System.out.println("消费者获取生产者消息："+msg);
                            try {
                                //模拟应答等待时间
                                Thread.sleep(1000);
                            } catch (Exception e) {
                                
                            }finally {
                               channel.basicAck(envelope.getDeliveryTag(), false);  //手动应答 告诉消息队列服务器 消费成功
                            }
                        }
                  };
                //牵手模式设置  默认自动应答模式  true:自动应答模式  
                  channel.basicConsume(QUEUE_NAME, false, defaultConsumerr);//    fanse手动应答          
    
            }
    }

结果：

![1179709-20181112101934728-1505960832.png][]

![1179709-20181112101948366-739627675.png][]

上面可以做个AOP~~

可以参考下 Confirm模式

[RabbitMQ]: https://www.cnblogs.com/toov5/p/9945017.html
[1179709-20181112095626875-43343987.png]: /images/20220128/20b633e3d68d425ca77e0104c9c4ba96.png
[1179709-20181112101934728-1505960832.png]: /images/20220128/7d2959c0132e47d199b20b81ca7439a7.png
[1179709-20181112101948366-739627675.png]: /images/20220128/9837882d6bce4a9e93a571e4a07c9ee9.png