RabbitMQ消息确认

文章目录RabbitMQ 消息确认机制详解原生 Java Client vs Spring AMQP一、核心概念二、非 Spring 环境原生 Java Client2.1 basicConsume 的 autoAck 参数2.2 自动确认autoAck true2.3 手动确认autoAck false① basicAck — 确认成功② basicReject — 拒绝单条消息③ basicNack — 拒绝一条或多条消息2.4 三种手动确认方法对比三、Spring 环境Spring AMQP3.1 AcknowledgeMode.NONE — 等同自动确认3.2 AcknowledgeMode.AUTO — 默认模式推荐3.3 AcknowledgeMode.MANUAL — 完全手动控制监听方法注入 ChannelSpring 手动模式下三种方法说明使用 AmqpHeaders.ACKNOWLEDGMENT另一种写法四、两种环境对比总结五、注意事项RabbitMQ 消息确认机制详解原生 Java Client vs Spring AMQP在 RabbitMQ 中消息确认Acknowledgement决定了消息何时从队列中移除。理解这一点是避免消息丢失的关键。本文分两部分说明非 Spring 环境使用原生rabbitmq-client的basicConsumeSpring 环境使用 Spring AMQP 的AcknowledgeMode一、核心概念RabbitMQ 采用生产者确认Publisher Confirm和消费者确认Consumer Ack两套机制。本文聚焦消费者确认模式含义消息何时从队列移除自动确认消息投递给消费者后Broker 立即删除收到即删除手动确认消费者处理成功后主动通知 Broker调用 ack 后删除手动确认更可靠消费者宕机或处理失败时未确认的消息会重新投递。二、非 Spring 环境原生 Java Client2.1basicConsume的autoAck参数// 第二个参数 autoAck// true 自动确认收到即删除// false 手动确认需主动 ack/nack/rejectchannel.basicConsume(Constants.QUEUE_NAME,false,consumer);2.2 自动确认autoAck true消息一旦投递给消费者RabbitMQ 就将其从队列移除不管业务是否处理成功。DefaultConsumerconsumernewDefaultConsumer(channel){OverridepublicvoidhandleDelivery(StringconsumerTag,Envelopeenvelope,AMQP.BasicPropertiesproperties,byte[]body){StringmsgnewString(body);System.out.println(收到消息: msg);// 无需手动 ack消息已被自动确认}};// true 表示自动确认channel.basicConsume(Constants.QUEUE_NAME,true,consumer);风险处理过程中消费者崩溃消息会永久丢失。2.3 手动确认autoAck false开启手动确认后必须调用以下三种方法之一否则消息会一直占用消费者连接关闭后还会被重新投递。①basicAck— 确认成功channel.basicAck(deliveryTag,multiple);参数说明deliveryTag消息唯一标识从envelope.getDeliveryTag()获取multipletrue确认 deliveryTag 及之前所有未确认消息false只确认当前这条DefaultConsumerconsumernewDefaultConsumer(channel){OverridepublicvoidhandleDelivery(StringconsumerTag,Envelopeenvelope,AMQP.BasicPropertiesproperties,byte[]body)throwsIOException{StringmsgnewString(body);try{System.out.println(处理消息: msg);// 业务处理成功确认消息channel.basicAck(envelope.getDeliveryTag(),false);}catch(Exceptione){// 处理失败见下方 reject / nack}}};channel.basicConsume(Constants.QUEUE_NAME,false,consumer);②basicReject— 拒绝单条消息channel.basicReject(deliveryTag,requeue);参数说明deliveryTag消息标识requeuetrue放回队列重新投递false丢弃或进入死信队列// 处理失败拒绝消息并重新入队channel.basicReject(envelope.getDeliveryTag(),true);// 处理失败拒绝消息并丢弃channel.basicReject(envelope.getDeliveryTag(),false);basicReject只能拒绝单条消息。③basicNack— 拒绝一条或多条消息channel.basicNack(deliveryTag,multiple,requeue);例如105号已经确认成功,但是110号被拒接,那么批量拒绝 106~110号参数说明deliveryTag消息标识multipletrue拒绝 deliveryTag 及之前所有未确认消息requeue同basicReject// 拒绝当前消息重新入队channel.basicNack(envelope.getDeliveryTag(),false,true);// 批量拒绝之前所有未确认消息不重新入队channel.basicNack(envelope.getDeliveryTag(),true,false);2.4 三种手动确认方法对比方法作用批量操作典型场景basicAck确认消费成功支持multipletrue正常处理完成basicReject拒绝单条消息不支持单条处理失败basicNack拒绝一条或多条支持multipletrue批量失败、需要 nack 的场景三、Spring 环境Spring AMQPSpring 将确认模式封装为AcknowledgeMode枚举publicenumAcknowledgeMode{NONE,// 自动确认MANUAL,// 手动确认AUTO;// Spring 自动确认默认}3.1AcknowledgeMode.NONE— 等同自动确认消息投递给消费者后Broker 立即删除与原生autoAcktrue等价。yaml文件中配置spring:rabbitmq:listener:simple:acknowledge-mode:none方法RabbitListener(queuesmy.queue)publicvoidlisten(Stringmessage){System.out.println(收到: message);// 无需手动 ack消息已被自动确认}风险消费者崩溃时消息丢失。3.2AcknowledgeMode.AUTO— 默认模式推荐Spring 容器根据监听方法执行结果自动确认监听方法执行情况Spring 行为正常返回无异常自动basicAck抛出异常自动basicNack默认 requeuetrue消息重新入队yaml文件中配置spring:rabbitmq:listener:simple:acknowledge-mode:auto方法RabbitListener(queuesmy.queue)publicvoidlisten(Stringmessage){System.out.println(处理: message);// 方法正常结束 → 自动 ack// 抛出异常 → 自动 nack 并 requeue}也可在注解上单独指定RabbitListener(queuesmy.queue,ackModeAUTO)publicvoidlisten(Stringmessage){...}3.3AcknowledgeMode.MANUAL— 完全手动控制开发者自行决定何时确认与原生autoAckfalse等价可靠性最高。yaml文件中配置spring:rabbitmq:listener:simple:acknowledge-mode:none监听方法注入ChannelRabbitListener(queuesmy.queue,ackModeMANUAL)publicvoidlisten(Messagemessage,Channelchannel)throwsIOException{longdeliveryTagmessage.getMessageProperties().getDeliveryTag();StringbodynewString(message.getBody());try{System.out.println(处理: body);// 业务逻辑...// ① 确认成功channel.basicAck(deliveryTag,false);}catch(BusinessExceptione){// ② 拒绝单条true重新入队channel.basicReject(deliveryTag,true);}catch(FatalExceptione){// ③ 拒绝单条false丢弃或进死信队列channel.basicReject(deliveryTag,false);}catch(Exceptione){// ④ true表示批量拒绝,false表示不重新入队channel.basicNack(deliveryTag,true,false);}}Spring 手动模式下三种方法说明方法签名说明basicAckchannel.basicAck(deliveryTag, multiple)确认消费成功从队列移除basicRejectchannel.basicReject(deliveryTag, requeue)拒绝单条requeuetrue重新入队basicNackchannel.basicNack(deliveryTag, multiple, requeue)可批量拒绝requeue控制是否重新入队deliveryTag从message.getMessageProperties().getDeliveryTag()获取。Spring 注入的Channel是com.rabbitmq.client.ChannelAPI 与原生 Client 一致。使用AmqpHeaders.ACKNOWLEDGMENT另一种写法RabbitListener(queuesmy.queue,ackModeMANUAL)publicvoidlisten(Messagemessage,Header(AmqpHeaders.DELIVERY_TAG)longdeliveryTag,Channelchannel)throwsIOException{// 处理成功channel.basicAck(deliveryTag,false);}四、两种环境对比总结确认策略原生 ClientSpring AMQP收到即删除autoAck trueAcknowledgeMode.NONE框架自动确认无直接等价AcknowledgeMode.AUTO开发者手动确认autoAck falseAcknowledgeMode.MANUAL场景推荐模式日志、监控等允许丢失NONE/autoAcktrue一般业务默认AUTO支付、订单等强一致MANUAL/autoAckfalse五、注意事项手动模式下必须 ackMANUAL或autoAckfalse时不确认的消息会一直占用消费者连接断开后会重新投递。requeuetrue可能导致死循环处理逻辑有 bug 时消息会反复投递建议配合重试次数或死信队列DLX。Channel 不是线程安全的不要在多个线程中共享同一个 Channel 做 ack。Spring AUTO 模式的异常处理可配合RetryTemplate、RabbitListener的errorHandler或死信队列避免无限重试。multiple参数批量确认时要谨慎避免误确认未处理完的消息。