填写网站信息,初中毕业生怎么自考大专,网站建设体会心得,企业网站ui设计欣赏RabbitMQ如何保证发送的消息可靠#xff08;RabbitMQ的Confirm模式和2.Return模式#xff09; 1、RabbitMQ消息Confirm模式#xff08;保证从生产者到交换机的消息可靠#xff09;1.1、Confirm模式简介1.2、具体代码实现1.2.1、application.yml 开启确认模式1.2.2、生产者方… RabbitMQ如何保证发送的消息可靠RabbitMQ的Confirm模式和2.Return模式 1、RabbitMQ消息Confirm模式保证从生产者到交换机的消息可靠1.1、Confirm模式简介1.2、具体代码实现1.2.1、application.yml 开启确认模式1.2.2、生产者方式1实现RabbitTemplate.ConfirmCallback生产者发送消息方式2直接写在生产者发送消息类实现RabbitTemplate.ConfirmCallback方式3匿名内部类写法方式4lambda表达式写法 1.2.3、RabbitConfig做交换机和队列的绑定1.2.4、pom.xml配置文件1.2.5、测试 2、RabbitMQ消息Return模式保证从交换机的到队列的消息可靠2.1、具体代码实现2.1.1、applicaton.yml2.1.2、pom.xml2.1.3、启动类2.1.4、业务层方式1实现RabbitTemplate.ReturnsCallback回调类MyReturnCallback配置类service业务层 方式2MessageService类实现RabbitTemplate.ReturnsCallback方式3匿名内部类实现RabbitTemplate.ReturnsCallback方式4lambda表达式实现RabbitTemplate.ReturnsCallback2.1.5、测试 1、RabbitMQ消息Confirm模式保证从生产者到交换机的消息可靠
1.1、Confirm模式简介
消息的confirm确认机制是指生产者投递消息后到达了消息服务器Broker里面的exchange交换机exchange交换机会给生产者一个应答生产者接收到应答用来确定这条消息是否正常的发送到Broker的exchange中这也是消息可靠性投递的重要保障。
1.2、具体代码实现
1 配置文件application.yml 开启确认模式spring.rabbitmq.publisher-confirm-typecorrelated
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.011.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());}}方式4lambda表达式写法 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 version1.0 encodingUTF-8?
project xmlnshttp://maven.apache.org/POM/4.0.0 xmlns:xsihttp://www.w3.org/2001/XMLSchema-instancexsi:schemaLocationhttp://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsdmodelVersion4.0.0/modelVersiongroupIdcom.power/groupIdartifactIdrabbit_08_confirm01/artifactIdversion1.0-SNAPSHOT/versionnamerabbit_08_confirm01/namepropertiesmaven.compiler.source8/maven.compiler.sourcemaven.compiler.target8/maven.compiler.targetproject.build.sourceEncodingUTF-8/project.build.sourceEncoding/propertiesparentgroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-parent/artifactIdversion2.6.13/versionrelativePath//parentdependenciesdependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-web/artifactId/dependencydependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-test/artifactIdscopetest/scope/dependencydependencygroupIdorg.projectlombok/groupIdartifactIdlombok/artifactIdversion1.18.24/version/dependencydependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-amqp/artifactId/dependency/dependenciesbuildpluginsplugingroupIdorg.springframework.boot/groupIdartifactIdspring-boot-maven-plugin/artifactId/plugin/plugins/build/project1.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.012.1.2、pom.xml
?xml version1.0 encodingUTF-8?
project xmlnshttp://maven.apache.org/POM/4.0.0 xmlns:xsihttp://www.w3.org/2001/XMLSchema-instancexsi:schemaLocationhttp://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsdmodelVersion4.0.0/modelVersiongroupIdcom.power/groupIdartifactIdrabbit_09_return01/artifactIdversion1.0-SNAPSHOT/versionnamerabbit_09_return01/namepropertiesmaven.compiler.source8/maven.compiler.sourcemaven.compiler.target8/maven.compiler.targetproject.build.sourceEncodingUTF-8/project.build.sourceEncoding/propertiesparentgroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-parent/artifactIdversion2.6.13/versionrelativePath//parentdependenciesdependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-web/artifactId/dependencydependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-test/artifactIdscopetest/scope/dependencydependencygroupIdorg.projectlombok/groupIdartifactIdlombok/artifactIdversion1.18.24/version/dependencydependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-amqp/artifactId/dependency/dependenciesbuildpluginsplugingroupIdorg.springframework.boot/groupIdartifactIdspring-boot-maven-plugin/artifactId/plugin/plugins/build/project2.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());}
}方式2MessageService类实现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());}}核心代码
方式4lambda表达式实现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、测试
启动程序当消息从交换机 没有正确地 到达队列则会触发该方法。 启动程序如果消息从交换机 正确地 到达队列了那么就不会触发该方法。
