文章目录
在 RabbitMQ之生产者发布确认原理章节已经介绍了rabbitmq生产者是如何对消息进行发布确认保证消息不丢失的。本章节继续看下springboot整合rabbitmq后是如何保证消息不丢失的。
1. 消息需要发布确认的原因
消息正常是通过生产者生产消息传递到交换机，然后经过交换机路由到消息队列中，最后消费者消费，如下图所示
上述为正常情况，但也有情况会导致消息丢失的情况：第一种情况，交换机重启，当消息经过消费者投递后，恰巧交换机正在重启，会导致生产者投递消息失败，从而导致消息丢失；第二种情况，交换机没问题，消息投递到交换机后，交换机路由到消息队列过程中出了问题，比如routingKey错误导致路由不到队列中。
针对上述两种导致消息丢失的情况，下面采用消息确认发布机制，分别采取消息正确投递到交换机后回调接口来确认消息正确被投递，消息经交换机正确路由到队列中回调接口来确认消息正确被路由。
2. 消息发送交换机后回调接口ConfirmCallback ，保证消息在发送交换机处不丢失
当消息经生产者投递到交换机后，为避免消息丢失，需要回调RabbitTemplate.ConfirmCallback接口，回调接口后，尤其是要对投递失败的消息进行处理或者记录下来保证消息不丢失。该接口不管消息投递到交换机成功或者失败都会进行回调，未避免消息丢失，可以选择在回调接口中只处理或者登记投递失败的消息，达到消息不丢失的目的。
下面通过案例演示生产者投递消息到交换机后回调ConfirmCallback接口情况。
`
2.1 在application.properties中添加spring.rabbitmq.publisher-confirm-type=correlated配置，开启回调机制
spring.rabbitmq.host=192.168.xx.xxx
spring.rabbitmq.port=5672
spring.rabbitmq.username=admin
spring.rabbitmq.password=123
spring.rabbitmq.publisher-confirm-type=correlated
2.2 声明交换机和队列并进行绑定
声明confirm_exchange交换机，声明confirm_queue队列，并通过routingKey=confirm绑定交换机和队列。
import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
public class ConfirmConfig {
public static final String CONFIRM_EXCHANGE_NAME = "confirm_exchange";
public static final String CONFIRM_QUEUE_NAME = "confirm_queue";
public static final String ROUTING_KEY = "confirm";
/*声明交换机*/
@Bean
public DirectExchange confirmExchange(){
return new DirectExchange(CONFIRM_EXCHANGE_NAME);
}
/*声明队列*/
@Bean
public Queue confirmQueue(){
return QueueBuilder.durable(CONFIRM_QUEUE_NAME).build();
}
/*绑定交换机和队列*/
@Bean
public Binding exchangeBindingQueue(@Qualifier("confirmExchange") DirectExchange confirmExchange,
@Qualifier("confirmQueue") Queue confirmQueue){
return BindingBuilder.bind(confirmQueue).to(confirmExchange).with(ROUTING_KEY);
}
}
2.3 实现RabbitTemplate.ConfirmCallback接口
当消息由生产者发到交换机后会回调该接口中的confirm方法
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Component;
@Component
@Slf4j
public class ExchangeCallback implements RabbitTemplate.ConfirmCallback {
/* correlationData 内含消息内容
* ack 交换机接受成功或者失败。 true表示交换机接受消息成功， false表示交换机接受失败
* cause 表示失败原因
*/
@Override
public void confirm(CorrelationData correlationData, boolean ack, String cause) {
System.out.println("hello world");
String id = correlationData.getId();
String message = new String(correlationData.getReturnedMessage().getBody());
if (ack){
log.info("交换机收到消息id为{}, 消息内容为{}", id, message);
}else {
log.info("交换机未收到消息id为{}, 消息内容为{}, 原因为{}", id, message, cause);
}
}
}
2.4 创建生产者
创建生产者生产消息，本案例中生产者发送了2个消息，分别为hello rabbitmq 1和hello rabbitmq 2。
在创建生产者时要设置一下需要回调的接口ExchangeCallback ，在设置回调接口时用了java的@PostConstruct注解，该注解作用用来指定bean初始化的顺序，Constructor(构造方法) -> @Autowired(依赖注入) -> @PostConstruct(注释的方法)。也就是说在初始化Producer3对象时，先调了Producer3的默认无参构造函数；然后执行Autowired注解部分，从Spring IOC容器中寻找rabbitTemplate和exchangeCallback对象分别注入到Producer3的rabbitTemplate和exchangeCallback属性中；最后执行PostConstruct注解部分，把exchangeCallback设置到rabbitTemplate中。
在发送hello rabbitmq 2消息时故意把routingKey写错导致hello rabbitmq 2消息不能从交换机发送到队列中，为下一节做铺垫。
import com.lzj.config.ConfirmConfig;
import com.lzj.config.ExchangeCallback;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import javax.annotation.PostConstruct;
@Slf4j
@Component
public class Producer3 {
@Autowired
private RabbitTemplate rabbitTemplate;
@Autowired
private ExchangeCallback exchangeCallback;
@PostConstruct
public void init(){
rabbitTemplate.setConfirmCallback(exchangeCallback);
}
public void produceMessage(){
String message = "hello rabbitmq 1";
CorrelationData correlationData1 = new CorrelationData("1");
correlationData1.setReturnedMessage(new Message(message.getBytes()));
rabbitTemplate.convertAndSend(ConfirmConfig.CONFIRM_EXCHANGE_NAME, ConfirmConfig.ROUTING_KEY, message, correlationData1);
message = "hello rabbitmq 2";
CorrelationData correlationData2 = new CorrelationData("2");
correlationData2.setReturnedMessage(new Message(message.getBytes()));
rabbitTemplate.convertAndSend(ConfirmConfig.CONFIRM_EXCHANGE_NAME, "error_routingkey", message, correlationData2);
}
}
2.5 启动测试
@SpringBootApplication
@Slf4j
public class SpringbootDemo {
public static void main(String[] args) {
ConfigurableApplicationContext app = SpringApplication.run(SpringbootDemo.class, args);
Producer3 producer = app.getBean(Producer3.class);
producer.produceMessage();
}
}
启动上面测试案例，生产者发送了id为1的hello rabbitmq 1消息和id为2的hello rabbitmq 2，交换机收到消息也回调了ExchangeCallback 接口对2条消息都进行了确认，尤其是对于失败的消息要在此步保存失败的消息，避免消息在交换机这一步丢失。
2022-08-04 00:43:28.817 INFO 13512 --- [ main] com.lzj.producer.Producer3 : 生产者发送id为1, 消息内容为hello rabbitmq 1
2022-08-04 00:43:28.823 INFO 13512 --- [ main] com.lzj.producer.Producer3 : 生产者发送id为2, 消息内容为hello rabbitmq 2
2022-08-04 00:43:28.827 INFO 13512 --- [nectionFactory1] com.lzj.config.ExchangeCallback : 交换机收到消息id为1, 消息内容为hello rabbitmq 1
2022-08-04 00:43:28.828 INFO 13512 --- [nectionFactory2] com.lzj.config.ExchangeCallback : 交换机收到消息id为2, 消息内容为hello rabbitmq 2
但上面还有一个问题，虽然回调ExchangeCallback接口，可以保证消息到交换机一步不会丢失，但如果交换机到队列的过程中出现了问题，消息一样会丢失。比如上面生产者把routingKey写错了，就会导致hello rabbitmq 2消息从交换机路由不到队列中。下面创建消费者程序，看消费者消费confirm_queue队列中消息情况
import com.lzj.config.ConfirmConfig;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
@Slf4j
@Component
public class Consumer3 {
@RabbitListener(queues = ConfirmConfig.CONFIRM_QUEUE_NAME)
public void consuemrMessage(Message message, Channel channel){
String msg = new String(message.getBody());
log.info("消费者消费消息{}", msg);
}
}
然后重新启动测试代码，输出如下，可以看出消费端只消费了hello rabbitmq 1，而hello rabbitmq 2消息则丢失了。
2022-08-04 00:45:28.817 INFO 13512 --- [ main] com.lzj.producer.Producer3 : 生产者发送id为1, 消息内容为hello rabbitmq 1
2022-08-04 00:45:28.823 INFO 13512 --- [ main] com.lzj.producer.Producer3 : 生产者发送id为2, 消息内容为hello rabbitmq 2
2022-08-04 00:45:28.827 INFO 13512 --- [nectionFactory1] com.lzj.config.ExchangeCallback : 交换机收到消息id为1, 消息内容为hello rabbitmq 1
2022-08-04 00:45:28.828 INFO 13512 --- [nectionFactory2] com.lzj.config.ExchangeCallback : 交换机收到消息id为2, 消息内容为hello rabbitmq 2
2022-08-04 00:45:28.831 INFO 13512 --- [ntContainer#2-1] com.lzj.consumer.Consumer3 : 消费者消费消息hello rabbitmq 1
3. 消息经交换机路由到队列后回调接口ReturnCallback ，保证消息在发送队列处不丢失
为解决上一节中，消息由交换机和消息队列中异常，导致消息丢失问题，解决办法就是在添加消息从交换机路由到队列中失败后回调的接口，在回调接口中把失败的消息保存下来就可以避免消息丢失了。
在回调接口之前还需为RabbitMQ设置Mandatory标志，只有当该标志为true时，消息由交换机到队列失败后才会回调接口；如果该标志设置false时，消息由交换机路由到队列失败后自动丢弃消息，会导致消息丢失，这也是默认设置，所以如需保证消息不丢失，要打开Mandatory标志。
下面继续进行上面案例，在上面案例的基础上添加ReturnCallback接口实现
package com.lzj.config;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Component;
@Slf4j
@Component
public class QueueCallback implements RabbitTemplate.ReturnCallback {
@Override
public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
log.info("消息 {} 经交换机 {} 通过routingKey={} 路由到队列失败，失败code为：{}， 失败原因为：{}",
new String(message.getBody()), exchange, routingKey, replyCode, replyText);
}
}
然后修改上面的生产者Producer3 ，只需修改2点：第一点，为RabbitTemplate 设置Mandatory标志；第二点，把ReturnCallback的实现加入监听。
import com.lzj.config.ConfirmConfig;
import com.lzj.config.ExchangeCallback;
import com.lzj.config.QueueCallback;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import javax.annotation.PostConstruct;
@Slf4j
@Component
public class Producer3 {
@Autowired
private RabbitTemplate rabbitTemplate;
@Autowired
private ExchangeCallback exchangeCallback;
@Autowired
private QueueCallback queueCallback;
@PostConstruct
public void init(){
rabbitTemplate.setConfirmCallback(exchangeCallback);
/**
* true：交换机无法将消息进行路由时，会将该消息返回给生产者
* false：如果发现消息无法进行路由，则直接丢弃
*/
rabbitTemplate.setMandatory(true);
rabbitTemplate.setReturnCallback(queueCallback);
}
public void produceMessage(){
String message = "hello rabbitmq 1";
CorrelationData correlationData1 = new CorrelationData("1");
correlationData1.setReturnedMessage(new Message(message.getBytes()));
rabbitTemplate.convertAndSend(ConfirmConfig.CONFIRM_EXCHANGE_NAME, ConfirmConfig.ROUTING_KEY, message, correlationData1);
log.info("生产者发送id为{}, 消息内容为{}", correlationData1.getId(), message);
message = "hello rabbitmq 2";
CorrelationData correlationData2 = new CorrelationData("2");
correlationData2.setReturnedMessage(new Message(message.getBytes()));
rabbitTemplate.convertAndSend(ConfirmConfig.CONFIRM_EXCHANGE_NAME, "error_routingkey", message, correlationData2);
log.info("生产者发送id为{}, 消息内容为{}", correlationData2.getId(), message);
}
}
下面重新启动上面的测试案例，可以得出如下测试结果，生产者生产的hello rabbitmq 1和hello rabbitmq 2消息都已发到交换机，也都被交换机确认了，但hello rabbitmq 2被交换机路由到队列时由于routingKey错误导致路由失败，已在ReturnCallback接口回调中被记录下来，最终正确被路由到队列中的消息只有hello rabbitmq 1，从打印日志看也只有hello rabbitmq 1被消费了。
2022-08-05 00:39:29.670 INFO 6832 --- [ main] com.lzj.producer.Producer3 : 生产者发送id为1, 消息内容为hello rabbitmq 1
2022-08-05 00:39:29.683 INFO 6832 --- [ main] com.lzj.producer.Producer3 : 生产者发送id为2, 消息内容为hello rabbitmq 2
2022-08-05 00:39:29.702 INFO 6832 --- [nectionFactory3] com.lzj.config.ExchangeCallback : 交换机收到消息id为1, 消息内容为hello rabbitmq 1
2022-08-05 00:39:29.704 INFO 6832 --- [nectionFactory1] com.lzj.config.QueueCallback : 消息 hello rabbitmq 2 经交换机 confirm_exchange 通过routingKey=error_routingkey 路由到队列失败，失败code为：312， 失败原因为：NO_ROUTE
2022-08-05 00:39:29.705 INFO 6832 --- [nectionFactory2] com.lzj.config.ExchangeCallback : 交换机收到消息id为2, 消息内容为hello rabbitmq 2
2022-08-05 00:39:29.707 INFO 6832 --- [ntContainer#2-1] com.lzj.consumer.Consumer3 : 消费者消费消息hello rabbitmq 1
结论：从上面2个案例可以看出，通过ConfirmCallback和ReturnCallback接口的回调，保证了消息在交换机和队列处消息不丢失。