中企动力做网站怎么样,cn域名做seo,网站建设新闻发布注意什么,做国外网站 国外人能看到吗1.首先是创建项目
在一个父工程 mq_demo 的基础上建立两个子模块#xff0c;生产者模块publisher#xff0c;消费者模块 consumer 创建项目#xff1a; 建立成功#xff1a; 删除多余文件 创建子模块1#xff1a;publisher#xff08;生产者模块#xff09; 右键---…1.首先是创建项目
在一个父工程 mq_demo 的基础上建立两个子模块生产者模块publisher消费者模块 consumer 创建项目 建立成功 删除多余文件 创建子模块1publisher生产者模块 右键-----new ----module 选中Java填写publisher选中maven确认父模块 创建成功 同理创建子模块2consumer消费者模式 至此项目创建完毕
2.进行基本配置pom.xml、application.yml
引入依赖父模块引入依赖子模块共享父模块依赖 pom.xml dependencies!--AMQP依赖包含 rabbitmq --dependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-amqp/artifactId/dependency!--单元测试--dependencygroupIdjunit/groupIdartifactIdjunit/artifactIdversion4.13.2/versionscopetest/scope/dependencydependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter/artifactId/dependencydependencygroupIdorg.projectlombok/groupIdartifactIdlombok/artifactIdoptionaltrue/optional/dependencydependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-test/artifactIdscopetest/scope/dependency/dependenciesapplication.yml logging:pattern:dateformat: yyyy-MM-dd HH:mm:ss.SSSlevel:mq.listener: debug
spring:rabbitmq:host: 127.0.0.1port: 5672username: guestpassword: guestvirtual-host: /
结构图 建立具体的包结构以及要用的一些类 消费者启动类ConsumerApplication.class
SpringBootApplication
public class ConsumerApplication {public static void main(String[] args) {SpringApplication.run(ConsumerApplication.class, args);}}生产者启动类PublisherApplication.class
SpringBootApplication
public class PublisherApplication {public static void main(String[] args) {org.springframework.boot.SpringApplication.run(PublisherApplication.class, args);}
}消费者监听类SpringRabbitListerner.class
Component
public class SpringRabbitListerner {RabbitListener(queues queue.simple)public void listenSimpleQueueMessage(String msg){System.out.println(简单模式-消费者消费消息msg);}
}生产者启动类SpringAmqpTest.class
ExtendWith(SpringExtension.class)
SpringBootTest
public class SpringAmqpTest {Autowiredprivate RabbitTemplate rabbitTemplate;Testpublic void testSendSimpleMessage() {String simpleQueue queue.simple;String message hello world;rabbitTemplate.convertAndSend(simpleQueue,message);}
}做到这里就已经可以进行mq的消息进行发送和获取了。
3.Spring AMQP的五种工作模式
生产者启动类SpringAmqpTest.class
ExtendWith(SpringExtension.class)
SpringBootTest
public class SpringAmqpTest {Autowiredprivate RabbitTemplate rabbitTemplate;Testpublic void testSendSimpleMessage() {String simpleQueue queue.simple;String message hello world;rabbitTemplate.convertAndSend(simpleQueue,message);}Testpublic void testSendWorkQueue(){String workQueue queue.work;for(int i1 ;i10;i){String message hello world..i;rabbitTemplate.convertAndSend(workQueue,message);}}Testpublic void testSendFanout(){String fanoutExchange amq.fanout;String message hello world fanout..;rabbitTemplate.convertAndSend(fanoutExchange,,message);}Testpublic void testSendDirect(){String directExchange amq.direct;String message hello world direct..;rabbitTemplate.convertAndSend(directExchange,red,message);}}
消费者监听类SpringRabbitListerner.class
Component
public class SpringRabbitListerner {RabbitListener(queues queue.simple)public void listenSimpleQueueMessage(String msg){System.out.println(简单模式-消费者消费消息msg);}RabbitListener(queues queue.work)public void listenWorkQueueMessage1(String msg){System.out.println(工作模式-消费者消费消息msg);}RabbitListener(queues queue.work)public void listenWorkQueueMessage2(String msg){System.out.println(工作模式-消费者消费消息2msg);}RabbitListener(queues queue.fanout1)public void listenFanoutQueueMessage1(String msg){System.out.println(发布订阅模式-消费者1消费消息msg);}RabbitListener(queues queue.fanout2)public void listenFanoutQueueMessage2(String msg){System.out.println(发布订阅模式-消费者2消费消息msg);}RabbitListener(queues queue.direct1)public void listenDirectQueueMessage(String msg){System.out.println(路由模式-消费者消费消息msg);}RabbitListener(queues queue.direct2)public void listenTopicQueueMessage1(String msg){System.out.println(路由模式-消费者1消费消息msg);}}所用到的配置类 主要是建立交换机、建立队列、绑定交换机和队列关系的 订阅者模式
Configuration
public class FunoutConfig {Beanpublic FanoutExchange fanoutExchange() {return new FanoutExchange(amq.fanout);}Beanpublic Queue queue1() {return new Queue(queue.fanout1);}Beanpublic Queue queue2() {return new Queue(queue.fanout2);}Beanpublic Binding binding1(Queue queue1, FanoutExchange fanoutExchange) {return BindingBuilder.bind(queue1).to(fanoutExchange);}Beanpublic Binding binding2(Queue queue2, FanoutExchange fanoutExchange) {return BindingBuilder.bind(queue2).to(fanoutExchange);}}路由模式
Configuration
public class DirectConfig {Beanpublic DirectExchange directExchange(){return new DirectExchange(amq.direct);}Beanpublic Queue directQueue1(){return new Queue(queue.direct1);}Beanpublic Queue directQueue2(){return new Queue(queue.direct2);}Beanpublic Binding directBinding1(Queue queue1, DirectExchange directExchange){return BindingBuilder.bind(queue1).to(directExchange).with(yellow);}Beanpublic Binding directBinding2(Queue queue2, DirectExchange directExchange){return BindingBuilder.bind(queue2).to(directExchange).with(red);}}2025-6-13
4.mq的高级特性
1、消息的安全性问题----生产者、消费者确认、mq持久化
2、延迟队列----做定时任务
3、惰性队列----处理数据积压
3、模拟集群----处理单点崩溃4.1 消息队列的安全性问题 生产者确认生产者发送消息---到交换机到队列---返回ack给mq生产者发送消息---没到交换机---返回nack给mq生产者发送消息---到交换机没到队列---返回ack触发回调模块剥离代码部分 1.生产者确认机制 yml配置 logging:pattern:dateformat: yyyy-MM-dd HH:mm:ss.SSSlevel:com.example.demo: debug
spring:rabbitmq:host: 127.0.0.1port: 5672username: guestpassword: guestvirtual-host: /#springAMQP 实现生产者确认publisher-confirm-type: correlated #simple 同步等待confirm结果直到超时correlated异步回调定义confirmcollback,MQ返回结果时会回调ConfirmCallbackpublisher-returns: true # 开启publish-return 功能同样是基于callback机制不过是定义ReturnCallbacktemplate:mandatory: true # 定义消息路由失败时的策略。true则调用ReturnCallbackfalse则直接丢弃消息
消息到交换机—ack的回应
package mq.demo;import lombok.extern.slf4j.Slf4j;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;
import java.util.UUID;RunWith(SpringRunner.class)
SpringBootTest
Slf4j
public class SpringPublisherTest {Autowiredprivate RabbitTemplate rabbitTemplate;Testpublic void simpleSendMessage(){//1.准备消息String msg hello spring amqp;//2.准备CorrelationDate//2.1消息idCorrelationData correlationData new CorrelationData(UUID.randomUUID().toString());//2.2准备ConfirmCallbackrabbitTemplate.setConfirmCallback((correlationData1, ack, cause) - {if (ack) {log.info(消息确认成功ID: {}, correlationData1 ! null ? correlationData1.getId() : 无ID);} else {log.error(消息确认失败ID: {}, 原因: {}, correlationData1 ! null ? correlationData1.getId() : 无ID, cause);}});rabbitTemplate.convertAndSend(amq.direct,yellow,msg,correlationData);}
}消息到了交换机ack----消息没到队列 路由发送失败—触发回调
package mq.config;import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.annotation.Configuration;Slf4j
Configuration
public class CommonConfig implements ApplicationContextAware {Overridepublic void setApplicationContext(ApplicationContext applicationContext) throws BeansException {//获取rabbitmq对象RabbitTemplate rabbitTemplate applicationContext.getBean(RabbitTemplate.class);//配置 ReturnCallback 当发送者发送消息到MQMQ返回失败会调用此方法记住是路由发送失败rabbitTemplate.setReturnsCallback(msg - {//记录日志log.error(消息发送到队列失败 响应码{} 失败原因{} 交换机{} 路由key{},msg.getReplyCode(),msg.getReplyText(),msg.getExchange(),msg.getRoutingKey());});}
}持久化解决mq宕机造成的交换机丢失、队列丢失、消息丢失 消费者确认 消费者收到消息—消费者处理消息—消费者返回ack 消费者收到消息—消费者处理消息—消费者应该返回nack 针对第二种情况衍生出消费者确认机制 1.nonemq发完消息自动删除不等ack 2.手动处理try-catch然后自己捕捉异常捕捉到了返回给mq服务器nack 3.自动处理auto系统自己捕捉异常然后返回nack、 针对消费者确认机制可能出现循环的问题 解决办法配上retry机制设定本地重试 1.消费者自己处理耗尽次数自动放弃消息 2.耗尽次数返还消息给mq队列循环 3.耗尽次数将消息送到error交换机 模块剥离代码部分
logging:pattern:dateformat: yyyy-MM-dd HH:mm:ss.SSSlevel:mq.listener: debug
spring:rabbitmq:host: 127.0.0.1port: 5672username: guestpassword: guestvirtual-host: /listener:simple:prefetch: 1#acknowledge-mode: none #关闭ackMQ假定消费者获取消息后会成功处理。因此消息投递后立即被删除acknowledge-mode: auto #自动ack模式由spring检测listener代码是否出现异常没有异常则返回ack有异常则返回nack#acknowledge-mode: manual #手动ack,需要在业务代码结束后调用api发送ackretry:enabled: true # 开启消费者失败重试initial-interval: 1000ms #初试的失败等待时长为1秒multiplier: 3 # 失败重试的间隔倍数下次失败的等待时长倍数下次等待时长 multiplier * last-intervalmax-attempts: 4 # 最大失败重试次数
配置error交换机次数耗尽就把消息送到error交换机
package mq.demo.config;import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.retry.MessageRecoverer;
import org.springframework.amqp.rabbit.retry.RepublishMessageRecoverer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;Configuration
public class ErrorMessageConfig {Beanpublic DirectExchange errorMessageExchange() {return new DirectExchange(error.direct);}Beanpublic Queue errorMessageQueue() {return new Queue(error.queue);}Beanpublic Binding errorMessageBinding() {return new Binding(error.queue,Binding.DestinationType.QUEUE,error.direct,error,null);}Beanpublic MessageRecoverer republicMessageRecoverer(RabbitTemplate rabbitTemplate) {return new RepublishMessageRecoverer(rabbitTemplate, error.direct, error);}}4.2.延迟队列
初始死信交换机当一个队列中的消息满足下列情况之一时可以成为死信dead letter
1、消费者使用basic.reject或basic.nack声明消费失败并且消息的requeue的参数设置为false
2、消息是一个过期消息超时无人消费
3、要投递的队列消息堆积满了最早的消息可能成为死信
如果该队列配置了dead-letter-exchange属性指定了一个交换机那么队列中的死信就会投递到这个交换机这个交换机称为死信交换机Dead Letter Exchange简称DLX。
死信交换机的运行原理小总结和上面的有所重复 TTL延迟队列 TTL也就是Time-To-Live。如果一个队列中的消息TTL结束仍未消费则会变成死信ttl超时分为两种情况 1、消息所在的队列设置了存活时间 2、消息本身设置了存活时间 TTL的运行原理如图
模块剥离代码部分 首先绑定死信交换机和死信队列并设置消费者去消费消息 RabbitListener(bindings QueueBinding(value Queue(name dl.queue),exchange Exchange(value dl.direct),//默认是direct持久化key dl))public void listenDlQueue(String msg){log.info(测试死信队列接收到了延迟消息消息内容为{}msg);}
然后建立延迟交换机和延迟队列的绑定在延迟队列上再绑定死信交换机 附上代码
package mq.demo.config;import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;Configuration
public class TTLMessage {Beanpublic DirectExchange ttlExchange() {return new DirectExchange(ttl.direct);}Beanpublic Queue ttlQueue() {return QueueBuilder.durable(ttl.queue).ttl(10000).deadLetterExchange(dl.direct).deadLetterRoutingKey(dl).build();}Beanpublic Binding bindingTTLQueue(DirectExchange ttlExchange, Queue ttlQueue) {return BindingBuilder.bind(ttlQueue).to(ttlExchange).with(ttl);}
}发送消息代码
package mq.demo;import lombok.extern.slf4j.Slf4j;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageBuilder;
import org.springframework.amqp.core.MessageDeliveryMode;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;import java.nio.charset.StandardCharsets;
import java.util.UUID;RunWith(SpringRunner.class)
SpringBootTest
Slf4j
public class SpringPublisherTest {Autowiredprivate RabbitTemplate rabbitTemplate;Testpublic void testTTLMessage(){//1.准备消息Message message MessageBuilder.withBody(hello ttl amqp.getBytes(StandardCharsets.UTF_8))//.setExpiration(5000) 设置延迟时间.setDeliveryMode(MessageDeliveryMode.PERSISTENT).build();//2.发送消息rabbitTemplate.convertAndSend(ttl.direct,ttl,message);}
}mq呈现的状态和idea控制台的输出结果为 publisher生产者发送消息 ttl延迟队列首先收到消息 然后再指定时间没有人消息该消息消息自动转为死信并进入指定的延迟队列 然后被idea消费在idea控制台进行打印 评价就是我这里的思维逻辑是优点混乱的 1.我发送消息的前提是我mq里面得有已经建好的队列和交换机以及它们之间的绑定关系所以我应该是先启动消费者的启动类将这些交换机和队列存到bean工厂里面去。形成对象并且在mq里面呈现出来 2.我运行了很多次才弄好了 3.我项目如果是就是一个单一的项目不涉及微服务不分那么多微服务模块这个时候我应该怎么写。
4.3 惰性队列
消息堆积问题1.当生产者发送消息的速度超过了消费者处理消息的速度就会导致队列中的消息堆积直到队列存储消息达到上限。最早接收到的消息可能成为死信会被丢弃这就是消息堆积问题。2.解决消息堆积有三种思路1、增加更多消费者提高消费者消费速度就是5种模式的第二种--工作者模式work2、在消费者内开启线程池加快消息处理速度3、扩大队列容积提高堆积上限3.惰性队列特征1、接收到消息后直接存入磁盘而非内存2、消费者要消费消息时才会从磁盘中读取并加载到内存3、支持数百万条的消息存储惰性队列 两种声明方式 1、基于Bean
package mq.demo.config;import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.QueueBuilder;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;Configuration
public class LazyConfig {Beanpublic Queue lazyQueue() {return QueueBuilder.durable(lazy.queue).lazy().build();}Beanpublic Queue normalQueue() {return QueueBuilder.durable(normal.queue).build();}}
2、基于注解略 给建立好的两个队列各发送100万条信息 附上代码 Testpublic void testLazyMessage(){//2.发送消息for (int i 1; i 1000000; i){//1.准备消息Message message MessageBuilder.withBody(hello lazy amqp.getBytes(StandardCharsets.UTF_8)).setDeliveryMode(MessageDeliveryMode.NON_PERSISTENT).build();rabbitTemplate.convertAndSend(lazy.queue,message);}}Testpublic void testNormalMessage(){//2.发送消息for (int i 1; i 1000000; i){//1.准备消息Message message MessageBuilder.withBody(hello lazy amqp.getBytes(StandardCharsets.UTF_8)).setDeliveryMode(MessageDeliveryMode.NON_PERSISTENT).build();rabbitTemplate.convertAndSend(normal.queue,message);}}lazy队列的话他是直接发送到磁盘的不进内存 normal队列的话进内存大概3万条超过了就刷进磁盘 下面这张图数据里没有pageout数据有些不准确 normal的话应该是in memory进内存一部分超过了就进磁盘
4.4 集群模式暂时略
