肇庆高要建设局网站,网站上那些兼职网页怎么做,网站建设的整体框架,网站服务器 电信2.8日学习打卡 一.springboot整合RabbitMQ 之前我们使用原生JAVA操作RabbitMQ较为繁琐#xff0c;接下来我们使用 SpringBoot整合RabbitMQ#xff0c;简化代码编写 创建SpringBoot项目#xff0c;引入RabbitMQ起步依赖 !-- RabbitMQ起步依赖 -- dependency接下来我们使用 SpringBoot整合RabbitMQ简化代码编写 创建SpringBoot项目引入RabbitMQ起步依赖 !-- RabbitMQ起步依赖 -- dependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starteramqp/artifactId /dependency编写配置文件 spring:rabbitmq:host: 192.168.66.100port: 5672username: jjypassword: jjyvirtual-host: / # 日志格式 logging:pattern:console: %d{HH:mm:ss.SSS} %clr(%-5level) --- [%-15thread] %cyan(%-50logger{50}):%msg%nSpringBoot整合RabbitMQ时需要在配置类创建队列和交换机 写法如下 package com.jjy.springrabbitmqdemo;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 RabbitConfig {private final String EXCHANGE_NAME boot_topic_exchange;private final String QUEUE_NAME boot_queue;//创建交换机Bean(bootExchange)public Exchange getExchange(){return ExchangeBuilder.topicExchange(EXCHANGE_NAME)//交换机类型.durable(true)//是否持久化.build();}//创建队列Bean(bootQueue)public Queue getMessageQueue() {return new Queue(QUEUE_NAME); // 队列名}//交换机绑定队列Beanpublic Binding bindMessageQueue(Qualifier(bootExchange) Exchange exchange, Qualifier(bootQueue) Queue queue){return BindingBuilder.bind(queue).to(exchange).with(#.message.#).noargs();} } SpringBoot整合RabbitMQ_编写生产者 SpringBoot整合RabbitMQ时提供了工具类RabbitTemplate发送 消息编写生产者时只需要注入RabbitTemplate即可发送消息。 package com.jjy.springrabbitmqdemo;import org.junit.jupiter.api.Test; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.test.context.SpringBootTest;SpringBootTest public class TestProducer {//注入RabbitTemplate工具类Autowiredprivate RabbitTemplate rabbitTemplate;Test/*** 发送消息* 参数1交换机* 参数2路由key* 参数3要发送的消息*/public void testSendMessage(){rabbitTemplate.convertAndSend(boot_topic_exchange,message,双十一开始了);} } SpringBoot整合RabbitMQ_编写消费者 消费者 package com.jjy.rabbitmqcosspring.consumer;import org.springframework.amqp.rabbit.annotation.RabbitListener; import org.springframework.stereotype.Component;//消费者 Component public class Consumer {//监听队列RabbitListener(queues boot_queue)public void listenMessage(String message){System.out.println(监听的消息: message);} } 整合后的代码就是不用自己去实例化创建连接工厂连接信道让spring容器来控制实例的创建到销毁。 代码的实现有生产者和消费者、还有配置类创建交换机跟队列及其绑定操作都独立为一个类共3个类yml文件中配置rabbitmq的一些属性。 Direct类型默认匹配发送 它会把消息路由到那些binding key与routing key完全匹配的Queue中。 它是一个一对一的模型一条消息一定会被发到指定的一个队列完全匹配。 配置代码 import org.springframework.amqp.core.Binding; import org.springframework.amqp.core.BindingBuilder; import org.springframework.amqp.core.TopicExchange; import org.springframework.amqp.core.Queue; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; Configuration public class RabbitDirectConfig {Beanpublic Queue directQueue(){//参数介绍//1.队列名 2.是否持久化 3.是否独占 4.自动删除 5.其他参数return new Queue(directQueue-One,false,false,false,null);}Beanpublic Queue directQueue2(){//参数介绍//1.队列名 2.是否持久化 3.是否独占 4.自动删除 5.其他参数return new Queue(directQueue-Two,false,false,false,null);}Beanpublic DirectExchange directExchange(){//参数介绍//1.交换器名 2.是否持久化 3.自动删除 4.其他参数return new DirectExchange(MqSendService-One,false,false,null);}Beanpublic Binding bingExchange(){return BindingBuilder.bind(directQueue()) //绑定队列.to(directExchange()) //队列绑定到哪个交换器.with(One); //绑定路由key,必须指定}Beanpublic Binding bingExchange2(){return BindingBuilder.bind(directQueue2()) //绑定队列.to(directExchange()) //队列绑定到哪个交换器.with(Two); //绑定路由key,必须指定} } Topic类型拓展匹配发送 它是Direct类型的一种扩展提供灵活的匹配规则。 routing key为一个句点号 . 分隔的字符串我们将被句点号“. ”分隔开的每一段独立的字符串称为一个单词如One.Twobinding key与routing key一样也是句点号 . 分隔的字符串binding key中可以存在两种特殊字符 * 与 # 用于做模糊匹配其中“*”用于匹配一个单词“#”用于匹配多个单词可以是零个 import org.springframework.amqp.core.Binding; import org.springframework.amqp.core.BindingBuilder; import org.springframework.amqp.core.TopicExchange; import org.springframework.amqp.core.Queue; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration;Configuration public class RabbitTopicConfig {Beanpublic Queue topicQueue(){//参数介绍//1.队列名 2.是否持久化 3.是否独占 4.自动删除 5.其他参数return new Queue(topicQueue-One,false,false,false,null);}Beanpublic Queue topicQueue2(){//参数介绍//1.队列名 2.是否持久化 3.是否独占 4.自动删除 5.其他参数return new Queue(topicQueue-Two,false,false,false,null);}Beanpublic TopicExchange topicExchange(){//参数介绍//1.交换器名 2.是否持久化 3.自动删除 4.其他参数return new TopicExchange(Topic-Ex,false,false,null);}Beanpublic Binding bingExchange(){return BindingBuilder.bind(topicQueue()) //绑定队列.to(topicExchange()) //队列绑定到哪个交换器.with(*.Two.*); //路由key,必须指定}Beanpublic Binding bingExchange2(){return BindingBuilder.bind(topicQueue2()) //绑定队列.to(topicExchange()) //队列绑定到哪个交换器.with(#); //路由key,必须指定} } Fanout 类型广播发送 它会把所有发送到该Exchange的消息路由到所有与它绑定的Queue中。 它是一种一对多的类型无法指定Binding Key发送的一条消息会被发到绑定的所有队列。 配置代码 import org.springframework.amqp.core.*; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration;Configuration public class RabbitFanoutConfig {Beanpublic Queue fanoutQueue(){//参数介绍//1.队列名 2.是否持久化 3.是否独占 4.自动删除 5.其他参数return new Queue(fanoutQueue-One,false,false,false,null);}Beanpublic Queue fanoutQueue2(){//参数介绍//1.队列名 2.是否持久化 3.是否独占 4.自动删除 5.其他参数return new Queue(fanoutQueue-Two,false,false,false,null);}Beanpublic FanoutExchange fanoutExchange(){//参数介绍//1.交换器名 2.是否持久化 3.自动删除 4.其他参数return new FanoutExchange(Fanout-Ex,false,false,null);}Beanpublic Binding bingExchange(){return BindingBuilder.bind(fanoutQueue()) //绑定队列.to(fanoutExchange()); //队列绑定到哪个交换器}Beanpublic Binding bingExchange2(){return BindingBuilder.bind(fanoutQueue()) //绑定队列.to(fanoutExchange()); //队列绑定到哪个交换器}} Headers键值对匹配不常用) headers类型的Exchange不依赖于routing key与binding key的匹配规则来路由消息而是根据发送的消息内容中的headers属性进行匹配。 在绑定Queue与Exchange时指定一组键值对当消息发送到ExchangeRabbitMQ会取到该消息的headers也是一个键值对的形式对比其中的键值对是否完全匹配Queue与Exchange绑定时指定的键值对如果完全匹配则消息会路由到该Queue否则不会路由到该Queue。 该类型不常用暂不提供代码。 Message消息 当执行诸如 basicPublish() 之类的操作时内容作为字节数组参数传递而其他属性作为单独的参数传入。 public class Message {private final MessageProperties messageProperties;private final byte[] body;public Message(byte[] body, MessageProperties messageProperties) {this.body body;this.messageProperties messageProperties;}public byte[] getBody() {return this.body;}public MessageProperties getMessageProperties() {return this.messageProperties;}... } MessageProperties 接口定义了几个常见的属性例如“messageId”“timestamp”、“contentType”等等。 还可以通过调用 setHeader(String key, Object value) 方法扩展这些属性 二. 消息的可靠性投递 RabbitMQ消息投递的路径为 生产者 — 交换机 — 队列 — 消费者 在RabbitMQ工作的过程中每个环节消息都可能传递失败那么RabbitMQ是如何监听消息是否成功投递的呢 确认模式confirm可以监听消息是否从生产者成功传递到交换机。退回模式return可以监听消息是否从交换机成功传递到队列。消费者消息确认Consumer Ack可以监听消费者是否成功处理消息。 三种模式刚好监听完RabbitMQ的一整套流程。即我们能够由这三种模式得到消息的传递及处理的结果。 确认模式confirm 确认模式confirm可以监听消息是否从生产者成功传递到交换机 生产者配置文件开启确认模式 rabbitmq:host: 192.168.66.100port: 5672username: jjypassword: jjyvirtual-host: /# 开启确认模式publisher-confirm-type: correlatedpackage com.jjy.rabbitproducer;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 RabbitmqConfig {private final String EXCHNAGE_NAMEmy_topic_exchange;private final String QUEUE_NAMEmy_queue;Bean(bootExchange)public Exchange getExchange(){return ExchangeBuilder.topicExchange(EXCHNAGE_NAME)//交换机类型.durable(true).build();}// 2.创建队列Bean(bootQueue)public Queue getMessageQueue(){return QueueBuilder.durable(QUEUE_NAME) // 队列持久化.build();}Beanpublic Binding bindMessageQueue(Qualifier(bootExchange) Exchange exchange, Qualifier(bootQueue) Queue queue){return BindingBuilder.bind(queue).to(exchange).with(my_routing).noargs();} } SpringBootTest public class ProduceTest {Autowiredprivate RabbitTemplate rabbitTemplate;Testpublic void sendMessage(){// 定义确认模式的回调方法消息向交换机发送后会调用confirm方法rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {/*** 被调用的回调方法* param correlationData 相关配置信息* param ack 交换机是否成功收到了消息* param cause 失败原因*/Overridepublic void confirm(CorrelationData correlationData, boolean ack, String cause) {if (ack){System.out.println(confirm接受成功!);}else{System.out.println(confirm接受失败原因为cause);// 做一些处理。}}});rabbitTemplate.convertAndSend(my_topic_exchange,my_routing,send message...);} }退回模式return 退回模式return可以监听消息是否从交换机成功传递到队列 使用方法如下 生产者配置文件开启退回模式 spring:rabbitmq:host: 192.168.66.100port: 5672username: jjypassword: jjyvirtual-host: /# 开启确认模式publisher-confirm-type: correlated# 开启回退模式publisher-returns: truepackage com.jjy.rabbitproducer;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 RabbitmqConfig {private final String EXCHNAGE_NAMEmy_topic_exchange;private final String QUEUE_NAMEmy_queue;Bean(bootExchange)public Exchange getExchange(){return ExchangeBuilder.topicExchange(EXCHNAGE_NAME)//交换机类型.durable(true).build();}// 2.创建队列Bean(bootQueue)public Queue getMessageQueue(){return QueueBuilder.durable(QUEUE_NAME) // 队列持久化.build();}Beanpublic Binding bindMessageQueue(Qualifier(bootExchange) Exchange exchange, Qualifier(bootQueue) Queue queue){return BindingBuilder.bind(queue).to(exchange).with(my_routing).noargs();} }Testpublic void testReturn(){// 定义退回模式的回调方法。交换机发送到队列失败后才会执行returnedMessage方法rabbitTemplate.setReturnsCallback(new RabbitTemplate.ReturnsCallback(){/*** param returned 失败后将失败信息封装到参数中*/Overridepublic void returnedMessage(ReturnedMessage returned) {System.out.println(消息对象returned.getMessage());System.out.println(错误码returned.getReplyCode());System.out.println(错误信息returned.getReplyText());System.out.println(交换机returned.getExchange());System.out.println(路由键returned.getRoutingKey());}});rabbitTemplate.convertAndSend(my_topic_exchange,my_routing1,send message...);}消费者消息确认Ack 在RabbitMQ中消费者接收到消息后会向队列发送确认签收的消息只有确认签收的消息才会被移除队列。这种机制称为消费者消息确认Consumer Acknowledge简称Ack。类似快递员派送快递也需要我们签收否则一直存在于快递公司的系统中。 消息分为自动确认和手动确认。自动确认指消息只要被消费者接收到无论是否成功处理消息则自动签收并将消息从队列中移除。但是在实际开发中收到消息后可能业务处理出现异常那么消息就会丢失。此时需要设置手动签收即在业务处理成功再通知签收消息如果出现异常则拒签消息让消息依然保留在队列当 中。 自动确认spring.rabbitmq.listener.simple.acknowledge“none”手动确认spring.rabbitmq.listener.simple.acknowledge“manual” 消费者配置开启手动签收 spring:rabbitmq:host: 192.168.0.162port: 5672username: itbaizhanpassword: itbaizhanvirtual-host: /# 开启手动签收listener:simple:acknowledge-mode: manualpackage com.jjy.rabbitconsumer;import com.rabbitmq.client.Channel; import org.springframework.amqp.core.Message; import org.springframework.amqp.rabbit.annotation.RabbitListener; import org.springframework.stereotype.Component;import java.io.IOException;Component public class AckConsumer { // RabbitListener(queues my_queue) // public void listenMessage(String Mesage){ // int i1/0; // System.out.println(成功接收到消息Mesage); // } //RabbitListener(queues my_queue) public void listenMessage(Message message, Channel channel) throws InterruptedException, IOException {//消息投递序号消息每次投递该值都会1long deliveryTag message.getMessageProperties().getDeliveryTag();try{int i1/0;System.out.println(成功接收到消息message);// 签收消息/*** 参数1消息投递序号* 参数2是否一次可以签收多条消息*/channel.basicAck(deliveryTag,true);} catch (Exception e){System.out.println(消息消费失败);Thread.sleep(2000);// 拒签消息/*** 参数1消息投递序号* 参数2是否一次可以拒签多条消息* 参数3拒签后消息是否重回队列*/channel.basicNack(deliveryTag,true,true);}}} 三.RabbitMQ高级特性 消费端限流 之前我们讲过MQ可以对请求进行“削峰填谷”即通过消费端限流的方式限制消息的拉取速度达到保护消费端的目的。 消费端限流的写法如下 1 生产者批量发送消息 Test public void testSendBatch() {// 发送十条消息for (int i 0; i 10; i) {rabbitTemplate.convertAndSend(my_topic_e xchange, my_routing, send message...i);} }2 消费端配置限流机制 spring:rabbitmq:host: 192.168.66.100port: 5672username: jjypassword: jjyvirtual-host: /listener:simple:# 限流机制必须开启手动签收acknowledge-mode: manual# 消费端最多拉取5条消息消费签收后不满5 条才会继续拉取消息。prefetch: 53.消费者监听队列 Component public class OosConsimer {//RabbitListener(queues my_queue)public void listenMessage(Message message, Channel channel) throws IOException, InterruptedException {// 1.获取消息System.out.println(new String(message.getBody()));// 2.业务处理Thread.sleep(3000);//3.签收long deliveryTag message.getMessageProperties().getDeliveryTag();channel.basicAck(deliveryTag,true);} }就是说从生产端发送过来的消息在队列等待消费端接收如果消费端处理消息业务的速度相对较慢积累的消息过多从而处理不过来资源耗尽会导致系统性能降低或瘫痪。 因为消费端每秒处理消息的条数有限所以我们需要在消费端进行一个限流故而限制了队列消息的投递。 即消费端限流也就是限制队列投递到消费端的流也可以说是在队列与消费端之间进行一个限流。 利用限流实现不公平分发 在RabbitMQ中多个消费者监听同一条队列则队列默认采用的轮询分发。但在某种场景下这种策略并不是很好例如消费者1处理任务的速度非常快而其他消费者处理速度却很慢。此时如果采用公平分发则消费者1有很大一部分时间处于空闲状态。此时可以采用不公平分发即谁处理的快谁处理的消息多。 公平分发则不能在yml文件中设置限流(prefetch)公平分发即给多个消费者平分消息进行消费。这样会导致处理快的消费者在等待故而浪费资源降低性能。 不公平分发则需要在yml文件中设置限流(prefetch)并且prefetch: 1即设置为1;不公平分发即每次拉取一条消息谁处理得快就继续处理这样可以极大的节约资源从而提高性能。 1 生产者批量发送消息 Test public void testSendBatch() {// 发送十条消息for (int i 0; i 10; i) {rabbitTemplate.convertAndSend(my_topic_e xchange, my_routing, send message...i);} }2 消费端配置不公平分发 spring:rabbitmq:host: 192.168.66.100port: 5672username: jjypassword: jjyvirtual-host: /listener:simple:# 限流机制必须开启手动签收acknowledge-mode: manual# 消费端最多拉取1条消息消费这样谁处理 的快谁拉取下一条消息实现了不公平分发prefetch: 13 编写两个消费者 Component public class UnfairConsumer {// 消费者1RabbitListener(queues my_queue)public void listenMessage1(Message message, Channel channel) throws Exception{//1.获取消息System.out.println(消费者1:newString(message.getBody(),UTF-8));//2. 处理业务逻辑Thread.sleep(500); // 消费者1处理快//3. 手动签收channel.basicAck(message.getMessageProperties().getDeliveryTag(),true);// 消费者2RabbitListener(queues my_queue)public void listenMessage2(Messagemessage, Channel channel) throws Exception{//1.获取消息System.out.println(消费者2:newString(message.getBody(),UTF-8));//2. 处理业务逻辑Thread.sleep(3000);// 消费者2处理慢//3. 手动签收channel.basicAck(message.getMessageProperties().getDeliveryTag(),true);}}消息存活时间 RabbitMQ可以设置消息的存活时间Time To Live简称TTL当消息到达存活时间后还没有被消费会被移出队列。RabbitMQ可以对队列的所有消息设置存活时间也可以对某条消息设置存活时间 设置队列所有消息存活时间 就是说需要在配置类RabbitConfig中设置队列所有消息的存活时间 return QueueBuilder.durable(QUEUE_NAME)//队列持久化.ttl(10000)//设置队列的所有消息存活10s.build();即在创建bean队列时就要设置队列所有消息的存活时间。 **设置某条消息的存活时间 ** 就是说只需要在发送的时候指定它的存活时间即可。 实现比较稍微麻烦一点创建消息属性并设置存活时间然后创建消息对象消息对象 将消息属性作为参数并且传入发送的消息最后再将消息对象作为参数传给交换机即可实现对单条消息设置存活时间。 //发送消息并设置该消息的存活时间Testpublic void testSendMessage(){//1.创建消息属性MessageProperties messageProperties new MessageProperties();//2.设置存活时间messageProperties.setExpiration(10000);//3.创建消息对象Message message new Message(sendMessage....getBytes(),messageProperties);//4.发送消息rabbitTemplate.convertAndSend(my_topic_exchange1,my_routing,message);} 若设置中间的消息的存活时间当过期时该消息不会被移除但是该消息已经不会被消费了需要等到该消息到队里顶端才会被移除。 因为队列是头出尾进故而要移除它需要等到它在顶端时才可以。 在队列设置存活时间也在单条消息设置存活时间则以时间短的为准。 消息过期后并不会马上移除消息只有消息消费到队列顶 端时才会移除该消息。 Testpublic void testSendMessage2() {for (int i 0; i 10; i) {if (i 5) {// 1.创建消息属性MessageProperties messageProperties new MessageProperties();// 2.设置存活时间messageProperties.setExpiration(10000 );// 3.创建消息对象Message message new Message((send message... i).getBytes(), messageProperties);// 4.发送消息rabbitTemplate.convertAndSend(my_topi c_exchange, my_routing, message);} else {rabbitTemplate.convertAndSend(my_topi c_exchange, my_routing, send message... i);}}}在以上案例中i5的消息才有过期时间10s后消息并没有马上被移除但该消息已经不会被消费了当它到达队列顶端时会被移除。 优先级队列 假设在电商系统中有一个订单催付的场景即客户在一段时间内未付款会给用户推送一条短信提醒但是系统中分为大型商家和小型商家。比如像苹果小米这样大商家一年能给我们创造很大的利润所以在订单量大时他们的订单必须得到优先处理此时就需要为不同的消息设置不同的优先级此时我们要使用优先级队列 优先级队列用法如下 1 创建队列和交换机 Configuration public class RabbitmqConfig3 {private final String EXCHANGE_NAMEpriority_exchange;private final String QUEUE_NAMEpriority_queue;Bean(EXCHANGE_NAME)public Exchange priorityExchange(){return ExchangeBuilder.topicExchange(EXCHANGE_NAME)//交换机类型.durable(true)//是否持久化.build();}Bean(QUEUE_NAME)public Queue producerQueue(){return QueueBuilder.durable(QUEUE_NAME)//队列持久化//设置队列的最大优先级最大可以设置到255官网推荐不要超过10,如果设置太高比较浪费资源.maxPriority(10).build();}Beanpublic Binding bindPriority(Qualifier(EXCHANGE_NAME) Exchange exchange, Qualifier(QUEUE_NAME) Queue queue){return BindingBuilder.bind(queue).to(exchange).with(my_routing).noargs();} }2 编写生产者 Testpublic void testPriority() {for (int i 0; i 10; i) {if (i 5) {// i为5时消息的优先级较高MessageProperties messageProperties new MessageProperties();messageProperties.setPriority(9);Message message new Message((send message... i).getBytes(StandardCharsets.UTF_8), messageProperties);rabbitTemplate.convertAndSend(priority_exchange, my_routing, message);} else {rabbitTemplate.convertAndSend(priority_exchange, my_routing, send message... i);}}}3 编写消费者 Component public class PriorityConsumer {RabbitListener(queues priority_queue)public void listenMessage(Message message, Channel channel) throws IOException {System.out.println(new String(message.getBody(),utf-8));//手动签收channel.basicAck(message.getMessageProperties().getDeliveryTag(),true);} }如果我的内容对你有帮助请点赞评论收藏。创作不易大家的支持就是我坚持下去的动力