精通RabbitMQ之消息确认回执和拒绝

野性酷女 2022-04-03 15:07 676阅读 0赞

### 精通RabbitMQ之消息确认回执和拒绝 ###

*  消费消息确认
 *   *  显式确认模式
     *   *  确认消息API
         *  拒绝消息API
 *  发送消息确认
 *   *  事务
     *  confirm

# 消费消息确认 #

从安全角度考虑，网络是不可靠的，消费者是有可能在处理消息的时候失败。而我们总是希望我们的消息不能因为处理失败而丢失，基于此原因，rabbitmq提供了一个消息确认（message acknowledgements） 的概念：当一个消息从队列中投递给消费者（consumer）后，消费者会通知一下消息中间件（rabbitmq），这个可以是系统自动autoACK的也可以由处理消息的应用操作。

当 **“消息确认”** 被启用的时候，rabbitmq不会完全将消息从队列中删除，直到它收到来自消费者的确认回执（acknowledgement）。

为了解决这个问题，rabbitmq提供了2种处理模式来解决这个问题：

*  **自动确认模式**（automatic acknowledgement model）：当RabbbitMQ将消息发送给应用后，消费者端自动回送一个确认消息。（使用AMQP方法：basic.deliver或basic.get-ok）。
 *  **显式确认模式**（explicit acknowledgement model）：RabbbitMQ不会完全将消息从队列中删除，直到消费者发送一个确认回执（acknowledgement）后再删除消息。（使用AMQP方法：basic.ack）。

## 显式确认模式 ##

在显式确认模式下，消费者可以自由选择什么时候发送确认回执（acknowledgement）。消费者可以在收到消息后立即发送，或将未处理的消息存储后发送，或等到消息被处理完毕后再发送确认回执。

如果一个消费者在尚未发送确认回执的情况下挂掉了，那rabbitmq会将消息重新投递给另一个消费者。如果当时没有可用的消费者了，消息代理会死等下一个注册到此队列的消费者，然后再次尝试投递。

消费者在获取队列消息时，可以指定autoAck参数，采用显式确认模式，需要指定`autoAck = flase`，在显式确认模式，RabbitMQ不会为未ack的消息设置超时时间，它判断此消息是否需要重新投递给消费者的唯一依据是消费该消息的消费者连接是否已经断开。如果断开连接，RabbitMQ也没有收到ACK，则Rabbit MQ会安排该消息重新进入队列，等待投递给下一个消费者。

在显式确认模式，确认回执的案例如下：

//设置非自动回执
    boolean autoAck = false;
    Channel finalChannel = channel;
    try { 
        channel.basicConsume("test", autoAck, "test-consumer-tag",
            new DefaultConsumer(finalChannel) { 
                @Override
                public void handleDelivery(String consumerTag,
                               Envelope envelope,
                               AMQP.BasicProperties properties,
                               byte[] body)
                throws IOException
                { 
    	            //发布的每一条消息都会获得一个唯一的deliveryTag，deliveryTag在channel范围内是唯一的 
                    long deliveryTag = envelope.getDeliveryTag();
                    //第二个参数是批量确认标志。如果值为true，则执行批量确认，此deliveryTag之前收到的消息全部进行确认; 如果值为false，则只对当前收到的消息进行确认
                    finalChannel.basicAck(deliveryTag, true);
                }
           });
    } catch (IOException e) { 
         e.printStackTrace();
    }

上面我们显式的成功回执了我们的消息，但是假如我们发现我们的消费者处理不了这个消息需要其他的消费者处理怎么办呢，我们还可以**拒绝消息**。

### 确认消息API ###

/** * Acknowledge one or several received * messages. * @param deliveryTag 发布的每一条消息都会获得一个唯一的deliveryTag，deliveryTag在channel范围内是唯一的 * @param multiple 批量确认标志。如果值为true，则执行批量确认，此deliveryTag之前收到的消息全部进行确认; 如果值为false，则只对当前收到的消息进行确认 * @throws java.io.IOException if an error is encountered */
        void basicAck(long deliveryTag, boolean multiple) throws IOException;

### 拒绝消息API ###

/** * Reject a message. * @param deliveryTag 发布的每一条消息都会获得一个唯一的deliveryTag，deliveryTag在channel范围内是唯一的 * @param requeue 是否重回队列 如果值为true，则重新放入RabbitMQ的发送队列，如果值为false，则通知RabbitMQ销毁这条消息 * @throws java.io.IOException if an error is encountered */
        void basicReject(long deliveryTag, boolean requeue) throws IOException;

有没有发现，我们似乎只能拒绝一条消息，后面rabbitMQ又补充了basicNack一次对多条消息进行拒绝

/** * Reject one or several received messages. * @param deliveryTag 发布的每一条消息都会获得一个唯一的deliveryTag，deliveryTag在channel范围内是唯一的 * @param multiple 批量确认标志。如果值为true，则执行批量确认，此deliveryTag之前收到的消息全部进行拒绝; 如果值为false，则只对当前收到的消息进行拒绝 * @param requeue 是否重回队列 如果值为true，则重新放入RabbitMQ的发送队列，如果值为false，则通知RabbitMQ销毁这条消息 * @throws java.io.IOException if an error is encountered */
        void basicNack(long deliveryTag, boolean multiple, boolean requeue)
                throws IOException;

这里我们需要注意一下，如果我们的队列目前只有一个消费者，请注意**不要拒绝消息并放回队列导致消息在同一个消费者身上无限循环无法消费的情况发生。**

# 发送消息确认 #

我们除了要考虑消费者消息失败可能失败的情况，我们还需要考虑，消息的发布者在将消息发送出去之后，消息到底有没有正确到达消息中间件呢，如果没到达，我们又需要怎么处理呢？

RabbitMQ为我们提供了两种方式来解决这个问题：

*  **事务**：通过AMQP事务机制实现，这也是AMQP协议层面提供的解决方案。（性能太差，基本不用）
 *  **confirm**：通过将channel设置成confirm模式来实现；

## 事务 ##

RabbitMQ事物的处理包装channel调用代码为：

Tx.SelectOk txSelect() throws IOException;
    Tx.CommitOk txCommit() throws IOException;
    Tx.RollbackOk txRollback() throws IOException;

`txSelect`主要用于将当前channel设置成transaction模式，`txCommit`用于提交事务，`txRollback`用于回滚事务。  
只要我们使用过事物的同学都知道，事物没有什么好解释的，我这边就给大家一个案例，不了解事物的同学也不要着急，因为它性能太差，官方主动弃用，所以我们不了解也无所谓，重点关注第二种：

try { 
        // 开启事务
        channel.txSelect();
        // 往test队列中发出一条消息
        channel.basicPublish("test", "test", null, messageBodyBytes);
        // 提交事务
        channel.txCommit();
    } catch (Exception e) { 
        e.printStackTrace();
        // 事务回滚
        try { 
            channel.txRollback();
        } catch (IOException e1) { 
            e1.printStackTrace();
        }
    }

## confirm ##

下面，我们重点说一下confirm机制：

在confirm机制下，我们可以将channel设置成confirm模式，一旦channel进入confirm模式，所有在该channel上面发布的消息都将会被指派一个唯一的ID(从1开始)，一旦消息被投递到所有匹配的队列之后，broker就会发送一个确认给生产者(包含消息的唯一ID)，这就使得生产者知道消息已经正确到达目的队列了，如果消息和队列是可持久化的，那么确认消息会在将消息写入磁盘之后发出，broker回传给生产者的确认消息中delivery-tag域包含了确认消息的序列号，此外broker也可以设置basic.ack的multiple域，表示到这个序列号之前的所有消息都已经得到了处理；

confirm模式最大的好处在于它是异步的，一旦发布一条消息，生产者应用程序就可以在等信道返回确认的同时继续发送下一条消息，当消息最终得到确认之后，生产者应用便可以通过回调方法来处理该确认消息，如果RabbitMQ因为自身内部错误导致消息丢失，就会发送一条nack消息，生产者应用程序同样可以在回调方法中处理该nack消息；

confirm机制和transaction事务模式是不能够共存的，已经处于transaction事务模式的channel不能被设置为confirm模式，同理，反过来也一样。通常我们可以通过调用channel的confirmSelect方法将channel设置为confirm模式。如果没有设置no-wait标志的话，broker会返回confirm.select-ok表示同意生产者当前channel信道设置为confirm模式。

SortedSet<Long> confirmSet = Collections.synchronizedSortedSet(new TreeSet<Long>());
    try { 
    	//设置为confirm模式
        channel.confirmSelect();
    } catch (IOException e) { 
        e.printStackTrace();
    }
    //设置监听
    channel.addConfirmListener(new ConfirmListener() { 
    
        //处理消息成功回执
       @Override
       public void handleAck(long deliveryTag, boolean multiple) throws IOException { 
            System.out.println("Ack, SeqNo: " + deliveryTag + ", multiple: " + multiple);
            if (multiple) { 
                confirmSet.headSet(deliveryTag + 1).clear();
            }else { 
                confirmSet.remove(deliveryTag);
            }
        }
    
         //处理失败回执
        @Override
        public void handleNack(long deliveryTag, boolean multiple) throws IOException { 
            System.out.println("Nack, SeqNo: " + deliveryTag + ", multiple: " + multiple);
            if (multiple) { 
                confirmSet.headSet(deliveryTag + 1).clear();
            } else { 
                confirmSet.remove(deliveryTag);
            }
        }
    
    });
    
    for(int i =0;i<100;i++) { 
        // 查看下一个要发送的消息的序号
        long nextSeqNo = channel.getNextPublishSeqNo();
        try { 
            channel.basicPublish("test", "test", MessageProperties.PERSISTENT_TEXT_PLAIN, messageBodyBytes);
        } catch (IOException e) { 
            e.printStackTrace();
        }
        confirmSet.add(nextSeqNo);
    }