RabbitMQ事务以及confirm机制

喜欢ヅ旅行 2021-10-15 06:23 335阅读 0赞

本文是参考下面这篇博客然后进行复现浓缩后的总结  
[https://www.cnblogs.com/vipstone/p/9350075.html][https_www.cnblogs.com_vipstone_p_9350075.html]

> 正常情况下，如果消息经过交换器进入队列就可以完成消息的持久化，但如果消息在没有到达broker之前出现意外，那就造成消息丢失，有没有办法可以解决这个问题？RabbitMQ有两种方式来解决这个问题：

*  通过AMQP提供的事务机制实现；
 *  使用发送者确认模式实现；

### 事务使用 ###

事物的实现主要是对于信道（Channel）的设置，其中主要的方法有三个：

1.  channel.txSelect()声明启动事务模式；
2.  channel.txComment()提交事务；
3.  channel.txRollback()回滚事务；

#### 尝试代码实现： ####

public static void main(String[] args) throws IOException, TimeoutException { 
        ConnectionFactory factory = new ConnectionFactory();
        factory.setUsername("guest");
        factory.setPassword("guest");
        factory.setHost("localhost");
        factory.setPort(5672);
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        // 队列名， 持久化， 是否排外， 非自动删除
        channel.queueDeclare("queueName",true, false, false, null);
        String message = "这是一个测试消息";
        try { 
            channel.txSelect();
            // String exchange, String routingKey, BasicProperties props, byte[] body
            channel.basicPublish("exchangeName", "queueName", MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes(StandardCharsets.UTF_8));
            channel.txCommit();
        }catch (Exception e){ 
            channel.txRollback();
        }finally { 
            channel.close();
            connection.close();
        }
    }

#### 客户端与rabbitMQ的交互流程如下： ####

1.  客户端发送给服务器Tx.Select(开启事务模式)
2.  服务器端返回Tx.Select-Ok（开启事务模式ok）
3.  推送消息
4.  客户端发送给事务提交Tx.Commit
5.  服务器端返回Tx.Commit-Ok

#### 消费者模式使用事务 ####

假设消费者模式中使用了事务，并且在消息确认之后进行了事务回滚，那么RabbitMQ会产生什么样的变化？结果分为两种情况：

*  autoAck=false手动确认的时候是支持事务的，也就是说即使你已经手动确认了消息，但客户端也会在确认事务返回消息之后，再做决定是确认消息还是重新放回队列，如果你手动确认之后，又回滚了事务，那么以事务回滚为主，此条消息会重新放回队列；
 *  autoAck=true如果确认为true的情况是不支持事务的，也就是说你即使在收到消息之后在回滚事务也是于事无补的，队列已经把消息移除了；

### 二、Confirm发送方确认模式 ###

Confirm发送方确认模式使用和事务类似，也是通过设置Channel进行发送方确认的。Confirm的三种实现方式：  
方式一：channel.waitForConfirms()普通发送方确认模式；  
方式二：channel.waitForConfirmsOrDie()批量确认模式；  
方式三：channel.addConfirmListener()异步监听发送方确认模式；

#### 1.普通发送方确认 ####

public static void main(String[] args) throws IOException, TimeoutException { 
        ConnectionFactory factory = new ConnectionFactory();
        factory.setUsername("guest");
        factory.setPassword("guest");
        factory.setHost("localhost");
        factory.setPort(5672);
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        // 队列名， 持久化， 是否排外， 非自动删除
        channel.queueDeclare("queueName",true, false, false, null);
        String message = "这是一个测试消息";
        // String exchange, String routingKey, BasicProperties props, byte[] body
        channel.basicPublish("exchangeName", "queueName", MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes(StandardCharsets.UTF_8));
        if(channel.waitForConfirms()){ 
            System.out.println("消息发送成功");
        }
    }

#### 2.批量确认模式 ####

public static void main(String[] args) throws IOException, TimeoutException { 
        ConnectionFactory factory = new ConnectionFactory();
        factory.setUsername("guest");
        factory.setPassword("guest");
        factory.setHost("localhost");
        factory.setPort(5672);
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        // 队列名， 持久化， 是否排外， 非自动删除
        channel.queueDeclare("queueName",true, false, false, null);
        channel.confirmSelect();
        for(int i = 0; i < 10; i++){ 
            String message = "这是一个测试消息" + i;
            // String exchange, String routingKey, BasicProperties props, byte[] body
            channel.basicPublish("exchangeName", "queueName", MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes(StandardCharsets.UTF_8));
        }
        channel.waitForConfirmsOrDie();//对所有消息进行等待，只要有一个未发送就会返回异常
    }

#### 3.异步监听发送方确认模式 ####

public void test2() throws IOException, TimeoutException, InterruptedException { 
        ConnectionFactory factory = new ConnectionFactory();
        factory.setUsername("guest");
        factory.setPassword("guest");
        factory.setHost("localhost");
        factory.setPort(5672);
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        // 队列名， 持久化， 是否排外， 非自动删除
        channel.queueDeclare("queueName",true, false, false, null);
        channel.confirmSelect();
        for(int i = 0; i < 10; i++){ 
            String message = "这是一个测试消息" + i;
            // String exchange, String routingKey, BasicProperties props, byte[] body
            channel.basicPublish("exchangeName", "queueName", MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes(StandardCharsets.UTF_8));
        }
        channel.addConfirmListener(new ConfirmListener() { 
            @Override
            public void handleAck(long l, boolean b) throws IOException { 
                System.out.println("消息发送失败，标识：" + l);
            }
    
            @Override
            public void handleNack(long l, boolean b) throws IOException { 
                System.out.println(String.format("消息发送成功，标识：%d， 是否是多个同时确认：%b", l, b));
            }
        });
    }

> 可以看出，代码是异步执行的，消息确认有可能是批量确认的，是否批量确认在于返回的multiple的参数，此参数为bool值，如果true表示批量执行了deliveryTag这个值以前的所有消息，如果为false的话表示单条确认。

### 总结： ###

Confirm批量确定和Confirm异步模式性能相差不大，Confirm模式要比事务快10倍左右。

[https_www.cnblogs.com_vipstone_p_9350075.html]: https://www.cnblogs.com/vipstone/p/9350075.html