RabbitMQ------发布确认（单个确认、批量确认、未确认）（四）

超、凢脫俗 2024-04-01 15:20 61阅读 0赞

## RabbitMQ------发布确认（四） ##

### 发布确认原理 ###

生产者将信道设置为confirm模式，一旦信道进入confirm模式，所有再该信道上面发布的消息都会被指派一个唯一的ID（从1开始），一旦消息被投递到所有匹配的队列后，broker就会发送一个确认给生产者（包含消息的唯一ID），这就使得生产者知道消息已经正确到达目的地的队列了，如果消息和队列是可以持久化的，那么确认消息就会将消息写入磁盘之后发出，broker回传给生产者的确认消息中deliver-tag域包含了确认消息的序列号，此外broker也可以设置basic，ack的multiple域，标识到这个序列号之前所有的消息都已经得到了处理。  
confirm模式最大的好处在于，它是异步的，一旦发布一条消息，生产者应用程序就可以在等信道返回确认的同时，继续发送下一条消息，当消息最终端得到确认之后，生产者应用便可以通过回调方法来处理该确认消息，如果RabbitMQ因为自身内部错误导致消息丢失，就会发送一条nack消息，生产者应用程序同样可以再回调方法中处理该nack消息。  
（**生产者将消息发送到队列中，消息写在磁盘上，达到持久化的目标，RabbitMQ再给消息发送消息确认，才能够达到消息永不丢失的目的**）

### 发布确认策略 ###

#### 1.开启发布确认的方法 ####

发布确认默认没有开始，如果要开启需要调用confirmSelect方法，每当想要使用发布确认，都需要在channel上调用该方法。

//一个连接中有多个信道
            //获取信道
            Channel channel = connection.createChannel();
            //开启发布确认
            channel.confirmSelect();

#### 2. 单个确认发布、批量确认发布、异步确认发布（3种确认发布策略） ####

#### 2.1 单个确认发布 ####

简单的确认方式，它是一种**同步确认发布模式**。意味着生产者发布一个消息后，需等待确认结果，才能够发布下一个消息。  
缺点：效率低，发布慢。  
开启确认模式。

//开启确认模式
            channel.confirmSelect();

等待确认结果。

boolean b = channel.waitForConfirms();

单个确认模式，发布一千条消息结果

//单个确认,结果耗时722ms
        public static  void publishMessageIndividually() throws Exception {
        
            Channel channel = RabbitMqUtiles.getChannel();
            //队列声明
            String queueName = UUID.randomUUID().toString();
            channel.queueDeclare(queueName,true,false,false,null);
            //开启确认模式
            channel.confirmSelect();
            //开始时间
            long beginTime = System.currentTimeMillis();
            for (int i = 0;i< 1000 ;i++){
        
                String mes = i + "";
                channel.basicPublish("",queueName,null,mes.getBytes());
                //单个消息就马上进行发布确认
                boolean b = channel.waitForConfirms();
                if (b){
        
                    System.out.println("消息发布确认成功");
                }
            }
            //结束时间
            long endTime = System.currentTimeMillis();
            System.out.println("发布消息1000条单独确认消息,用时："+(endTime-beginTime));
        }

#### 2.2 批量确认发布 ####

缺点：当发生故障，导致发布出现问题时，不知道是哪个消息出现问题。并且这种方案仍然是同步的，一样阻塞消息的发布。  
优点：性能比单个确认发布高一点。  
隔100条确认一次，  
代码相对单笔确认，只修改了确认方法的调用逻辑位置

waitForConfirms

代码：

//批量确认,每100条确认一次  147ms
        public static  void publishMessageBatch() throws Exception {
        
            Channel channel = RabbitMqUtiles.getChannel();
            //队列声明
            String queueName = UUID.randomUUID().toString();
            channel.queueDeclare(queueName,true,false,false,null);
            //开启确认模式
            channel.confirmSelect();
            //开始时间
            long beginTime = System.currentTimeMillis();
            int y =  100;
            for (int i = 0;i< 1000 ;i++){
        
                String mes = i + "";
                channel.basicPublish("",queueName,null,mes.getBytes());
                if (i%y ==  0){
        
                    //确认发布
                    boolean b = channel.waitForConfirms();
                    if (b){
        
                        System.out.println("消息发布确认成功");
                    }
                }
            }
            //结束时间
            long endTime = System.currentTimeMillis();
            System.out.println("发布消息1000条批量确认消息,用时："+(endTime-beginTime));
        }

#### 2.3 异步确认发布 ####

异步确认虽然编程逻辑比其他两个复杂，但是性价比最高。它是利用回调函数，来达到消息可靠性传递，这个中间件也是通过函数来保证是否投递成功。  
broker：消息的实体，包含交换机和队列等。异步通知  
无论消息队列是否收到，都会对生产者进行应答。确认应答、未确认应答。  
需要开启对消息的监听。

//添加异步确认,消息监听器
    //ConfirmCallback ackCallback, ConfirmCallback nackCallback
     /**
     *  ackCallback 监听哪些消息确认成功了
      *  nackCallback 监听哪些消息确认失败了
         */
     channel.addConfirmListener(ackCallback,nackCallback);

并且实现消息回调的方法。

//确认消息，回调处理
            /**
             * deliveryTag  消息的标价
             * multiple 是否为批量确认
             */
            ConfirmCallback ackCallback = (long deliveryTag, boolean multiple)->{
        
                System.out.println("消息确认"+deliveryTag);
            };
            //未确认消息，回调处理
            ConfirmCallback nackCallback = (long deliveryTag, boolean multiple) ->{
        
                System.out.println("消息未确认"+deliveryTag);
            };

实例代码：

#### 如何处理未确认消息处理 ####

最好的解决方案，**将未确认消息放在一个基于内存的，能被发布线程访问的队列**，比如说ConcurrentLinkedQueue这个队列（并发链路式队列），confirm callbacks（确认回调）与发布线程之间进行消息传递。它可以在确认回调与发布线程之间进行消息传递。  
此处是使用ConcurrentSkipListMap，用来记录消息与消息的标识。  
两步：  
1.发送消息时，将消息和消息标记，添加到map中  
2.确认成功后，将消息移除，则容器中剩下的都是未确认的消息。  
代码示例：

//异步确认 62ms
        public static  void publishMessageAsync() throws Exception {
        
            Channel channel = RabbitMqUtiles.getChannel();
            //队列声明
            String queueName = UUID.randomUUID().toString();
            channel.queueDeclare(queueName,true,false,false,null);
            //开启确认模式
            channel.confirmSelect();
    
            /**
             * 准备一个线程 安全、有序的哈希表  适用于高并发的情况下
             * 1.轻松地将序号与消息进行关联
             * 2.轻松的批量删除条目，只要给序号
             * 3.支持高并发（多线程）
              */
            ConcurrentSkipListMap<Long,String> skipListMap = new ConcurrentSkipListMap();
    
            //确认消息，回调处理
            /**
             * deliveryTag  消息的标价
             * multiple 是否为批量确认
             */
            ConfirmCallback ackCallback = (long deliveryTag, boolean multiple)->{
        
                //2.删除掉已经确认的消息
                if (multiple){
        
                    ConcurrentNavigableMap<Long, String> confiemed = skipListMap.headMap(deliveryTag);
                    confiemed.clear();
                }else {
        
                    skipListMap.remove(deliveryTag);
                }
                System.out.println("消息确认"+deliveryTag);
            };
            //未确认消息，回调处理
            ConfirmCallback nackCallback = (long deliveryTag, boolean multiple) ->{
        
                String s = skipListMap.get(deliveryTag);
                System.out.println("消息未确认"+deliveryTag);
                System.out.println("消息未确认"+s);
            };
            //添加异步确认,消息监听器
            //ConfirmCallback ackCallback, ConfirmCallback nackCallback
            /**
             *  ackCallback 监听哪些消息确认成功了
             *  nackCallback 监听哪些消息确认失败了
             */
            channel.addConfirmListener(ackCallback,nackCallback);
            //开始时间
            long beginTime = System.currentTimeMillis();
            for (int i = 0;i< 1000 ;i++){
        
                String mes = i + "";
                channel.basicPublish("",queueName,null,mes.getBytes());
                //1.此处记录下所有要发送的消息，消息总和
                skipListMap.put(channel.getNextPublishSeqNo(), mes);
            }
    
            //结束时间
            long endTime = System.currentTimeMillis();
            System.out.println("发布消息1000条批量确认消息,用时："+(endTime-beginTime));
        }

### 结论 ：建议使用异步确认方式，单个确认耗时太久，批量确认无法判断具体未确认消息节点。 ###

核心方法：开启发布确认（ channel.confirmSelect();）