RabbitMQ消息确认机制之事务机制

逃离我推掉我的手 2021-08-28 12:25 558阅读 0赞

**一 问题提出**

生产者发送消息出去之后，不知道到底有没有发送到RabbitMQ服务器， 默认是不知道的。而且有的时候我们在发送消息之后，后面的逻辑出问题了，我们不想要发送之前的消息了，需要撤回该怎么办。

**二 解决方案**

1 AMQP 事务机制

2 Confirm 模式

本次关注事务机制

**三 事务机制说明**

AMPQ提供了以下几个方法：

txSelect  将当前channel设置为transaction模式
    txCommit  提交当前事务
    txRollback  事务回滚

**四 代码——正常情况**

1 工具类

/**
    * Copyright (C), 2020-2020, 软件公司
    * FileName: TxSend
    * Author:   cakin
    * Date:     2020/4/25
    * Description: 事务生产者
    */
    package com.rabbitmq.tx;
    
    /**
    * @ClassName: TxSend
    * @Description: 事务生产者
    * @Date: 2020/4/25
    * @Author: cakin
    */
    
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.util.ConnectionUtils;
    import lombok.extern.slf4j.Slf4j;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    /**
    * 生产者
    */
    @Slf4j
    public class TxSend {
    
        /**
         * 队列名
         */
        private static final String QUEUE_NAME = "test_queue_tx";
    
        public static void main( String[] args ) throws IOException, TimeoutException {
            // 获取连接
            Connection connection = ConnectionUtils.getConnection();
            // 从连接开一个通道
            Channel channel = connection.createChannel();
            // 从通道声明一个队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    
            try {
                channel.txSelect(); // 开启事务
    
                // 发送消息
                String message = "hello, tx message";
                channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
                log.info(" [x] Sent message : '" + message + "'");
    
                channel.txCommit(); // 提交事务
            } catch (Exception e) {
                channel.txRollback(); // 回滚事务
                log.info("send message txRollback");
            }
    
            channel.close();
            connection.close();
        }
    }

2 生产者

/**
    * Copyright (C), 2020-2020, 软件公司
    * FileName: TxSend
    * Author:   cakin
    * Date:     2020/4/25
    * Description: 事务生产者
    */
    package com.rabbitmq.tx;
    
    /**
    * @ClassName: TxSend
    * @Description: 事务生产者
    * @Date: 2020/4/25
    * @Author: cakin
    */
    
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.util.ConnectionUtils;
    import lombok.extern.slf4j.Slf4j;
    
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    /**
    * 生产者
    */
    @Slf4j
    public class TxSend {
    
        /**
         * 队列名
         */
        private static final String QUEUE_NAME = "test_queue_tx";
    
        public static void main( String[] args ) throws IOException, TimeoutException {
            // 获取连接
            Connection connection = ConnectionUtils.getConnection();
            // 从连接开一个通道
            Channel channel = connection.createChannel();
            // 从通道声明一个队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    
            try {
                channel.txSelect(); // 开启事务
    
                // 发送消息
                String message = "hello, tx message";
                channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
                log.info(" [x] Sent message : '" + message + "'");
    
                channel.txCommit(); // 提交事务
            } catch (Exception e) {
                channel.txRollback(); // 回滚事务
                log.info("send message txRollback");
            }
    
    
            channel.close();
            connection.close();
        }
    }

3 消费者

/**
    * Copyright (C), 2020-2020, 软件公司
    * FileName: TxRecv
    * Author:   cakin
    * Date:     2020/4/25
    * Description:消费者
    */
    package com.rabbitmq.tx;
    
    import com.rabbitmq.client.AMQP;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.DefaultConsumer;
    import com.rabbitmq.client.Envelope;
    import com.rabbitmq.util.ConnectionUtils;
    import lombok.extern.slf4j.Slf4j;
    
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    
    /**
    * @ClassName: TxRecv
    * @Description: 消费者
    * @Date: 2020/4/25
    * @Author: cakin
    */
    @Slf4j
    public class TxRecv {
        /**
         * 队列名
         */
        private static final String QUEUE_NAME = "test_queue_tx";
    
    
        public static void main( String[] args ) throws IOException, TimeoutException {
            // 获取连接
            Connection connection = ConnectionUtils.getConnection();
    
    
            // 打开通道
            Channel channel = connection.createChannel();
    
    
            // 申明要消费的队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    
    
            // 消费消息
            channel.basicConsume(QUEUE_NAME, true, new DefaultConsumer(channel) {
                @Override
                public void handleDelivery( String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body ) throws IOException {
                    // 接收到的消息
                    String message = new String(body);
                    log.info(" [x] Received '" + message + "'");
                }
            });
        }
    }

4 测试

生产端

15:04:33.686 [main] INFO com.rabbitmq.tx.TxSend -  [x] Sent message : 'hello, tx message'

消费端

15:04:33.693 [pool-1-thread-5] INFO com.rabbitmq.tx.TxRecv -  [x] Received 'hello, tx message'

能正常收发消息。

**五 代码——异常情况**

1 生产者

/**
    * Copyright (C), 2020-2020, 软件公司
    * FileName: TxSendException
    * Author:   cakin
    * Date:     2020/4/25
    * Description: 事务生产者发送消息后异常
    */
    package com.rabbitmq.tx;
    
    /**
    * @ClassName: TxSendException
    * @Description: 事务生产者发送消息后异常
    * @Date: 2020/4/25
    * @Author: cakin
    */
    
    
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.util.ConnectionUtils;
    import lombok.extern.slf4j.Slf4j;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    /**
    * 生产者
    */
    @Slf4j
    public class TxSendException {
    
        /**
         * 队列名
         */
        private static final String QUEUE_NAME = "test_queue_tx";
    
        public static void main( String[] args ) throws IOException, TimeoutException {
            // 获取连接
            Connection connection = ConnectionUtils.getConnection();
            // 从连接开一个通道
            Channel channel = connection.createChannel();
            // 从通道声明一个队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    
            try {
                channel.txSelect(); // 开启事务
    
                // 发送消息
                String message = "hello, tx message exception";
                channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
                log.info(" [x] Sent message : '" + message + "'");
                // 构造异常
                int i = 1 / 0;
    
                channel.txCommit(); // 提交事务
            } catch (ArithmeticException e) {
                channel.txRollback(); // 回滚事务
                log.info("send message exception txRollback");
            }
    
            channel.close();
            connection.close();
        }
    }

2 测试

生成端

15:09:37.806 [main] INFO com.rabbitmq.tx.TxSendException -  [x] Sent message : 'hello, tx message exception'
    15:09:37.816 [main] INFO com.rabbitmq.tx.TxSendException - send message exception txRollback

消费端：无消息

**五 说明**

事务确实能够解决producer与broker之间消息确认的问题，只有消息成功被broker接受，事务提交才能成功，否则我们便可以在捕获异常进行事务回滚操作同时进行消息重发，但是使用事务机制的话会降低RabbitMQ的性能，那么有没有更好的方法既能保障producer知道消息已经正确送到，又能基本上不带来性能上的损失呢？从AMQP协议的层面看是没有更好的方法，但是RabbitMQ提供了一个更好的方案，即将channel信道设置成confirm模式。

**六 参考**

[https://blog.csdn.net/saytime/article/details/80541423][https_blog.csdn.net_saytime_article_details_80541423]

[https_blog.csdn.net_saytime_article_details_80541423]: https://blog.csdn.net/saytime/article/details/80541423