RabbitMQ 事务机制

女爷i 2024-03-22 22:08 60阅读 0赞

##### 一、使用RabbitMQ可能存在的问题 #####

1.  服务器异常数据丢失问题

解决方案：采用持久化数据，即声明队列时设置

queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments)
    
    durable = true

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

1.  AMQP 事务机制【本章节】
2.  Confirm 模式

##### 二、事务机制 #####

AMQP提供了以下几个方法：

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

##### 三、代码演示 #####

###### 生产者 ######

package com.rabbitmq.tx;
    
    import com.rabbitmq.client.BuiltinExchangeType;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.util.ConnectionUtil;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    /**
     * 生产者
     */
    public class TxSend {
        
    
        private static final String QUEUE_NAME = "test_queue_tx";
    
        public static void main(String[] args) throws IOException, TimeoutException {
        
            // 获取连接
            Connection connection = ConnectionUtil.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());
                System.out.println(" [x] Sent message : '" + message + "'");
                
                channel.txCommit(); // 提交事务
            }catch (Exception e){
        
                channel.txRollback(); // 回滚事务
                System.out.println("send message txRollback");
            }
    
            channel.close();
            connection.close();
        }
    
    }

###### 消费者 ######

package com.rabbitmq.tx;
    
    import com.rabbitmq.client.*;
    import com.rabbitmq.util.ConnectionUtil;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    /**
     * 消费者
     */
    public class TxRecv {
        
    
        private static final String QUEUE_NAME = "test_queue_tx";
    
        public static void main(String[] args) throws IOException, TimeoutException {
        
            // 获取连接
            Connection connection = ConnectionUtil.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);
                    System.out.println(" [x] Received '" + message + "'");
                }
            });
    
        }
    
    }

##### 四、测试结果 #####

能够正常收发。  
这时候生产者加上异常

package com.rabbitmq.tx;
    
    import com.rabbitmq.client.BuiltinExchangeType;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.util.ConnectionUtil;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    /**
     * 生产者
     */
    public class TxSend {
        
    
        private static final String QUEUE_NAME = "test_queue_tx";
    
        public static void main(String[] args) throws IOException, TimeoutException {
        
            // 获取连接
            Connection connection = ConnectionUtil.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());
                System.out.println(" [x] Sent message : '" + message + "'");
                int i = 1/0;
                channel.txCommit(); // 提交事务
            }catch (Exception e){
        
                channel.txRollback(); // 回滚事务
                System.out.println("send message txRollback");
            }
    
            channel.close();
            connection.close();
        }
    
    }

生产者

[x] Sent message : 'hello, tx message'
    send message txRollback

消费者  
消费者并没有接收到消息。说明生产者的消息回滚了，事务生效。

##### 五、缺点 #####

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