RabbitMQ学习笔记 - 发送方确认机制

墨蓝 2022-01-11 05:05 288阅读 0赞

**参考：<<RabbitMQ实战指南>>**

RabbitMQ发送消息可能会遇到一个问题，发送方不知道消息是否真正到达了RabbitMQ。  
在上篇博客中介绍了在AMQP协议层面提供了事务机制解决，但是使用事务机制后性能下降严重，这里还可以使用一种轻量级的方式 ------ 发送方确认机制（publisher confirm）。

生产者将信道设置成confirm（确认）模式，一旦信道进入confirm模式，所有在信道上发送的消息都会被指派一个唯一的ID（从1开始），一旦消息被投递到所有匹配的队列之后，RabbiMQ就会发送一个确认（Basic.Ack）给生产者（包含消息的唯一ID），这就使得生产者知道消息已经正确到达了。如果消息是持久化的，那么确认消息会在消息写入磁盘后发出。

RabbitMQ回传给生产者的确认消息中的deliverTag包含了确认消息的序号，此外RabbitMQ也可以设置channel.basicAck方法中的 **multiple** 参数，表示到这个序号之前的所有消息都已经得到了处理。

发送确认机制流程如下：

*  生产者通过调用 channel.confirmSelect 方法(即 Confirm.Select 命令)将信道设置为confirm模式
 *  RabbitMQ会返回 Confirm.Select-Ok 命令表示同意生产者将当前信道设置为confirm模式
 *  生产者发送消息，RabbitMQ会 ack（Basic.Ack命令）或者nack（Basic.Nack命令）这些消息。注：一条消息要么ack，要么nack
 *  生产者调用channel.waitForConfirms方法，等待RabbitMQ的确认返回；也可以提供一个回调方法处理服务器确认的消息  
    ![发送方确认机制][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L215dHRfMTA1NjY_size_16_color_FFFFFF_t_70]

三种confirm方式：

*  普通confirm方法：每发送一条消息后，调用channel.waitForConfirms方法，等待服务器的确认返回
 *  批量confirm方法：每发送一批消息后，调用channel.waitForConfirms方法，等待服务器的确认返回
 *  异步confirm方法：提供一个回调方法，服务端确认了一条或者多条消息后客户端会回调这个方法进行处理

RabbitMQ客户端中有4个和channel.waitForConfirms() 类似的方法：  
（假如channel没有被置为confirm模式，下面这些方法都会抛出java.lang.IllegalStateException；假如中断则抛出java.lang.InterruptedException）

*  boolean waitForConfirms() throws InterruptedException;
    
     *  等待直到上一批次所有消息被broker Basic.Ack/Basic.Nack
 *  boolean waitForConfirms(long timeout) throws InterruptedException, TimeoutException;
    
     *  timeout表示超时时间，一旦等待RabbitMQ回应超时就会抛出java.util.concurrent.TimeoutException
 *  void waitForConfirmsOrDie() throws IOException, InterruptedException;
    
     *  和waitForConfirms() 类似，只要有任一条消息被Basic.Nack，就会抛出java.io.IOException
 *  void waitForConfirmsOrDie(long timeout) throws IOException, InterruptedException, TimeoutException;
    
     *  和waitForConfirms(long timeout)类似，超时就抛出java.util.concurrent.TimeoutException、只要有任一条消息被Basic.Nack，就会抛出java.io.IOException

（1）普通Confirm方法：

这种方式和事务机制类似，发送一条消息后，立即调用channel.waitForConfirms()方法，之后等待RabbitMQ确认，这实际上也是一种串行同步等待方式。

在同步等待的方式下，publisher confirm机制发送一条消息需要通信交互的命令是2条：Basic.Publish 和 Basic.Ack；事务机制是3条：Basic.Publish、Tx.Commit / Tx.Commit-Ok (或者 Tx.Rollback / Tx.Rollback-Ok)，事务机制多了一个命令帧报文的交互 ，所以QPS会略微下降。

示例代码：

...
    // 将信道置为 publisher confirm 模式
    channel.confirmSelect();
    
    channel.basicPublish(exchangeName, routingKey, MessageProperties.PERSISTENT_TEXT_PLAIN, "Test Msg".getBytes());
    if (!channel.waitForConfirms()) { 
        // 发送失败处理
        System.out.println("send message fail");
    }
    ...

（2）批量Confirm方法：

客户端程序需要定期或者定量（达到多少条)，或者两者结合起来调用 channel.waitForConfirms() 来等待 RabbitMQ 的确认返回。相比于前面示例中的普通confirm方法，批量极大地提升了confirm的效率。

这里需要注意，当出现RabbitMQ返回Basic.Nack或者超时情况时，客户端需要将这一批次的消息全部重发，这会带来重复消息，并且消息经常丢失时，批量 confirm 的性能应该是不升反降的。

（3）异步Confirm方法（建议使用这种方式）：

在客户端Channel接口中提供的addConfirmListener()方法可以添加ConfirmListener这个回调接口，ConfirmListener接口包含两个方法：handleAck() 和 handleNack()，分别用来处理 RabbitMQ 回传的 Basic.Ack 和 Basic.Nack。

参数说明：

*  deliveryTag：在 publisher confirm模式下用来标记消息的唯一有序序号
 *  multiple：消息确认有可能是批量确认的
    
     *  true：批量确认了deliveryTag这个值之前的消息
     *  false：单条确认

...
    // 将信道置为 publisher confirm 模式
    channel.confirmSelect();
    
    List<String> messages = new ArrayList<>(10);
    for (int i = 0; i < 10; i++) { 
        String message = "Test Msg " + i;
        messages.add(message);
        channel.basicPublish(exchangeName, routingKey, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());
    }
    
    channel.addConfirmListener(new ConfirmListener() { 
        @Override
        public void handleAck(long deliveryTag, boolean multiple) throws IOException { 
            System.out.println(String.format("已确认消息，deliveryTag=%d，multiple=%b", deliveryTag, multiple));
        }
    
        @Override
        public void handleNack(long deliveryTag, boolean multiple) throws IOException { 
            System.out.println(String.format("未确认消息，deliveryTag=%d，multiple=%b", deliveryTag, multiple));
            // TODO 发送失败处理，如重新发送
        }
    });
    ...

事务机制和publisher confirm机制对比：

*  事务机制在一条消息发送之后会使发送端阻塞，以等待 RabbitMQ 的回应，之后才能继续发送下一条消息。
 *  事务机制和publisher confirm机制两者是互斥的，不能共存。如果将已开启 **事务模式** 的信道再设直为 publisher confirm 模式， RabbitMQ会报错；同样，将已开启 publisher confirm 模式的信道再设置为事务模式，RabbitMQ也会报错。
 *  发送方确认机制最大的好处在于它是异步的，可以批量发送消息后，等待RabbitMQ确认；也可以使用异步Confirm方式， 生产者通过回调方法来处理该确认消息。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L215dHRfMTA1NjY_size_16_color_FFFFFF_t_70]: /images/20220111/680d5be8015d461e97692c36f6ead581.png