【RabbitMQ】消息可靠性投递（一）Producer-＞Broker

待我称王封你为后i 2022-11-03 11:19 147阅读 0赞

> 系列文章：
> 
>  *  [【RabbitMQ】如何保证消息可靠性投递？][RabbitMQ]
>  *  [【RabbitMQ】消息可靠性投递（一）Producer-＞Broker][RabbitMQ_Producer-_Broker]
>  *  [【RabbitMQ】消息可靠性投递（二）Exchange-＞Queue][RabbitMQ_Exchange-_Queue]
>  *  [【RabbitMQ】消息可靠性投递（三）Queue存储消息][RabbitMQ_Queue]
>  *  [【RabbitMQ】消息可靠性投递（四）Queue–＞Consumer][RabbitMQ_Queue_Consumer]

第一个环节是生产者发送消息到Broker。可能因为网络或者Broker的问题导致消息发送失败，生产者不能确定Broker有没有正确的接收。 在 RabbitMQ 里面提供了两种机制服务端确认机制，也就是在生产者发送消息给RabbitMQ的服务端的时候，服务端会通过某种方式返回一个应答，只要生产者收到了这个应答，就知道消息发送成功了。

第一种是Transaction（事务）模式，第二种Confirm（确认）模式。

## 1.事务模式ACK ##

通过`channel.txSelect()`方法把信道设置成事务模式，然后就可以发布消息给RabbitMQ了

*  若`channel.txCommit()`的方法调用成功，就说明事务提交成功，则消息一定到达了RabbitMQ中。
 *  如果在事务提交执行之前由于RabbitMQ异常崩溃或者其他原因抛出异常，这个时候便可以将其捕获，进而通过执行`channel.txRollback()`方法来实现事务回滚。

public class TransactionProducer {
         // 原生API
        private final static String QUEUE_NAME = "ORIGIN_QUEUE";
    
        public static void main(String[] args) throws Exception {
        
            ConnectionFactory factory = new ConnectionFactory();
            factory.setUri(ResourceUtil.getKey("rabbitmq.uri"));
           
            Connection conn = factory.newConnection(); // 建立连接        
            Channel channel = conn.createChannel(); // 创建消息通道
    
            String msg = "Hello world, Rabbit MQ";
            // 声明队列（默认交换机AMQP default，Direct）
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    
            try {
        
                channel.txSelect(); // 事务模式
                // 发送消息
                // String exchange, String routingKey, BasicProperties props, byte[] body
                channel.basicPublish("", QUEUE_NAME, null, (msg).getBytes());
                // int i =1/0;
                channel.txCommit(); // 提交，阻塞
                System.out.println("消息发送成功");
            } catch (Exception e) {
        
                channel.txRollback(); // 回滚
                System.out.println("消息已经回滚");
            }
            channel.close();
            conn.close();
        }
    }

在事务模式里面，只有收到了服务端的 Commit-OK的指令，才能提交成功。所以可以解决生产者和服务端确认的问题。但是事务模式有一个缺点，它是**阻塞的**，一条消息没有发送完毕，不能发送下一条消息，它可能会榨干RabbitMQ服务器的性能。所以不建议在生产环境使用。

使用SpringAMQP时，在构造RabbitTemplate的Bean时设置，因为RabbitTemplate封装了channel

rabbitTemplate.setChannelTransacted(true);

那么有没有其他可以保证消息被Broker接收，但是又不大量消耗性能的方式呢？这个就是第二种模式，叫做确认（Confirm）模式。

## 2.确认模式ACK ##

生产者通过调用 channel 的 confirmSelect 方法将 channel 设置为 confirm 模式。该模式下，所有在该信道上发布的消息都会被分派一个唯一的ID(从1开始)，当消息被投递到所有匹配的队列后，broker 就会发送一个(包含消息的唯一 ID 的)确认给发送端，如果 RabbitMQ 因为自身内部错误导致消息丢失，就会发送一条nack消息，发送端的 Confirm Listener 会去监听应答。

> broker回传给发送端的确认消息中，包含了 deliver-tag （确认消息的ID） 和 此外 basic.ack 的 multiple 域（表示到这个ID之前的所有消息都已经得到了处理）。

确认模式（Confirm）有三种具体实现：

### **1.1 普通确认模式** ###

在生产者这边通过调用`channel.confirmSelect()`方法将信道设置为Confirm模式，然后发送消息。一旦消息被投递到所有匹配的队列后，RabbitMQ 就会发送一个确认（Basic.Ack）给生产者，也就是调用 `channel.waitForConfirms()`返回 true，这样生产者就知道消息被服务端接收了。

public class NormalConfirmProducer {
        
    
        private final static String QUEUE_NAME = "ORIGIN_QUEUE";
    
        public static void main(String[] args) throws Exception {
        
            ConnectionFactory factory = new ConnectionFactory();
            factory.setUri(ResourceUtil.getKey("rabbitmq.uri"));
    
            // 建立连接
            Connection conn = factory.newConnection();
            // 创建消息通道
            Channel channel = conn.createChannel();
    
            String msg = "Hello world, Rabbit MQ ,Normal Confirm";
            // 声明队列（默认交换机AMQP default，Direct）
            // String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    
            // 开启发送方确认模式
            channel.confirmSelect();
    
            channel.basicPublish("", QUEUE_NAME, null, msg.getBytes());
            // 普通Confirm，发送一条，确认一条
            if (channel.waitForConfirms()) {
        
                System.out.println("消息发送成功" );
            }
    
            channel.close();
            conn.close();
        }
    }

这种发送1条确认1条的方式消息还不是太高，所以还有一种批量确认的方式。

### **1.2 批量确认模式** ###

批量确认就是在开启Confirm模式后，先发送一批消息。只要`channel.waitForConfirmsOrDie()`方法没有抛出异常，就代表消息都被服务端接收了。

批量确认的方式比单条确认的方式效率要高，但是对于不同的业务，到底发送多少条消息确认一次？

*  数量太少，效率提升不上去。
 *  数量多的话，又会带来另一个问题，比如我们发1000条消息才确认一次，如果前面999 条消息都被服务端接收了，如果第1000条消息被拒绝了，那么前面所有的消息都要重发。

public class BatchConfirmProducer {
        
        private final static String QUEUE_NAME = "ORIGIN_QUEUE";
    
        public static void main(String[] args) throws Exception {
        
            ConnectionFactory factory = new ConnectionFactory();
            factory.setUri(ResourceUtil.getKey("rabbitmq.uri"));
    
            // 建立连接
            Connection conn = factory.newConnection();
            // 创建消息通道
            Channel channel = conn.createChannel();
    
            String msg = "Hello world, Rabbit MQ ,Batch Confirm";
            // 声明队列（默认交换机AMQP default，Direct）
            // String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    
            try {
        
                channel.confirmSelect();
                for (int i = 0; i < 5; i++) {
        
                    // 发送消息
                    // String exchange, String routingKey, BasicProperties props, byte[] body
                    channel.basicPublish("", QUEUE_NAME, null, (msg +"-"+ i).getBytes());
                }
                // 批量确认结果，ACK如果是Multiple=True，代表ACK里面的Delivery-Tag之前的消息都被确认了
                // 比如5条消息可能只收到1个ACK，也可能收到2个（抓包才看得到）
                // 直到所有信息都发布，只要有一个未被Broker确认就会IOException
                channel.waitForConfirmsOrDie();
                System.out.println("消息发送完毕，批量确认成功");
            } catch (Exception e) {
        
                // 发生异常，可能需要对所有消息进行重发
                e.printStackTrace();
            }
    
            channel.close();
            conn.close();
        }
    }

有没有一种方式，可以一边发送一边确认的呢？这个就是异步确认模式

### **1.3 异步确认模式** ###

异步确认模式需要添加一个 ConfirmListener，并且用一个 SortedSet 来维护没有被确认的消息。

public class AsyncConfirmProducer {
        
        private final static String QUEUE_NAME = "ORIGIN_QUEUE";
    
        public static void main(String[] args) throws Exception {
        
            ConnectionFactory factory = new ConnectionFactory();
            factory.setUri(ResourceUtil.getKey("rabbitmq.uri"));
    
            // 建立连接
            Connection conn = factory.newConnection();
            // 创建消息通道
            Channel channel = conn.createChannel();
    
            String msg = "Hello world, Rabbit MQ, Async Confirm";
            // 声明队列（默认交换机AMQP default，Direct）
            // String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    
            // 用来维护未确认消息的deliveryTag
            final SortedSet<Long> confirmSet = Collections.synchronizedSortedSet(new TreeSet<Long>());
    
            // 这里不会打印所有响应的ACK；ACK可能有多个，有可能一次确认多条，也有可能一次确认一条
            // 异步监听确认和未确认的消息
            // 如果要重复运行，先停掉之前的生产者，清空队列
            channel.addConfirmListener(new ConfirmListener() {
        
            	// 处理未确认的消息
    		    // deliverTag：交付标签，标识服务端处理到哪条消息了
    		    // multiple：是否批量处理模式
                public void handleNack(long deliveryTag, boolean multiple) throws IOException {
        
                    System.out.println("Broker未确认消息，标识：" + deliveryTag);
                    if (multiple) {
        
                        // headSet表示后面参数之前的所有元素，全部删除
                        confirmSet.headSet(deliveryTag + 1L).clear();
                    } else {
        
                        confirmSet.remove(deliveryTag);
                    }
                    // 这里添加重发的方法
                }
    			
    			// 处理已确认的消息
    		    // multiple如果true，表示批量执行了deliveryTag这个值以前（小于deliveryTag的）的所有消息，
    		    // 如果为false的话表示单条确认
                public void handleAck(long deliveryTag, boolean multiple) throws IOException {
        
                    // 如果true表示批量执行了deliveryTag这个值以前（小于deliveryTag的）的所有消息，如果为false的话表示单条确认
                    System.out.println(String.format("Broker已确认消息，标识：%d，多个消息：%b", deliveryTag, multiple));
                    if (multiple) {
        
                        // headSet表示后面参数之前的所有元素，全部删除
                        confirmSet.headSet(deliveryTag + 1L).clear();
                    } else {
        
                        // 只移除一个元素
                        confirmSet.remove(deliveryTag);
                    }
                    System.out.println("未确认的消息:"+confirmSet);
                }
            });
    
            // 开启发送方确认模式
            channel.confirmSelect();
            for (int i = 0; i < 10; i++) {
        
            	// 获取消息的唯一ID，之后要加入sortedSet
                long nextSeqNo = channel.getNextPublishSeqNo();
                // 发送消息
                // String exchange, String routingKey, BasicProperties props, byte[] body
                channel.basicPublish("", QUEUE_NAME, null, (msg +"-"+ i).getBytes());
                confirmSet.add(nextSeqNo);
            }
            System.out.println("所有消息:"+confirmSet);
    
            // 这里注释掉的原因是如果先关闭了，可能收不到后面的ACK
            //channel.close();
            //conn.close();
        }
    }

比如我们上面发送了10条消息，分别运行了两次，得到的结果如下：![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNTkyNw_size_16_color_FFFFFF_t_70_pic_center][]  
可以看到第一遍是异步确认了3次，第二遍运行是异步确认了6次。

上面演示的是amqp-client原生api，Confirm模式是在Channel上开启的；而Spring AMQP中RabbitTemplate对Channel的Confirm回调进行了封装，叫做ConfimrCallback。

// 构建RabbitTemplate的Bean时配置
    @Bean 
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
        
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
    
        rabbitTemplate.setChannelTransacted(true);
    	
        // 当消息成功到达exchange，但是没有队列与之绑定的时候触发的ack回调
        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
        
            public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        
                if (!ack) {
        
                    System.out.println("发送消息失败：" + cause);
                    throw new RuntimeException("发送异常：" + cause);
                }
            }
        });
    
        return rabbitTemplate;
    }

[RabbitMQ]: https://blog.csdn.net/weixin_43935927/article/details/109742645
[RabbitMQ_Producer-_Broker]: https://blog.csdn.net/weixin_43935927/article/details/114199354
[RabbitMQ_Exchange-_Queue]: https://editor.csdn.net/md/?articleId=114199350
[RabbitMQ_Queue]: https://blog.csdn.net/weixin_43935927/article/details/114199372
[RabbitMQ_Queue_Consumer]: https://blog.csdn.net/weixin_43935927/article/details/114199379
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNTkyNw_size_16_color_FFFFFF_t_70_pic_center]: /images/20221024/9baf063c9d6c45448aa957f5a90dd1d9.png