RabbitMQ高级特性--消息确认机制

目录

一、消息确认

1.消息确认机制

2.手动确认方法

二、代码示例

1. AcknowledgeMode.NONE

1.1 配置文件

1.2 生产者

 1.3 消费者

1.4 运行程序 

 2.AcknowledgeMode.AUTO

3.AcknowledgeMode.MANUAL

一、消息确认

1.消息确认机制

生产者发送消息之后，到达消费端之后，可能会有以下情况：

1. 消息处理成功；

2. 消息处理异常。

RabbitMQ向消费者发送消息后，就会把这条消息删除掉，那么第二种情况就会造成消息丢失。

那么如何确保消息端已经被成功接收了并且被正确处理了呢？

为了确保消息从队列可靠的到达消费者，RabbitMQ提供了消息确认机制（Messageacknowledment）。

消费者在订阅队列时，可以指定autoAck参数，根据这个参数，消息确认机制分为以下两种：

自动确认：当autoAck等于true时，RabbitMQ会自动把发送出去的消息置为确认，然后从内存（或者磁盘）中删除，而不管消费者是否真正的接收到消息，自动确认模式适用于对于消息可靠性要求不高的场景。

手动确认：当autoAck等于false时，RabbitMQ会等待消费者显式的调用BasicAck命令，回复确认信号后才从内存（或者磁盘）中删除，这种方式适用于对消息可靠性要求较高的场景。

自动确认代码示例：

DefaultConsumer consumer = new DefaultConsumer(channel) {
 @Override
 public void handleDelivery(String consumerTag, Envelope envelope, 
AMQP.BasicProperties properties, byte[] body) throws IOException {
 System.out.println("接收到消息: " + new String(body));
 }
};
channel.basicConsume(Constants.TOPIC_QUEUE_NAME1, true, consumer);

当autoAck参数置为false，对于RabbitMQ服务器来说，队列中的消息分为了两个部分：

一是等待发送给消费者的消息；二是已经发送给消费者，但是还没收到消费者确认信号的消息。

如果RabbitMQ一直没有收到消费者的确认信号，并且消费此消息的消费者已经断开连接，则RabbitMQ会重新安排这条消息进入队列，等待投递给下一个消费者，当然也有可能是原来的那个消费者。

从RabbitMQ的Web管理平台上也可以看到当前队列中Ready状态和Unacked状态的消息数。 

Ready：等待投递给消费者的消息数。

Unacked：已经投递给消费者，但是未收到消费者确认信号的消息数。

2.手动确认方法

消费者在收到消息后，可以选择确认，也可以选择跳过或者直接拒绝确认，RabbitMQ也提供了不同的确认方法，消费者客户端可以调用与其对应的channel的相关方法，共有以下三种：

肯定确认: Channel.basicAck(long deliveryTag, boolean multiple);

RabbitMQ 已经知道该消息并且成功的处理消息，可以将其丢弃。

参数说明：

deliveryTag：消息的唯一标识，它是一个单调递增的64位的长整型值，deliveryTag是每个信道（Channel）独立维护的，所以在每个信道上都是唯一的，当消费者确认（ack）一条消息时，必须使用对应的信道进行确认。

multiple：是否批量确认，在某些情况下，为了减少网络流量，可以对一系列连续的deliveryTag进行批量确认，值为true则会一次性ack所以小于等于指定deliveryTag的消息，值为false，则只确认当前deliveryTag的消息。

否定确认: Channel.basicReject(long deliveryTag, boolean requeue);

否定确认: Channel.basicNack(long deliveryTag, boolean multiple,
boolean requeue);

public enum AcknowledgeMode {
 NONE,
 MANUAL,
 AUTO;
}

1. AcknowledgeMode.NONE

1.1 配置文件

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

1.2 生产者

public class Constant {
 public static final String ACK_EXCHANGE_NAME = "ack_exchange";
 public static final String ACK_QUEUE = "ack_queue";
}

/*
以下为消费端⼿动应答代码⽰例配置
*/
@Bean("ackExchange")
public Exchange ackExchange(){
 return
ExchangeBuilder.topicExchange(Constant.ACK_EXCHANGE_NAME).durable(true).build()
;
}
//2. 队列
@Bean("ackQueue")
public Queue ackQueue() {
 return QueueBuilder.durable(Constant.ACK_QUEUE).build();
}
//3. 队列和交换机绑定 Binding
@Bean("ackBinding")
public Binding ackBinding(@Qualifier("ackExchange") Exchange exchange, 
@Qualifier("ackQueue") Queue queue) {
 return BindingBuilder.bind(queue).to(exchange).with("ack").noargs();
}

import com.xiaowu.rabbitmq.constant.Constant;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
@RestController
@RequestMapping("/producer")
public class ProductController {
 @Autowired
 private RabbitTemplate rabbitTemplate;
 @RequestMapping("/ack")
 public String ack(){
 rabbitTemplate.convertAndSend(Constant.ACK_EXCHANGE_NAME, "ack", 
"consumer ack test...");
 return "发送成功!";
 }
}

 1.3 消费者

import com.xiaowu.rabbitmq.constant.Constant;
import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
@Component
public class AckQueueListener {
 //指定监听队列的名称
 @RabbitListener(queues = Constant.ACK_QUEUE)
 public void ListenerQueue(Message message, Channel channel) throws
Exception {
 System.out.printf("接收到消息: %s, deliveryTag: %d%n", new
String(message.getBody(),"UTF-8"), 
message.getMessageProperties().getDeliveryTag());
 //模拟处理失败
 int num = 3/0;
 System.out.println("处理完成");
 }
}

1.4 运行程序 

启动生产者可以从RabbitMQ Web管理界面看到如下：

再启动消费者，控制台输出：

接收到消息: consumer ack test..., deliveryTag: 1
2024-04-29T17:03:57.797+08:00 WARN 16952 --- [ntContainer#0-1] 
s.a.r.l.ConditionalRejectingErrorHandler : Execution of Rabbit message 
listener failed.
org.springframework.amqp.rabbit.support.ListenerExecutionFailedException: 
Listener method 'public void 
com.xiaowu.rabbitmq.listener.AckQueueListener.ListenerQueue(org.springframework.a
mqp.core.Message,com.rabbitmq.client.Channel) throws java.lang.Exception'
threw exception
//....

管理界面：

可以看到消息处理失败但是消息已经从管理界面移除。 

 2.AcknowledgeMode.AUTO

将配置文件修改为：

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

再次启动程序，控制台不断输出错误信息：

接收到消息: consumer ack test..., deliveryTag: 1
2024-04-29T17:07:06.114+08:00 WARN 16488 --- [ntContainer#0-1] 
s.a.r.l.ConditionalRejectingErrorHandler : Execution of Rabbit message 
listener failed.
org.springframework.amqp.rabbit.support.ListenerExecutionFailedException: 
Listener method 'public void 
com.xiaowu.rabbitmq.listener.AckQueueListener.ListenerQueue(org.springframework.a
mqp.core.Message,com.rabbitmq.client.Channel) throws java.lang.Exception'
threw exception
 
接收到消息: consumer ack test..., deliveryTag: 2
2024-04-29T17:07:07.161+08:00 WARN 16488 --- [ntContainer#0-1] 
s.a.r.l.ConditionalRejectingErrorHandler : Execution of Rabbit message 
listener failed.
org.springframework.amqp.rabbit.support.ListenerExecutionFailedException: 
Listener method 'public void 
com.xiaowu.rabbitmq.listener.AckQueueListener.ListenerQueue(org.springframework.a
mqp.core.Message,com.rabbitmq.client.Channel) throws java.lang.Exception'
threw exception
 
接收到消息: consumer ack test..., deliveryTag: 3
2024-04-29T17:07:08.208+08:00 WARN 16488 --- [ntContainer#0-1] 
s.a.r.l.ConditionalRejectingErrorHandler : Execution of Rabbit message 
listener failed.
org.springframework.amqp.rabbit.support.ListenerExecutionFailedException: 
Listener method 'public void 
com.xiaowu.rabbitmq.listener.AckQueueListener.ListenerQueue(org.springframework.a
mqp.core.Message,com.rabbitmq.client.Channel) throws java.lang.Exception'
threw exception
 
接收到消息: consumer ack test..., deliveryTag: 4
2024-04-29T17:07:09.254+08:00 WARN 16488 --- [ntContainer#0-1] 
s.a.r.l.ConditionalRejectingErrorHandler : Execution of Rabbit message 
listener failed.
org.springframework.amqp.rabbit.support.ListenerExecutionFailedException: 
Listener method 'public void 
com.xiaowu.rabbitmq.listener.AckQueueListener.ListenerQueue(org.springframework.a
mqp.core.Message,com.rabbitmq.client.Channel) throws java.lang.Exception'
threw exception
 

3.AcknowledgeMode.MANUAL

spring:
 rabbitmq:
 addresses: amqp:
 listener:
 simple:
 acknowledge-mode: manual

消费者手动确认逻辑：

import com.xiaowu.rabbitmq.constant.Constant;
import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
@Component
public class AckQueueListener {
 //指定监听队列的名称
 @RabbitListener(queues = Constant.ACK_QUEUE)
 public void ListenerQueue(Message message, Channel channel) throws
Exception {
 long deliveryTag = message.getMessageProperties().getDeliveryTag();
 try {
 //1. 接收消息
 System.out.printf("接收到消息: %s, deliveryTag: %d%n", new
String(message.getBody(),"UTF-8"), 
message.getMessageProperties().getDeliveryTag());
 //2. 处理业务逻辑
 System.out.println("处理业务逻辑");
 //⼿动设置⼀个异常, 来测试异常拒绝机制
// int num = 3/0;
 //3. ⼿动签收
 channel.basicAck(deliveryTag, true);
 } catch (Exception e) {
 //4. 异常了就拒绝签收
 //第三个参数requeue, 是否重新发送, 如果为true, 则会重新发送,,若为false, 
则直接丢弃
 channel.basicNack(deliveryTag, true,true);
 }
 }
}

 这个代码运行的结果是正常的, 运行后消息会被签收: Ready为0, unacked为0。

异常时拒绝：

import com.xiaowu.rabbitmq.constant.Constant;
import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
@Component
public class AckQueueListener {
 //指定监听队列的名称
 @RabbitListener(queues = Constant.ACK_QUEUE)
 public void ListenerQueue(Message message, Channel channel) throws
Exception {
 long deliveryTag = message.getMessageProperties().getDeliveryTag();
 try {
 //1. 接收消息
 System.out.printf("接收到消息: %s, deliveryTag: %d%n", new
String(message.getBody(),"UTF-8"), 
message.getMessageProperties().getDeliveryTag());
 //2. 处理业务逻辑
 System.out.println("处理业务逻辑");
 //⼿动设置⼀个异常, 来测试异常拒绝机制
 int num = 3/0;
 //3. ⼿动签收
 channel.basicAck(deliveryTag, true);
 } catch (Exception e) {
 //4. 异常了就拒绝签收
 //第三个参数requeue, 是否重新发送, 如果为true, 则会重新发送,,若为false, 
则直接丢弃
 channel.basicNack(deliveryTag, true,true);
 }
 }
}

接收到消息: consumer ack test..., deliveryTag: 1
处理业务逻辑
接收到消息: consumer ack test..., deliveryTag: 2
处理业务逻辑
接收到消息: consumer ack test..., deliveryTag: 3
处理业务逻辑
接收到消息: consumer ack test..., deliveryTag: 4
处理业务逻辑
接收到消息: consumer ack test..., deliveryTag: 5
处理业务逻辑
接收到消息: consumer ack test..., deliveryTag: 6
处理业务逻辑