原文地址：https://blog.csdn.net/qq_29595463/article/details/109527115
文章目录
1. 消息确定机制及其配置
2. 自动确认和手动确认细节
2.1 测试代码配置
2.2 自动确认
2.3 手动确认
2.3.1 常用API
2.3.2 代码案例
3. 消息重回队列
3.1 默认—消息nack后重回队头
3.2 解决方案—消息重发送到队尾
3.3 如何处理异常消息
推荐阅读
1. 消息确定机制及其配置
RabbitMq消费者的消息确定机制：
NONE：无应答，rabbitmq默认consumer正确处理所有请求。
AUTO：consumer自动应答，处理成功（注意：此处的成功确认是没有发生异常）发出ack，处理失败发出nack。rabbitmq发出消息后会等待consumer端应答，只有收到ack确定信息后才会将消息在rabbitmq清除掉。收到nack异常信息的处理方法由setDefaultRequeueReject()方法设置，这种模式下，发送错误的消息可以恢复。
MANUAL：基本等同于AUTO模式，区别是需要人为调用方法确认。
在配置文件中：
spring:
profiles:
active: test
rabbitmq:
host: localhost
port: 5672
username: guest
password: guest
template:
receive-timeout: 2000
reply-timeout: 2000
listener:
simple:
concurrency: 1
max-concurrency: 3
# 消费者预取1条数据到内存，默认为250条
prefetch: 1
# 确定机制
acknowledge-mode: manual
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
可以进行配置，也可以在代码中进行配置：
@Configuration
public class RabbitConfig {
@Autowired
private CachingConnectionFactory connectionFactory;
@Bean(name = "singleListenerContainer")
public SimpleRabbitListenerContainerFactory listenerContainerFactory() {
SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
/* setConnectionFactory：设置spring-amqp的ConnectionFactory。 */
factory.setConnectionFactory(connectionFactory);
factory.setConcurrentConsumers(1);
factory.setMaxConcurrentConsumers(10);
factory.setPrefetchCount(1);
factory.setDefaultRequeueRejected(true);
//手动确认。
factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
return factory;
}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
2. 自动确认和手动确认细节
2.1 测试代码配置
队列配置：
@Configuration
public class RabbitMqConfig {
// 简单的声明一个队列
@Bean
public Queue kinsonQueue1() {
return new Queue("kinson1");
}
}
1
2
3
4
5
6
7
8
生产者配置：
@RestController
public class RabbitMQController {
@Autowired
private RabbitTemplate rabbitTemplate;
//直接向队列中发送数据
@GetMapping("send")
public String send() {
for (int i = 0; i < 10; i++) {
String content = "Date:" + System.currentTimeMillis();
content = content + ":::" + i;
rabbitTemplate.convertAndSend("kinson1",i+"");
}
return "success";
}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
消费者配置
@Component
@Slf4j
public class CustomerRev {
//消费者处理消息缓慢
@RabbitListener(queues = {"kinson1"})
public void receiver3(Message msg, Channel channel) {
try {
//打印数据
String message = new String(msg.getBody(), StandardCharsets.UTF_8);
log.info("【开始】：{}",message);
Thread.sleep(30000);
log.info("【结束】:{}", message);
} catch (Exception e) {
log.error("错误信息：{}", e.getMessage());
}
}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
2.2 自动确认
生产者产生10笔消息，自动确认模式下，消息处理成功，消费者才会去获取下一笔消息；消息处理抛出异常，那么将会消息重回队列。
注：此时的prefetch=1，该消费者unack的消息只有一条。其余9条均在MQ中。
若是设置prefetch=250（默认），那么消费者会预期10条消息，unack的消息是10条，但是日志中依旧只是消费了一条消息。
【开始】：Date:1604557983496:::0
1
**总结：**对于自动确认来说，当方法没有异常执行完毕后，会对MQ发出ACK；若方法出现异常，会对MQ发出nack，消息会重回队列。要分清哪些是可以恢复的异常，哪些是不可以恢复的异常。不可恢复的异常，在消费者代码中捕获异常，并记录日志表或放入死信队列。可恢复的异常，那么放入业务队列中重试。
2.3 手动确认
2.3.1 常用API
channel.basicAck(msg.getMessageProperties().getDeliveryTag(),false);
ack表示确认消息。multiple：false只确认该delivery_tag的消息，true确认该delivery_tag的所有消息
channel.basicReject(msg.getMessageProperties().getDeliveryTag(),false);
Reject表示拒绝消息。requeue：false表示被拒绝的消息是丢弃；true表示重回队列
channel.basicNack(msg.getMessageProperties().getDeliveryTag(),false,false);
nack表示拒绝消息。multiple表示拒绝指定了delivery_tag的所有未确认的消息，requeue表示不是重回队列
2.3.2 代码案例
案例：消费者只有发送ACK后才会开始消费下一条消息。
@Component
@Slf4j
public class CustomerRev {
//消费者处理消息缓慢
@RabbitListener(queues = {"kinson1"})
public void receiver3(Message msg, Channel channel) {
try {
//打印数据
String message = new String(msg.getBody(), StandardCharsets.UTF_8);
log.info("【开始】：{}",message);
if("0".equals(message)){
Thread.sleep(1000);
}else {
Thread.sleep(300000000L);
}
log.info("【结束】:{}", message);
//ack表示确认消息。multiple：false只确认该delivery_tag的消息，true确认该delivery_tag的所有消息
channel.basicAck(msg.getMessageProperties().getDeliveryTag(),false);
} catch (Exception e) {
log.error("错误信息：{}", e.getMessage());
}
}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
3. 消息重回队列
自动确认模式下抛出异常，和手动确认模式下nack消息都可以重回队列。
3.1 默认—消息nack后重回队头
@Component
@Slf4j
public class CustomerRev {
//消费者处理消息缓慢
@RabbitListener(queues = {"kinson1"})
public void receiver3(Message msg, Channel channel) {
try {
//打印数据
String message = new String(msg.getBody(), StandardCharsets.UTF_8);
log.info("【开始】：{}",message);
if("0".equals(message)){
throw new RuntimeException("0的消息消费异常");
}
log.info("【结束】:{}", message);
//ack表示确认消息。multiple：false只确认该delivery_tag的消息，true确认该delivery_tag的所有消息
channel.basicAck(msg.getMessageProperties().getDeliveryTag(),false);
} catch (Exception e) {
throw new RuntimeException(e);
}
}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
（自动ACK下）当消息为0时，抛出异常，生产者产生10笔消息。
代码轮询打印0的消息的消费异常，其他消息均不消费。
当消息回滚到消息队列时，这条消息不会回到队列尾部，而是仍是在队列头部，这时消费者会立马又接收到这条消息进行处理，接着抛出异常，进行 回滚，如此反复进行。这种情况会导致消息队列处理出现阻塞，消息堆积，导致正常消息也无法运行。
3.2 解决方案—消息重发送到队尾
//消费者处理消息缓慢
@RabbitListener(queues = {"kinson1"})
public void receiver3(Message msg, Channel channel) throws IOException {
try {
//打印数据
String message = new String(msg.getBody(), StandardCharsets.UTF_8);
log.info("【开始】：{}",message);
if("0".equals(message)){
throw new RuntimeException("0的消息消费异常");
}
log.info("【结束】:{}", message);
//ack表示确认消息。multiple：false只确认该delivery_tag的消息，true确认该delivery_tag的所有消息
channel.basicAck(msg.getMessageProperties().getDeliveryTag(),false);
} catch (Exception e) {
//捕获异常后，重新发送到指定队列，自动ack不抛出异常即为ack
channel.basicPublish(msg.getMessageProperties().getReceivedExchange(),
msg.getMessageProperties().getReceivedRoutingKey(), MessageProperties.PERSISTENT_TEXT_PLAIN,
msg.getBody());
}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
手动确认的模式：
//手动进行应答
channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
//重新发送消息到队尾
channel.basicPublish(message.getMessageProperties().getReceivedExchange(),
message.getMessageProperties().getReceivedRoutingKey(), MessageProperties.PERSISTENT_TEXT_PLAIN,
JSON.toJSONBytes(new Object()));
1
2
3
4
5
6
这个不会导致其他消息被阻塞消费，但是会大量的抛出：
3.3 如何处理异常消息
如果一个消息体本身有误，会导致该消息体，一直无法进行处理，而服务器中刷出大量无用日志。解决这个问题可以采取两种方案：
1.一种是对于日常细致处理，分清哪些是可以恢复的异常，哪些是不可以恢复的异常。对于可以恢复的异常我们采取第三条中的解决方案，对于不可以处理的异常，我们采用记录日志，直接丢弃该消息方案。
2.另一种是我们对每条消息进行标记，记录每条消息的处理次数，当一条消息，多次处理仍不能成功时，处理次数到达我们设置的值时，我们就丢弃该消息，但需要记录详细的日志。
将业务队列绑定死信队列，当消息被丢弃后，进入到死信队列（代码修复后监听死信队列补偿消息）。可以避免我们手动的恢复消息。
重试机制（自动确认模式）：
@Component
@Slf4j
public class CustomerRev {
@RabbitListener(queues = {"kinson1"})
public void receiver3(Message msg, Channel channel) throws IOException {
try {
//打印数据
String message = new String(msg.getBody(), StandardCharsets.UTF_8);
log.info("【开始】：{}",message);
if("0".equals(message)){
throw new RuntimeException("0的消息消费异常");
}
log.info("【结束】:{}", message);
} catch (Exception e) {
//捕获异常后，重新发送到指定队列，自动确认不抛出异常即为ack
Integer retryCount;
Map<String, Object> headers = msg.getMessageProperties().getHeaders();
if(!headers.containsKey("retry-count")){
retryCount=0;
}else {
retryCount = (Integer)headers.get("retry-count");
}
//判断是否满足最大重试次数(重试3次)
if(retryCount++<3) {
headers.put("retry-count",retryCount);
//重新发送到MQ中
AMQP.BasicProperties basicProperties = new AMQP.BasicProperties().builder().contentType("text/plain").headers(headers).build();
channel.basicPublish(msg.getMessageProperties().getReceivedExchange(),
msg.getMessageProperties().getReceivedRoutingKey(), basicProperties,
msg.getBody());
}
}
}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
每次收到消息后，修改消息的header信息。来记录重试的次数。
推荐阅读
rabbitmq消息重回队列
————————————————
版权声明：本文为CSDN博主「小胖学编程」的原创文章，遵循CC 4.0 BY-SA版权协议，转载请附上原文出处链接及本声明。
原文链接：https://blog.csdn.net/qq_29595463/article/details/109527115