RabbitMQ如何保证发送的消息可靠（RabbitMQ的Confirm模式和2.Return模式）

1、RabbitMQ消息Confirm模式（保证从生产者到交换机的消息可靠）

1.1、Confirm模式简介

消息的confirm确认机制，是指生产者投递消息后，到达了消息服务器Broker里面的exchange交换机，exchange交换机会给生产者一个应答，生产者接收到应答，用来确定这条消息是否正常的发送到Broker的exchange中，这也是消息可靠性投递的重要保障。

1.2、具体代码实现

1 配置文件application.yml 开启确认模式：spring.rabbitmq.publisher-confirm-type=correlated
2 写一个类实现RabbitTemplate.ConfirmCallback，判断成功和失败的ack结果，可以根据具体的结果，如果ack为false，对消息进行重新发送或记录日志等处理；
设置rabbitTemplate的确认回调方法
3 rabbitTemplate.setConfirmCallback(messageConfirmCallBack);

1.2.1、application.yml 开启确认模式

server:port: 8080
spring:application:name: confirm-test01rabbitmq:host: 你的服务器IPport: 5672username: 你的账号password: 你的密码virtual-host: powerpublisher-confirm-type: correlated #开启生产者的确认模式，设置关联模式my:exchangeName: exchange.confirm.01queueName: queue.confirm.01

1.2.2、生产者

方式1：实现RabbitTemplate.ConfirmCallback

单独写一个类，实现RabbitTemplate.ConfirmCallback

package com.power.config;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 MyConfirmCallback implements RabbitTemplate.ConfirmCallback {@Overridepublic void confirm(CorrelationData correlationData, boolean ack, String cause) {if(ack){log.info("消息正确到达交换机");return;}//ack为false,消息没有到达交换机log.error("消息没有到达交换机，原因是：{}",cause);}
}

生产者发送消息

package com.power.service;import com.power.config.MyConfirmCallback;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageBuilder;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Service;import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.Date;@Service
@Slf4j
public class MessageService {@Resourceprivate RabbitTemplate rabbitTemplate;@Resourceprivate MyConfirmCallback confirmCallback;@PostConstruct//构造方法后执行，相当于初始化作用public void init(){rabbitTemplate.setConfirmCallback(confirmCallback);}@Beanpublic void sendMsg(){Message message = MessageBuilder.withBody("hello world".getBytes()).build();CorrelationData correlationData = new CorrelationData();//关联数据correlationData.setId("order_123456");//发送订单信息rabbitTemplate.convertAndSend("exchange.confirm.01","info",message,correlationData);log.info("消息发送完毕，发送时间是：{}",new Date());}
}

方式2：直接写在生产者发送消息类，实现RabbitTemplate.ConfirmCallback

package com.power.service;import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageBuilder;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Service;import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.Date;@Service
@Slf4j
public class MessageService implements RabbitTemplate.ConfirmCallback {@Resourceprivate RabbitTemplate rabbitTemplate;@PostConstruct//构造方法后执行，相当于初始化作用public void init(){rabbitTemplate.setConfirmCallback(this);}@Beanpublic void sendMsg(){Message message = MessageBuilder.withBody("hello world".getBytes()).build();CorrelationData correlationData = new CorrelationData();//关联数据correlationData.setId("order_123456");//发送订单信息rabbitTemplate.convertAndSend("exchange.confirm.01","info",message,correlationData);log.info("消息发送完毕，发送时间是：{}",new Date());}@Overridepublic void confirm(CorrelationData correlationData, boolean ack, String cause) {if(ack){log.info("消息正确到达交换机");return;}//ack为false,消息没有到达交换机log.error("消息没有到达交换机，原因是：{}",cause);}
}

方式3：匿名内部类写法

package com.power.service;import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageBuilder;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Service;import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.Date;@Service
@Slf4j
public class MessageService {@Resourceprivate RabbitTemplate rabbitTemplate;@PostConstruct//构造方法后执行，相当于初始化作用public void init(){rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {@Overridepublic void confirm(CorrelationData correlationData, boolean ack, String cause) {if(ack){log.info("消息正确到达交换机");return;}//ack为false,消息没有到达交换机log.error("消息没有到达交换机，原因是：{}",cause);}});}@Beanpublic void sendMsg(){Message message = MessageBuilder.withBody("hello world".getBytes()).build();CorrelationData correlationData = new CorrelationData();//关联数据correlationData.setId("order_123456");//发送订单信息rabbitTemplate.convertAndSend("exchange.confirm.01","info",message,correlationData);log.info("消息发送完毕，发送时间是：{}",new Date());}}

方式4：lambda表达式写法

package com.power.service;import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageBuilder;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Service;import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.Date;@Service
@Slf4j
public class MessageService {@Resourceprivate RabbitTemplate rabbitTemplate;@PostConstruct//构造方法后执行，相当于初始化作用public void init(){rabbitTemplate.setConfirmCallback(//lambda表达式写法(correlationData, ack, cause)->{log.info("关联id：{}",correlationData.getId());if(ack){log.info("消息正确到达交换机");return;}//ack为false,消息没有到达交换机log.error("消息没有到达交换机，原因是：{}",cause);});}@Beanpublic void sendMsg(){Message message = MessageBuilder.withBody("hello world".getBytes()).build();CorrelationData correlationData = new CorrelationData();//关联数据correlationData.setId("order_123456");//发送订单信息rabbitTemplate.convertAndSend("exchange.confirm.03","info",message,correlationData);log.info("消息发送完毕，发送时间是：{}",new Date());}}

1.2.3、RabbitConfig做交换机和队列的绑定

package com.power.config;import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;@Configuration
public class RabbitConfig {@Value("${my.exchangeName}")private String exchangeName;@Value("${my.queueName}")private String queueName;//创建直连交换机@Beanpublic DirectExchange directExchange(){return ExchangeBuilder.directExchange(exchangeName).build();}//创建队列@Beanpublic Queue queue(){return QueueBuilder.durable(queueName).build();}//交换机绑定队列@Beanpublic Binding binding(DirectExchange exchangeName,Queue queueName){return BindingBuilder.bind(queueName).to(exchangeName).with("info");}
}

1.2.4、pom.xml配置文件

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><groupId>com.power</groupId><artifactId>rabbit_08_confirm01</artifactId><version>1.0-SNAPSHOT</version><name>rabbit_08_confirm01</name><properties><maven.compiler.source>8</maven.compiler.source><maven.compiler.target>8</maven.compiler.target><project.build.sourceEncoding>UTF-8</project.build.sourceEncoding></properties><parent><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-parent</artifactId><version>2.6.13</version><relativePath/></parent><dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><scope>test</scope></dependency><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId><version>1.18.24</version></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId></dependency></dependencies><build><plugins><plugin><groupId>org.springframework.boot</groupId><artifactId>spring-boot-maven-plugin</artifactId></plugin></plugins></build></project>

1.2.5、测试

如果没有任何异常，消息会正常发送到交换机

如果程序存在异常，消息不会正常发送到交换机，如果当交换机的名字不对时，消息不会正常到底交换机的。

2、RabbitMQ消息Return模式（保证从交换机的到队列的消息可靠）

rabbitmq 整个消息投递的路径为：
producer —> exchange —> queue —> consumer

• 消息从 producer 到 exchange 则会返回一个 confirmCallback；
• 消息从 exchange –> queue 投递失败则会返回一个 returnCallback

我们可以利用这两个callback控制消息的可靠性投递；

开启 确认模式；
使用rabbitTemplate.setConfirmCallback设置回调函数，当消息发送到exchange后回调confirm方法。在方法中判断ack，如果为true，则发送成功，如果为false，则发送失败，需要处理；

注意配置文件中，开启 退回模式；

spring.rabbitmq.publisher-returns: true

使用rabbitTemplate.setReturnCallback设置退回函数，当消息从exchange路由到queue失败后，则会将消息退回给producer，并执行回调函数returnedMessage；

2.1、具体代码实现

2.1.1、applicaton.yml

server:port: 8080
spring:application:name: return-test01rabbitmq:host: 你的服务器IPport: 5672username: 你的账号password: 你的密码virtual-host: powerpublisher-returns: true #开启return模式my:exchangeName: exchange.return.01queueName: queue.return.01

2.1.2、pom.xml

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><groupId>com.power</groupId><artifactId>rabbit_09_return01</artifactId><version>1.0-SNAPSHOT</version><name>rabbit_09_return01</name><properties><maven.compiler.source>8</maven.compiler.source><maven.compiler.target>8</maven.compiler.target><project.build.sourceEncoding>UTF-8</project.build.sourceEncoding></properties><parent><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-parent</artifactId><version>2.6.13</version><relativePath/></parent><dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><scope>test</scope></dependency><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId><version>1.18.24</version></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId></dependency></dependencies><build><plugins><plugin><groupId>org.springframework.boot</groupId><artifactId>spring-boot-maven-plugin</artifactId></plugin></plugins></build></project>

2.1.3、启动类

package com.power;import com.power.service.MessageService;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;import javax.annotation.Resource;@SpringBootApplication
public class Application implements ApplicationRunner {@Resourceprivate MessageService messageService;public static void main(String[] args) {SpringApplication.run(Application.class);}@Overridepublic void run(ApplicationArguments args) throws Exception {messageService.sendMsg();}
}

2.1.4、业务层

方式1：实现RabbitTemplate.ReturnsCallback

回调类MyReturnCallback

package com.power.config;import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.ReturnedMessage;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Component;/*** 外部类* 写一个类实现一个接口*/
@Component
@Slf4j
public class MyReturnCallback implements RabbitTemplate.ReturnsCallback {@Overridepublic void returnedMessage(ReturnedMessage returnedMessage) {log.error("消息从交换机没有正确的投递到队列，原因是：{}",returnedMessage.getReplyText());}
}

配置类

package com.power.config;import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;@Configuration
public class RabbitConfig {@Value("${my.exchangeName}")private String exchangeName;@Value("${my.queueName}")private String queueName;//创建直连交换机@Beanpublic DirectExchange directExchange(){return ExchangeBuilder.directExchange(exchangeName).build();}//创建队列@Beanpublic Queue queue(){return QueueBuilder.durable(queueName).build();}//交换机绑定队列@Beanpublic Binding binding(DirectExchange directExchange,Queue queue){return BindingBuilder.bind(queue).to(directExchange).with("info");}
}

service业务层

package com.power.service;import com.power.config.MyReturnCallback;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageBuilder;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Service;import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.Date;@Service
@Slf4j
public class MessageService {@Resourceprivate RabbitTemplate rabbitTemplate;@Resourceprivate MyReturnCallback myReturnCallback;@PostConstructpublic void init(){rabbitTemplate.setReturnsCallback(myReturnCallback);//设置回调}@Beanpublic void sendMsg(){Message message = MessageBuilder.withBody("hello world".getBytes()).build();rabbitTemplate.convertAndSend("exchange.return.01","info111",message);log.info("消息发送完毕，发送时间是：{}",new Date());}
}

方式2：MessageService类实现RabbitTemplate.ReturnsCallback

package com.power.service;import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageBuilder;
import org.springframework.amqp.core.ReturnedMessage;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Service;import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.Date;@Service
@Slf4j
public class MessageService  implements RabbitTemplate.ReturnsCallback {@Resourceprivate RabbitTemplate rabbitTemplate;@PostConstructpublic void init(){rabbitTemplate.setReturnsCallback(this);//设置回调}@Beanpublic void sendMsg(){Message message = MessageBuilder.withBody("hello world".getBytes()).build();rabbitTemplate.convertAndSend("exchange.return.01","info111",message);log.info("消息发送完毕，发送时间是：{}",new Date());}@Overridepublic void returnedMessage(ReturnedMessage returnedMessage) {log.error("消息从交换机没有正确的投递到队列，原因是：{}",returnedMessage.getReplyText());}
}

方式3：匿名内部类实现RabbitTemplate.ReturnsCallback

package com.power.service;import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageBuilder;
import org.springframework.amqp.core.ReturnedMessage;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Service;import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.Date;@Service
@Slf4j
public class MessageService {@Resourceprivate RabbitTemplate rabbitTemplate;@PostConstructpublic void init(){rabbitTemplate.setReturnsCallback(new RabbitTemplate.ReturnsCallback() {@Overridepublic void returnedMessage(ReturnedMessage returned) {log.error("消息从交换机没有正确的投递到队列，原因是：{}",returned.getReplyText());}});}@Beanpublic void sendMsg(){Message message = MessageBuilder.withBody("hello world".getBytes()).build();rabbitTemplate.convertAndSend("exchange.return.01","info111",message);log.info("消息发送完毕，发送时间是：{}",new Date());}}

核心代码：

方式4：lambda表达式实现RabbitTemplate.ReturnsCallback

package com.power.service;import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageBuilder;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Service;import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.Date;@Service
@Slf4j
public class MessageService {@Resourceprivate RabbitTemplate rabbitTemplate;@PostConstructpublic void init(){rabbitTemplate.setReturnsCallback(returned-> {log.error("消息从交换机没有正确的投递到队列，原因是：{}",returned.getReplyText());});}@Beanpublic void sendMsg(){Message message = MessageBuilder.withBody("hello world".getBytes()).build();rabbitTemplate.convertAndSend("exchange.return.04","info111",message);log.info("消息发送完毕，发送时间是：{}",new Date());}}

核心

2.1.5、测试

启动程序，当消息从交换机 没有正确地 到达队列，则会触发该方法。

启动程序，如果消息从交换机 正确地 到达队列了，那么就不会触发该方法。