RabbitMQ 发送确认(Publisher Confirms)特性

￡神魔★判官ぃ 2023-06-09 10:16 21阅读 0赞

# 摘要 #

RabbitMQ的发布者确认(Publisher Confirms)特性用以实现消息的可靠投递。当在某个通道(channel)上开启发布确认后，客户端发布的消息会被MQ服务器(broker)异步的确认。本篇主要介绍3个基于发布确认特性，保证被发布消息安全到达MQ服务器的方案，并比较各自优缺点。

# 开启发布者确认 #

发布者确认是RabbitMQ对AMQP 0.9.1协议的扩展，因此默认情况下未启用它们。发布者确认在通道级别使用`confirmSelect`方法启用：

Channel channel = connection.createChannel();
    channel.confirmSelect();

注意，每个需要开启的发布者确认的通道都必须调用此方法。开启后，发布消息时不需要再调用。

# 方案1：单条消息发布确认 #

这个方案是最简单明了的实现，即发布消息并同步等待其确认：

while (thereAreMessagesToPublish()) { 
        byte[] body = ...;
        BasicProperties properties = ...;
        channel.basicPublish(exchange, queue, properties, body);
        // uses a 5 second timeout
        channel.waitForConfirmsOrDie(5_000);
    }

在发送一条消息后，我们通过`Channel＃waitForConfirmsOrDie（long）`方法等待该消息的确认。在确认消息后，该方法将立即返回；如果未在超时时间内确认该消息或该消息没有被确认（这意味着MQ服务器出于某种原因无法处理该消息），则该方法将抛出异常。异常的处理通常包括记录错误消息和重试发送消息。  
此方案实现非常简单，容易理解，但有个主要缺点：由于确认方法阻塞了其他消息的发送，会显著降低消息发送的吞吐量。如果你的应用场景对消息吞吐量的要求不高（每秒不过百），那推荐使用这个方案。

> 发布者确认到底是同步的还是异步的？  
> 可以把`waitForConfiirmOrDie`方法看做是一个同步工具，它让客户端可以同步的方式执行。但背后是一个异步的过程，客户端是异步的接受MQ服务器的确认通知。

# 方案2： 批次发布消息并确认 #

为了提高吞吐量，我们可以先发送一个批次的消息，然后等待整个批次消息的确认：

int batchSize = 100;
    int outstandingMessageCount = 0;
    while (thereAreMessagesToPublish()) { 
        byte[] body = ...;
        BasicProperties properties = ...;
        channel.basicPublish(exchange, queue, properties, body);
        outstandingMessageCount++;
        if (outstandingMessageCount == batchSize) { 
            ch.waitForConfirmsOrDie(5_000);
            outstandingMessageCount = 0;
        }
    }
    if (outstandingMessageCount > 0) { 
        ch.waitForConfirmsOrDie(5_000);
    }

上面例子中，没发送100条消息后，调用一次`waitForConfirmOrDie`来确认所有未确认过的消息。对比方案1，该方案可以显著提升吞吐量（对于远程MQ节点，最多可提升20-30倍），但也存在一个缺点：如果某条消息投递失败而抛出异常了，我们不知道到底是哪条消息出的问题。

# 方案3：异步的处理消息确认 #

MQ服务器是异步的确认消息的，我们只需要在客户端注册一个回调来处理确认通知：

Channel channel = connection.createChannel();
    channel.confirmSelect();
    channel.addConfirmListener((sequenceNumber, multiple) -> { 
        // code when message is confirmed
    }, (sequenceNumber, multiple) -> { 
        // code when message is nack-ed
    });

需要注册2个回调：一个处理成功的消息，一个处理失败的消息。每个回调都有两参数：

*  sequenceNumber：标识已确认或未确认消息的数字。后面会说明如何将其与已发布的消息相关联。
 *  multiple：这是一个布尔值。如果为false，则只确认一条消息；如果为true，则确认所有序列号小于等于sequenceNumber的消息。

可以在发布之前使用`Channel＃getNextPublishSeqNo（）`获得序列号：

int sequenceNumber = channel.getNextPublishSeqNo());
    ch.basicPublish(exchange, queue, properties, body);

使消息与序列号相关联的一种简单方法是使用映射。假设我们要发布的消息时字符串。这是一个使用映射将发布序列号与消息的字符串主体相关联的代码示例：

ConcurrentNavigableMap<Long, String> outstandingConfirms = new ConcurrentSkipListMap<>();
    // ... code for confirm callbacks will come later
    String body = "...";
    outstandingConfirms.put(channel.getNextPublishSeqNo(), body);
    channel.basicPublish(exchange, queue, properties, body.getBytes());

现在发布代码使用map来追踪发布出去的消息。确认的结果是消息发布成功，我们需要将map清空，但如果是发布失败，需要做些记录错误日志等操作：

ConcurrentNavigableMap<Long, String> outstandingConfirms = new ConcurrentSkipListMap<>();
    ConfirmCallback cleanOutstandingConfirms = (sequenceNumber, multiple) -> { 
        if (multiple) { 
            ConcurrentNavigableMap<Long, String> confirmed = outstandingConfirms.headMap(
              sequenceNumber, true
            );
            confirmed.clear();
        } else { 
            outstandingConfirms.remove(sequenceNumber);
        }
    };
    
    channel.addConfirmListener(cleanOutstandingConfirms, (sequenceNumber, multiple) -> { 
        String body = outstandingConfirms.get(sequenceNumber);
        System.err.format(
          "Message with body %s has been nack-ed. Sequence number: %d, multiple: %b%n",
          body, sequenceNumber, multiple
        );
        cleanOutstandingConfirms.handle(sequenceNumber, multiple);
    });
    // ... publishing code

上面代码包含一个清理未确认map的回调。注意这个回调对于各种multiple取值都可以处理。当客户端收到确认消息  
时回调被执行。处理消息发送失败的回调，获取消息内容并打印错误日志。然后使用第一个回调的方法清空未确认消息map.

> 在回调中重发消息？  
> 你可能觉得在处理发布失败确认通知的回调函数中，直接重新发布消息很合理。其实这是错误的方式，因为在客户端发布消息的线程和执行回调函数的线程是分离的，在后者中通道(channel)不应该执行操作。  
> 正确的重发方法是将发布失败的消息放入一个队列中（推荐使用`ConcurrentLinkedQueue`）。由发布消息的线程从队列中取消息发布。

# 总结 #

一些应用需要保证发布的消息成功投递到MQ服务器。RabbitMQ的发布者确认特性可满足这个需求。发布者确认本质上是个异步的过程，但提供同步的处理方式。实现消息确认的方案需要结合具体场景和应用的限制，典型的方案有这些：

<table> 
 <thead> 
  <tr> 
   <th>方案</th> 
   <th>特点</th> 
   <th>限制</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>单条发布，同步等待确认</td> 
   <td>实现简单</td> 
   <td>吞吐量低</td> 
  </tr> 
  <tr> 
   <td>批次发布，同步等待批次确认</td> 
   <td>实现简单，吞吐量高</td> 
   <td>难以定位问题</td> 
  </tr> 
  <tr> 
   <td>异步处理确认</td> 
   <td>吞吐量高，能够定位问题</td> 
   <td>实现较复杂</td> 
  </tr> 
 </tbody> 
</table>

对于使用远程MQ服务器的吞吐量测试结果：

Published 50,000 messages individually in 231,541 ms
    Published 50,000 messages in batch in 7,232 ms
    Published 50,000 messages and handled confirms asynchronously in 6,332 ms

# 参考资料 #

[Publisher Confirms][]

[Publisher Confirms]: https://www.rabbitmq.com/tutorials/tutorial-seven-java.html