RabbitMQ消息确认

文章目录

• RabbitMQ 消息确认机制详解：原生 Java Client vs Spring AMQP
• • 一、核心概念
  • 二、非 Spring 环境（原生 Java Client）
  • • 2.1 `basicConsume` 的 `autoAck` 参数
    • 2.2 自动确认（`autoAck = true`）
    • 2.3 手动确认（`autoAck = false`）
    • • ① `basicAck` — 确认成功
      • ② `basicReject` — 拒绝单条消息
      • ③ `basicNack` — 拒绝一条或多条消息
    • 2.4 三种手动确认方法对比
  • 三、Spring 环境（Spring AMQP）
  • • 3.1 `AcknowledgeMode.NONE` — 等同自动确认
    • 3.2 `AcknowledgeMode.AUTO` — 默认模式（推荐）
    • 3.3 `AcknowledgeMode.MANUAL` — 完全手动控制
    • • 监听方法注入 `Channel`
      • Spring 手动模式下三种方法说明
      • 使用 `AmqpHeaders.ACKNOWLEDGMENT`（另一种写法）
  • 四、两种环境对比总结
  • 五、注意事项

RabbitMQ 消息确认机制详解：原生 Java Client vs Spring AMQP

在 RabbitMQ 中，消息确认（Acknowledgement） 决定了消息何时从队列中移除。理解这一点，是避免消息丢失的关键。

本文分两部分说明：

• 非 Spring 环境：使用原生 rabbitmq-client 的 basicConsume
• Spring 环境：使用 Spring AMQP 的 AcknowledgeMode

一、核心概念

RabbitMQ 采用 生产者确认（Publisher Confirm） 和 消费者确认（Consumer Ack） 两套机制。本文聚焦 消费者确认：

手动确认更可靠：消费者宕机或处理失败时，未确认的消息会重新投递。

二、非 Spring 环境（原生 Java Client）

2.1 basicConsume 的 autoAck 参数

// 第二个参数 autoAck：
// true = 自动确认（收到即删除）
// false = 手动确认（需主动 ack/nack/reject）
channel.basicConsume(Constants.QUEUE_NAME, false, consumer);

2.2 自动确认（autoAck = true）

消息一旦投递给消费者，RabbitMQ 就将其从队列移除，不管业务是否处理成功。

DefaultConsumer consumer = new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope,
AMQP.BasicProperties properties, byte[] body) {
String msg = new String(body);
System.out.println("收到消息: " + msg);
// 无需手动 ack，消息已被自动确认
}
};

// true 表示自动确认
channel.basicConsume(Constants.QUEUE_NAME, true, consumer);

风险：处理过程中消费者崩溃，消息会永久丢失。

2.3 手动确认（autoAck = false）

开启手动确认后，必须调用以下三种方法之一，否则消息会一直占用消费者，连接关闭后还会被重新投递。

① basicAck — 确认成功

channel.basicAck(deliveryTag, multiple);

DefaultConsumer consumer = new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope,
AMQP.BasicProperties properties, byte[] body) throws IOException {
String msg = new String(body);
try {
System.out.println("处理消息: " + msg);
// 业务处理成功，确认消息
channel.basicAck(envelope.getDeliveryTag(), false);
} catch (Exception e) {
// 处理失败，见下方 reject / nack
}
}
};

channel.basicConsume(Constants.QUEUE_NAME, false, consumer);

② basicReject — 拒绝单条消息

channel.basicReject(deliveryTag, requeue);

// 处理失败，拒绝消息并重新入队
channel.basicReject(envelope.getDeliveryTag(), true);

// 处理失败，拒绝消息并丢弃
channel.basicReject(envelope.getDeliveryTag(), false);

basicReject 只能拒绝 单条 消息。

③ basicNack — 拒绝一条或多条消息

channel.basicNack(deliveryTag, multiple, requeue);

例如105号已经确认成功,但是110号被拒接,那么批量拒绝 106~110号

// 拒绝当前消息，重新入队
channel.basicNack(envelope.getDeliveryTag(), false, true);

// 批量拒绝之前所有未确认消息，不重新入队
channel.basicNack(envelope.getDeliveryTag(), true, false);

2.4 三种手动确认方法对比

三、Spring 环境（Spring AMQP）

Spring 将确认模式封装为 AcknowledgeMode 枚举：

public enum AcknowledgeMode {
NONE, // 自动确认
MANUAL, // 手动确认
AUTO; // Spring 自动确认（默认）
}

3.1 AcknowledgeMode.NONE — 等同自动确认

消息投递给消费者后，Broker 立即删除，与原生 autoAck=true 等价。 yaml文件中配置

spring:
rabbitmq:
listener:
simple:
acknowledge-mode: none

方法

@RabbitListener(queues = "my.queue")
public void listen(String message) {
System.out.println("收到: " + message);
// 无需手动 ack，消息已被自动确认
}

风险：消费者崩溃时消息丢失。

3.2 AcknowledgeMode.AUTO — 默认模式（推荐）

Spring 容器根据监听方法执行结果自动确认：

yaml文件中配置

spring:
rabbitmq:
listener:
simple:
acknowledge-mode: auto

方法

@RabbitListener(queues = "my.queue")
public void listen(String message) {
System.out.println("处理: " + message);
// 方法正常结束 → 自动 ack
// 抛出异常 → 自动 nack 并 requeue
}

也可在注解上单独指定：

@RabbitListener(queues = "my.queue", ackMode = "AUTO")
public void listen(String message) { ... }

3.3 AcknowledgeMode.MANUAL — 完全手动控制

开发者自行决定何时确认，与原生 autoAck=false 等价，可靠性最高。

yaml文件中配置

spring:
rabbitmq:
listener:
simple:
acknowledge-mode: none

监听方法注入 Channel

@RabbitListener(queues = "my.queue", ackMode = "MANUAL")
public void listen(Message message, Channel channel) throws IOException {
long deliveryTag = message.getMessageProperties().getDeliveryTag();
String body = new String(message.getBody());

try {
System.out.println("处理: " + body);
// 业务逻辑…

// ① 确认成功
channel.basicAck(deliveryTag, false);

} catch (BusinessException e) {
// ② 拒绝单条，true重新入队
channel.basicReject(deliveryTag, true);

} catch (FatalException e) {
// ③ 拒绝单条，false丢弃（或进死信队列）
channel.basicReject(deliveryTag, false);

} catch (Exception e) {
// ④ true表示批量拒绝,false表示不重新入队
channel.basicNack(deliveryTag, true, false);
}
}

Spring 手动模式下三种方法说明

deliveryTag 从 message.getMessageProperties().getDeliveryTag() 获取。 Spring 注入的 Channel 是 com.rabbitmq.client.Channel，API 与原生 Client 一致。

使用 AmqpHeaders.ACKNOWLEDGMENT（另一种写法）

@RabbitListener(queues = "my.queue", ackMode = "MANUAL")
public void listen(Message message,
@Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag,
Channel channel) throws IOException {
// 处理成功
channel.basicAck(deliveryTag, false);
}

四、两种环境对比总结

五、注意事项