RabbitMQ消息确认机制：确保消息可靠传递的关键技术

RabbitMQ消息确认机制：确保消息可靠传递的关键技术

在分布式系统和微服务架构中，消息队列扮演着至关重要的角色，而消息确认机制则是保证消息可靠传递的核心技术。RabbitMQ提供了完善的消息确认机制，包括发布确认（Publisher Confirms）和消费确认（Consumer Acknowledgements）两大部分。深入理解这些机制的工作原理，并能在实际项目中正确实现，是构建可靠消息系统的关键所在。本文将从原理、实现、最佳实践等多个维度，全面讲解RabbitMQ的消息确认机制。

一、消息确认机制概述与核心价值

消息确认机制是RabbitMQ保证消息可靠传递的基石。在一个典型的消息传递场景中，消息从生产者出发，经过交换机的路由，到达队列存储，最终被消费者获取并处理。这个过程中任何一个环节出现问题都可能导致消息丢失或重复处理，而消息确认机制正是为解决这些问题而设计的。

消息确认机制的核心价值体现在三个方面：防止消息丢失、避免消息重复处理、以及提供可靠传递的端到端保障。防止消息丢失是最基本的需求，生产者需要确认消息已经成功写入RabbitMQ，消费者需要确认消息已经被正确处理。避免消息重复处理同样重要，因为在网络异常或系统故障的情况下，同一条消息可能被重复投递，可靠的消息系统需要具备幂等性处理能力。端到端保障则是指从生产者到消费者的整个链路都需要有确认机制，任何一个环节的不可靠都会影响整体可靠性。

在RabbitMQ中，消息确认分为两个层面：发布确认和消费确认。发布确认确保消息从生产者成功到达RabbitMQ的交换机，消费确认确保消息从队列被消费者正确处理。这两个层面的确认机制相互配合，共同构建起可靠的消息传递体系。

import com.rabbitmq.client.*;
import java.io.IOException;

public class MessageAcknowledgmentOverview {

public static void main(String[] args) throws Exception {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
factory.setUsername("guest");
factory.setPassword("guest");

Connection connection = factory.newConnection();
Channel channel = connection.createChannel();

// 声明队列
String queueName = "reliable.message.queue";
channel.queueDeclare(queueName, true, false, false, null);

System.out.println("=== RabbitMQ消息确认机制概述 ===");
System.out.println();
System.out.println("1. 发布确认（Publisher Confirms）");
System.out.println(" – 确保消息从生产者成功到达交换机");
System.out.println(" – 需要启用 publisher-confirms 和 publisher-returns");
System.out.println(" – 使用场景：关键业务消息、不可丢失的消息");
System.out.println();
System.out.println("2. 消费确认（Consumer Acknowledgements）");
System.out.println(" – 确保消息被消费者正确处理");
System.out.println(" – 支持手动确认和自动确认两种模式");
System.out.println(" – 使用场景：需要保证消息处理的业务场景");
System.out.println();
System.out.println("3. 确认模式对比");
System.out.println(" – 自动确认：消息投递给消费者后立即确认，存在消息丢失风险");
System.out.println(" – 手动确认：消息处理完成后显式确认，可靠性高");
System.out.println();

channel.close();
connection.close();
}
}

二、发布确认机制深度解析

发布确认（Publisher Confirms）是RabbitMQ提供的一种确认机制，用于确保消息从生产者成功发布到交换机。当启用发布确认后，RabbitMQ会在消息成功写入磁盘或被所有镜像节点接收后，向生产者发送确认响应。如果消息发布失败，RabbitMQ会发送否定确认（NACK），生产者可以根据这个结果进行重试或其他处理。

发布确认有三种模式：同步确认、异步确认和事务模式。同步确认会在每条消息发布后阻塞等待确认，简单但性能较差，适合小消息量的场景。异步确认通过回调机制通知确认结果，性能好但实现相对复杂。事务模式提供最强的一致性保证，但性能损耗严重，官方建议谨慎使用。

启用发布确认需要两步：首先在Channel级别启用确认回调，然后通过添加ConfirmListener来监听确认结果。在实际使用中，推荐使用异步确认模式，因为它既能保证可靠性又能保持良好的性能。

import com.rabbitmq.client.*;
import java.nio.charset.StandardCharsets;
import java.util.UUID;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.TimeUnit;

public class PublisherConfirmsExample {

public static void main(String[] args) throws Exception {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
factory.setUsername("guest");
factory.setPassword("guest");

Connection connection = factory.newConnection();
final Channel channel = connection.createChannel();

// 启用发布确认
channel.confirmSelect();

// 用于存储待确认的消息
final ConcurrentNavigableMap<Long, String> outstandingConfirms =
new ConcurrentSkipListMap<>();

// 设置确认监听器
channel.addConfirmListener(new ConfirmListener() {
@Override
public void handleAck(long deliveryTag, boolean multiple) throws IOException {
if (multiple) {
// 批量确认，删除所有小于等于deliveryTag的消息
ConcurrentNavigableMap<Long, String> confirmed =
outstandingConfirms.headMap(deliveryTag, true);
confirmed.clear();
System.out.println("批量确认消息，数量: " + confirmed.size());
} else {
// 单条确认
outstandingConfirms.remove(deliveryTag);
System.out.println("确认消息: " + deliveryTag);
}
}

@Override
public void handleNack(long deliveryTag, boolean multiple) throws IOException {
if (multiple) {
// 批量否定确认
ConcurrentNavigableMap<Long, String> nacked =
outstandingConfirms.headMap(deliveryTag, true);
System.out.println("批量NACK消息，数量: " + nacked.size());
// 这里需要重发这些消息
nacked.forEach((tag, msg) -> {
try {
System.out.println("重发消息: " + msg);
// 实际实现中需要重发消息
} catch (Exception e) {
e.printStackTrace();
}
});
nacked.clear();
} else {
// 单条否定确认
String message = outstandingConfirms.remove(deliveryTag);
System.out.println("NACK消息: " + message + "，需要重发");
// 实际实现中需要重发消息
}
}
});

// 发布消息示例
String queueName = "publisher.confirms.queue";
channel.queueDeclare(queueName, true, false, false, null);

System.out.println("开始发布消息…");
for (int i = 0; i < 10; i++) {
String message = "Message-" + i + "-" + UUID.randomUUID().toString();
long nextPublishSeqNo = channel.getNextPublishSeqNo();

// 记录待确认的消息
outstandingConfirms.put(nextPublishSeqNo, message);

channel.basicPublish("", queueName,
MessageProperties.PERSISTENT_TEXT_PLAIN,
message.getBytes(StandardCharsets.UTF_8));

System.out.println("发布消息 [" + i + "]: " + message);
}

// 等待所有确认
System.out.println("等待所有消息确认…");
if (channel.waitForConfirms()) {
System.out.println("所有消息确认成功！");
} else {
System.out.println("部分消息确认失败，请检查！");
}

// 使用超时等待
channel.waitForConfirmsOrDie(5, TimeUnit.SECONDS);
System.out.println("所有消息已确认！");

channel.close();
connection.close();
}
}

三、消费确认机制详解

消费确认（Consumer Acknowledgements）是消息从队列投递到消费者后，消费者向RabbitMQ反馈处理结果的机制。与发布确认不同，消费确认更关注消息的实际处理过程，确保每条消息都被正确处理后才从队列中删除。

消费确认有三种模式：自动确认（Auto Ack）、手动确认（Manual Ack）和批量确认。自动确认模式下，消息一旦投递给消费者就立即从队列中删除，如果消费者在处理过程中崩溃，消息就会丢失。手动确认模式下，消息投递后不会立即删除，需要消费者显式调用basic.ack确认处理成功，或者调用basic.nack/basic.reject拒绝消息。批量确认允许消费者在处理一定数量的消息后批量确认，提高效率。

在手动确认模式下，消息的确认有多种策略：basic.ack用于确认消息，可以指定是否批量确认；basic.nack用于否定确认，可以指定是否重新入队；basic.reject用于拒绝单条消息，可以指定是否重新入队。重新入队的消息会放到队列末尾，可能导致消息乱序，在设计系统时需要考虑这一点。

import com.rabbitmq.client.*;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

public class ConsumerAcknowledgmentExample {

public static void main(String[] args) throws Exception {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");

final Connection connection = factory.newConnection();
final Channel channel = connection.createChannel();

// 设置预取数量
int prefetchCount = 10;
channel.basicQos(prefetchCount);

String queueName = "consumer.acks.queue";
channel.queueDeclare(queueName, true, false, false, null);

final AtomicInteger successCount = new AtomicInteger(0);
final AtomicInteger failureCount = new AtomicInteger(0);

System.out.println("=== 手动确认消费示例 ===");
System.out.println("预取数量: " + prefetchCount);
System.out.println();

DeliverCallback deliverCallback = (consumerTag, delivery) -> {
String message = new String(delivery.getBody(), StandardCharsets.UTF_8);
long deliveryTag = delivery.getEnvelope().getDeliveryTag();

System.out.println("收到消息 [" + deliveryTag + "]: " + message);

try {
// 模拟消息处理
processMessage(message);

// 手动确认消息
channel.basicAck(deliveryTag, false);
successCount.incrementAndGet();
System.out.println("消息处理成功，已确认 [" + deliveryTag + "]");

} catch (Exception e) {
System.out.println("消息处理失败: " + e.getMessage());
// 否定确认，消息重新入队
channel.basicNack(deliveryTag, false, true);
failureCount.incrementAndGet();
}
};

CancelCallback cancelCallback = consumerTag -> {
System.out.println("消费者被取消: " + consumerTag);
};

// 开启手动确认消费
boolean autoAck = false;
channel.basicConsume(queueName, autoAck, deliverCallback, cancelCallback);

// 等待一段时间后打印统计
TimeUnit.SECONDS.sleep(30);

System.out.println();
System.out.println("=== 消费统计 ===");
System.out.println("成功处理: " + successCount.get() + " 条");
System.out.println("处理失败: " + failureCount.get() + " 条");

channel.close();
connection.close();
}

private static void processMessage(String message) throws Exception {
// 模拟消息处理逻辑
if (message.contains("error")) {
throw new RuntimeException("模拟处理异常");
}
// 模拟处理时间
Thread.sleep(100);
}
}

四、消息重试与幂等性处理

在实际应用中，消息处理失败是不可避免的，可能是由于网络问题、服务暂时不可用、数据异常等原因。RabbitMQ提供了多种机制来处理消息处理失败的情况，包括消息重试、死信队列等。但无论采用哪种机制，幂等性处理都是必须考虑的问题。

消息重试有两种主要方式：消息重新入队和消息重试次数限制。消息重新入队是将处理失败的消息重新放回队列末尾，等待下次处理。这种方式简单直接，但可能导致消息无限重试，需要配合重试次数限制和死信队列使用。消息重试次数限制可以通过在消息头中添加重试计数器来实现，达到最大重试次数后将消息转入死信队列。

幂等性处理是确保同一条消息被多次处理时产生相同结果的能力。实现幂等性的常用方法包括：使用消息ID进行去重、为每个业务操作生成唯一的事务ID、使用数据库的唯一约束等。在设计消息处理逻辑时，应该始终假设同一条消息可能被处理多次。

import com.rabbitmq.client.*;
import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Map;

public class MessageRetryAndIdempotencyExample {

private static final String QUEUE_NAME = "retry.idempotency.queue";
private static final String DLX_EXCHANGE = "dlx.exchange";
private static final String DLX_QUEUE = "dlx.queue";
private static final int MAX_RETRY_COUNT = 3;

public static void main(String[] args) throws Exception {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");

Connection connection = factory.newConnection();
Channel channel = connection.createChannel();

// 声明死信交换机和队列
channel.exchangeDeclare(DLX_EXCHANGE, BuiltinExchangeType.DIRECT, true);
channel.queueDeclare(DLX_QUEUE, true, false, false, null);
channel.queueBind(DLX_QUEUE, DLX_EXCHANGE, "retry");

// 声明主队列，配置死信交换机
Map<String, Object> args = new HashMap<>();
args.put("x-dead-letter-exchange", DLX_EXCHANGE);
args.put("x-dead-letter-routing-key", "retry");

channel.queueDeclare(QUEUE_NAME, true, false, false, args);

System.out.println("=== 消息重试与幂等性处理示例 ===");
System.out.println("最大重试次数: " + MAX_RETRY_COUNT);
System.out.println();

// 生产者：发送带重试计数器的消息
produceMessages(channel);

// 消费者：实现重试逻辑
consumeWithRetry(channel);

Thread.sleep(5000);

channel.close();
connection.close();
}

private static void produceMessages(Channel channel) throws Exception {
System.out.println("生产者发送测试消息…");
for (int i = 1; i <= 5; i++) {
String message = "TestMessage-" + i;
Map<String, Object> headers = new HashMap<>();
headers.put("x-retry-count", 0);

AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
.deliveryMode(2)
.headers(headers)
.build();

channel.basicPublish("", QUEUE_NAME, props,
message.getBytes(StandardCharsets.UTF_8));
System.out.println("发送消息: " + message);
}
}

private static void consumeWithRetry(Channel channel) throws Exception {
channel.basicQos(1);

DeliverCallback deliverCallback = (consumerTag, delivery) -> {
String message = new String(delivery.getBody(), StandardCharsets.UTF_8);
long deliveryTag = delivery.getEnvelope().getDeliveryTag();

// 获取重试次数
int retryCount = getRetryCount(delivery);
System.out.println("收到消息: " + message + "，重试次数: " + retryCount);

try {
// 模拟处理，第三次尝试才成功
if (retryCount < 2) {
throw new RuntimeException("模拟处理失败");
}

// 幂等性检查
if (isMessageProcessed(message)) {
System.out.println("消息已处理过，跳过: " + message);
} else {
// 处理消息
processMessage(message);
markMessageAsProcessed(message);
System.out.println("消息处理成功: " + message);
}

// 确认消息
channel.basicAck(deliveryTag, false);

} catch (Exception e) {
handleFailure(channel, delivery, retryCount, e);
}
};

channel.basicConsume(QUEUE_NAME, false, deliverCallback,
consumerTag -> {});
}

private static int getRetryCount(Delivery delivery) {
Map<String, Object> headers = delivery.getProperties().getHeaders();
if (headers != null && headers.containsKey("x-retry-count")) {
return ((Number) headers.get("x-retry-count")).intValue();
}
return 0;
}

private static void handleFailure(Channel channel, Delivery delivery,
int retryCount, Exception e) throws Exception {
long deliveryTag = delivery.getEnvelope().getDeliveryTag();

if (retryCount >= MAX_RETRY_COUNT) {
// 超过最大重试次数，发送到死信队列
System.out.println("超过最大重试次数，发送到死信队列: " + e.getMessage());
channel.basicNack(deliveryTag, false, false);
} else {
// 增加重试次数后重新入队
Map<String, Object> headers = new HashMap<>(delivery.getProperties().getHeaders());
headers.put("x-retry-count", retryCount + 1);

AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
.deliveryMode(2)
.headers(headers)
.build();

// 拒绝原消息
channel.basicNack(deliveryTag, false, false);

// 重新发布带重试计数器的消息
channel.basicPublish("", QUEUE_NAME, props, delivery.getBody());
System.out.println("消息重试次数+1，重新入队");
}
}

// 幂等性处理相关方法
private static boolean isMessageProcessed(String message) {
// 实际实现中应查询数据库或缓存
return message.contains("processed");
}

private static void processMessage(String message) throws Exception {
// 实际的消息处理逻辑
Thread.sleep(100);
}

private static void markMessageAsProcessed(String message) {
// 实际实现中应记录到数据库或缓存
System.out.println("标记消息已处理: " + message);
}
}

五、Spring AMQP中的消息确认实现

Spring AMQP提供了更简洁的编程模型来处理消息确认，通过配置和注解可以方便地实现发布确认和消费确认。Spring AMQP的消息确认主要通过以下组件实现：RabbitTemplate用于发布确认，SimpleRabbitListenerContainerFactory用于消费确认，RabbitTemplate和Channel回调用于处理返回的消息。

在Spring Boot中，配置消息确认需要创建相应的配置Bean，然后通过RabbitTemplate设置确认回调，或者通过@RabbitListener注解配合AcknowledgeMode来控制消费确认模式。Spring AMQP还提供了消息转换器（MessageConverter）和错误处理器（ErrorHandler）等组件，可以更方便地处理消息转换和异常情况。

import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class SpringAmqpConfig {

@Bean
public ConnectionFactory connectionFactory() {
CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
connectionFactory.setHost("localhost");
connectionFactory.setUsername("guest");
connectionFactory.setPassword("guest");

// 启用发布确认
connectionFactory.setPublisherConfirmType(CachingConnectionFactory.ConfirmType.CORRELATED);
// 启用发布返回
connectionFactory.setPublisherReturns(true);

return connectionFactory;
}

@Bean
public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
RabbitTemplate template = new RabbitTemplate(connectionFactory);

// 配置JSON消息转换器
template.setMessageConverter(new Jackson2JsonMessageConverter());

// 启用强制返回
template.setMandatory(true);

// 设置确认回调
template.setConfirmCallback((correlationData, ack, cause) -> {
if (ack) {
System.out.println("消息已确认: " +
(correlationData != null ? correlationData.getId() : "unknown"));
} else {
System.out.println("消息未确认，原因: " + cause);
// 可以在这里进行重发或其他处理
}
});

// 设置返回回调
template.setReturnsCallback(returned -> {
System.out.println("消息返回: " +
"\\n路由键: " + returned.getRoutingKey() +
"\\n交换机: " + returned.getExchange() +
"\\n响应码: " + returned.getReplyCode() +
"\\n响应文本: " + returned.getReplyText());
});

return template;
}

@Bean
public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(
ConnectionFactory connectionFactory) {
SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
factory.setConnectionFactory(connectionFactory);
factory.setMessageConverter(new Jackson2JsonMessageConverter());

// 设置确认模式为手动
factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
// 设置预取数量
factory.setPrefetchCount(10);
// 设置并发消费者数
factory.setConcurrentConsumers(3);
factory.setMaxConcurrentConsumers(10);

return factory;
}
}

import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;
import com.rabbitmq.client.Channel;

@Component
public class MessageConsumer {

@RabbitListener(queues = "order.queue")
public void handleOrderMessage(
@Payload OrderMessage order,
@Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag,
@Header(AmqpHeaders.CHANNEL) Channel channel) throws Exception {

try {
System.out.println("收到订单消息: " + order);

// 业务处理
processOrder(order);

// 手动确认
channel.basicAck(deliveryTag, false);
System.out.println("消息已确认: " + deliveryTag);

} catch (Exception e) {
System.out.println("消息处理失败: " + e.getMessage());

// 失败处理策略
if (isRetryableException(e)) {
// 可重试异常，重新入队
channel.basicNack(deliveryTag, false, true);
} else {
// 不可重试异常，拒绝并丢弃
channel.basicNack(deliveryTag, false, false);
// 记录到日志或发送到死信队列
handleFailedMessage(order, e);
}
}
}

@RabbitListener(queues = "notification.queue")
public void handleNotificationMessage(
@Payload NotificationMessage notification,
@Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag,
@Header(AmqpHeaders.CHANNEL) Channel channel) throws Exception {

try {
// 幂等性检查
if (isNotificationProcessed(notification.getNotificationId())) {
System.out.println("通知已处理，跳过: " + notification.getNotificationId());
channel.basicAck(deliveryTag, false);
return;
}

// 发送通知
sendNotification(notification);

// 标记为已处理
markNotificationAsProcessed(notification.getNotificationId());

// 确认消息
channel.basicAck(deliveryTag, false);

} catch (Exception e) {
System.out.println("通知发送失败: " + e.getMessage());
channel.basicNack(deliveryTag, false, true);
}
}

private void processOrder(OrderMessage order) throws Exception {
// 订单处理逻辑
}

private boolean isRetryableException(Exception e) {
// 判断是否为可重试异常
return e instanceof java.net.ConnectException ||
e instanceof java.net.SocketTimeoutException;
}

private void handleFailedMessage(Object message, Exception e) {
// 处理失败消息的逻辑
}

private void sendNotification(NotificationMessage notification) throws Exception {
// 发送通知的逻辑
}

private boolean isNotificationProcessed(String notificationId) {
// 检查通知是否已处理
return false;
}

private void markNotificationAsProcessed(String notificationId) {
// 标记通知为已处理
}
}

六、消息确认的监控与最佳实践

消息确认机制的有效实施需要配合完善的监控和日志体系。在生产环境中，应该监控以下关键指标：发布确认成功率、消费确认成功率、消息重试次数、死信队列消息数量等。这些指标可以帮助我们及时发现系统问题并进行优化。

消息确认的最佳实践包括：始终使用手动确认模式，特别是在处理重要业务消息时；合理设置预取数量（Qos），既保证吞吐量又不导致消息积压；实现幂等性处理，防止消息重复消费；设置合理的消息重试次数和重试间隔；使用死信队列处理无法正常消费的消息；记录完整的消息处理日志，便于问题排查。

import java.util.concurrent.atomic.AtomicLong;

public class MessageConfirmationMetrics {

private final AtomicLong publishedTotal = new AtomicLong(0);
private final AtomicLong publishedSuccess = new AtomicLong(0);
private final AtomicLong publishedFailed = new AtomicLong(0);

private final AtomicLong consumedTotal = new AtomicLong(0);
private final AtomicLong consumedSuccess = new AtomicLong(0);
private final AtomicLong consumedFailed = new AtomicLong(0);

private final AtomicLong retryCount = new AtomicLong(0);
private final AtomicLong deadLetterCount = new AtomicLong(0);

public void recordPublished() {
publishedTotal.incrementAndGet();
}

public void recordPublishSuccess() {
publishedSuccess.incrementAndGet();
}

public void recordPublishFailed() {
publishedFailed.incrementAndGet();
}

public void recordConsumed() {
consumedTotal.incrementAndGet();
}

public void recordConsumeSuccess() {
consumedSuccess.incrementAndGet();
}

public void recordConsumeFailed() {
consumedFailed.incrementAndGet();
}

public void recordRetry() {
retryCount.incrementAndGet();
}

public void recordDeadLetter() {
deadLetterCount.incrementAndGet();
}

public void printMetrics() {
System.out.println("=== 消息确认监控指标 ===");
System.out.println();
System.out.println("【发布统计】");
System.out.println(" 总发布数: " + publishedTotal.get());
System.out.println(" 成功数: " + publishedSuccess.get());
System.out.println(" 失败数: " + publishedFailed.get());
System.out.println(" 发布成功率: " + calculateRate(publishedSuccess.get(), publishedTotal.get()) + "%");
System.out.println();
System.out.println("【消费统计】");
System.out.println(" 总消费数: " + consumedTotal.get());
System.out.println(" 成功数: " + consumedSuccess.get());
System.out.println(" 失败数: " + consumedFailed.get());
System.out.println(" 消费成功率: " + calculateRate(consumedSuccess.get(), consumedTotal.get()) + "%");
System.out.println();
System.out.println("【重试统计】");
System.out.println(" 重试次数: " + retryCount.get());
System.out.println(" 死信消息: " + deadLetterCount.get());
}

private String calculateRate(long success, long total) {
if (total == 0) return "0.00";
return String.format("%.2f", (double) success / total * 100);
}
}

总结

RabbitMQ的消息确认机制是构建可靠消息系统的基础技术。通过发布确认，可以确保消息从生产者成功到达RabbitMQ；通过消费确认，可以确保消息被消费者正确处理。两种确认机制相互配合，提供了端到端的消息可靠性保障。

在实际应用中，需要根据业务需求选择合适的确认模式。对于关键业务消息，应该启用发布确认并使用手动确认模式；对于需要重试的消息，应该实现重试机制并设置最大重试次数；对于可能重复的消息，应该实现幂等性处理。此外，配合完善的监控和日志体系，可以及时发现和处理问题，确保系统的稳定运行。

掌握好消息确认机制的原理和使用方法，并遵循最佳实践进行开发，就能够构建出高可靠性的消息处理系统，为分布式系统和微服务架构提供坚实的技术支撑。