手机免费制作自己的网站,怎么用手机建网站,沈阳便宜做网站的,唐山哪个公司可以制作网站MQ
同步通信 异步通信 事件驱动优势#xff1a; 服务解耦 性能提升#xff0c;吞吐量提高 服务没有强依赖#xff0c;不担心级联失败问题 流量消峰
小结: 大多情况对时效性要求较高#xff0c;所有大多数时间用同步。而如果不需要对方的结果#xff0c;且吞吐…MQ
同步通信 异步通信 事件驱动优势 服务解耦 性能提升吞吐量提高 服务没有强依赖不担心级联失败问题 流量消峰
小结: 大多情况对时效性要求较高所有大多数时间用同步。而如果不需要对方的结果且吞吐量并发量较高则需要使用异步通信 MQ常见框架
MQ(MessageQueue)消息队列字面来看就是存放消息的队列也就是事件驱动架构中的Broker
消息就是事件比如支付成功了这个事件在MQ中就是一个消息 RabbitMQ,RocketMQ 适合处理业务若需要优化定制则选Rocket因为用Java写的
Kafka 适合处理日志海量数据且对数据安全性要求不高的场景ActiveMQ用的较少
RabbitMQ
RabbitMQ概述与安装
RabbitMQ是基于Erlang语言面向并发的语言天生为分布式系统而设计的开发的开源消息通信中间件官网地址https://www.rabbitmq.com/
参考课前资料(链接https://pan.baidu.com/s/1JuVKKFpUXg8TFxa_FoV3Gg 提取码1468) 来安装RabbitMQ 之后在浏览器输入http://192.168.83.130:15672/ 进入RabbitMQ管理页面按docker run中设置的账号密码进行登录
结果如下 mq整体架构 小结 常见消息模型 HelloWorld 案例 动手实践
案例 完成官方Demo中的hello world案例(链接https://pan.baidu.com/s/1JuVKKFpUXg8TFxa_FoV3Gg 提取码1468) 打开项目将ip调成自己的rabbitmq使用虚拟机或电脑的ip再运行一次PublisherTest中的 testSendMessage() 方法
发送一条消息。再运行ConsumerTest 中main方法来接收消息。 小结 SpringAMQP
AMOP(Advanced Message Queuing Protocol)高级消息队列协议大大简化消息发送和接收的代码量且与语言无关
SpringAmqp的官方地址https://spring.io/projects/spring-amqp AMQP依赖
!--AMQP依赖包含RabbitMQ--
dependency groupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-amqp/artifactId
/dependency在配置文件中添加mq连接信息
spring:rabbitmq:host: 192.168.83.130 # 主机名port: 5672 # 端口virtual-host: / # 虚拟主机 username: itcast # 用户名password: 123321 # 密码Basic Queue 简单队列模型
案例利用SpringAMQP实现HelloWorld中的基础消息队列功能
流程如下
1.在父工程中引入spring-amqp的依赖以及在publisher服务中编写配置
2.在publisher服务中利用RabbitTemplate的convertAndSend方法发送消息到simple.queue这个队列 SpringAMQP发送消息步骤引入依赖和设置配置----利用RabbitTemplate的convertAndSend方法
3.在consumer中编写代码接收消息 SpringAMQP接收消息步骤引入依赖和设置配置—》定义类添加Component注解类中声明方法添加RabbitListener注解
Work Queue 工作队列模型
Work queue工作队列可以提高消息处理速度避免队列消息堆积
比如队列 一秒来50条消息 一个消费者一秒处理40条消息那么需要两个消费者才能使得队列中消息被处理不丢失 案例实现一个队列绑定多个消费者 问题rabbitMQ消息预取会将50条消息平均分给消费者1和消费者2但消费者2处理速度慢因此在1s内处理不完publisher发过来的50条消息
解决方案让能者多劳设置preFetch,控制预取消息的上限 小结
发布、订阅模型-Fanout 注意exchange负责消息路由而不是存储queue负责存储路由失败则消息丢失
Fanout Exchange 会将接收到的消息路由到每一个跟其绑定的queue(广播)
案例利用SpringAMQP演示FanoutExchange的使用 step1 在consumer服务中声明Exchange、Queue、Binding绑定关系 step2 在consumer服务声明两个消费者
在consumer服务的SpringRabbitListener类中添加两个方法分别监听fanout.queue1和fanout.queue2
RabbitListener(queues fanout.queue1)
public void listenFanoutQueue1(String msg) {System.out.println(消费者1接收到Fanout消息【 msg 】);
}RabbitListener(queues fanout.queue2)
public void listenFanoutQueue2(String msg) {System.out.println(消费者2接收到Fanout消息【 msg 】);
}step3 在publisher服务发送消息到FanoutExchange
在publisher服务的SpringAmqpTest类中添加测试方法
Test
public void testFanoutExchange() {// 队列名称 String exchangeName itcast.fanout; // 消息String message hello, everyone!;// 发送消息参数分别是交互机名称、RoutingKey暂时为空、消息 rabbitTemplate.convertAndSend(exchangeName, , message);
}小结 发布、订阅模型-Direct 案例利用SpringAMQP演示DirectExchange的使用 步骤一 在consumer服务声明Exchange、Queue
1.在consumer服务中编写两个消费者方法分别监听direct.queue1和direct.queue2
2.并利用RabbitListener声明Exchange、Queue、RoutingKey
RabbitListener(bindings QueueBinding(value Queue(name direct.queue1),exchange Exchange(name itcast.direct, type ExchangeTypes.DIRECT),key {red,blue}
))
public void listenDirectQueue1(String msg){System.out.println(消费者1........接收到路由消息【 msg 】 LocalTime.now());
}RabbitListener(bindings QueueBinding(value Queue(name direct.queue2),exchange Exchange(name itcast.direct, type ExchangeTypes.DIRECT),key {red,yellow}
))
public void listenDirectQueue2(String msg){System.out.println(消费者2........接收到消路由息【 msg 】 LocalTime.now());
}步骤二 在publisher服务发送消息到DirectExchange
在publisher服务的SpringAmqpTest类中添加测试方法
Test
public void testDirectExchange() {//交换机名字String exchangeName itcast.direct;//消息String message 红色警报日本乱排核废水导致海洋生物变异惊现哥斯拉;//发送消息参数依次为交换机名称RoutingKey消息rabbitTemplate.convertAndSend(exchangeName, blue, message);
}从blue-yellow-red 运行三次得到结果如下 小结 发布、订阅模型-Topic 案例 利用SpringAMQP演示TopicExchange的使用 步骤一在consumer服务声明Exchange、Queue
1.在consumer服务中编写两个消费者方法分别监听topic.queue1和topic.queue2
2.并利用RabbitListener声明Exchange、Queue、RoutingKey
RabbitListener(bindings QueueBinding(value Queue(name topic.queue1),exchange Exchange(name itcast.topic, type ExchangeTypes.TOPIC),key china.#
))
public void listenTopicQueue1(String msg){System.out.println(消费者1........接收到路由消息【 msg 】 LocalTime.now());
}RabbitListener(bindings QueueBinding(value Queue(name topic.queue2),exchange Exchange(name itcast.topic, type ExchangeTypes.TOPIC),key #.news
))
public void listenTopicQueue2(String msg){System.out.println(消费者2........接收到消路由息【 msg 】 LocalTime.now());
}步骤二在publisher服务发送消息到TopicExchange
在publisher服务的SpringAmqpTest类中添加测试方法
Test
public void testTopicExchange() {//交换机名字String exchangeName itcast.topic;//消息String message 喜报孙悟空大战哥斯拉胜!;//发送消息参数依次为交换机名称RoutingKey消息rabbitTemplate.convertAndSend(exchangeName, china.news, message);
}小结 消息转化器
Spring的对消息对象的处理是由org.springframework.amqp.support.converter.MessageConverter来处理的。而默认实现是SimpleMessageConverter基于JDK的ObjectOutputStream完成序列化。如果要修改只需要定义一个MessageConverter 类型的Bean即可。推荐用JSON方式序列化步骤如下
在publisher服务引入依赖
dependency groupIdcom.fasterxml.jackson.core/groupId artifactIdjackson-databind/artifactId
/dependency 在publisher服务声明MessageConverter。原本应该放到配置类中但启动类也是配置类所以可以放启动类中
Bean
public MessageConverter jsonMessageConverter(){return new Jackson2JsonMessageConverter();
}案例 测试发送Object类型消息 结果如下(没有更改JDK序列化方式) 使用json序列化器之后 consumer接收消息过程
step1加jackson依赖依赖上面已经放父工程中就不用做了
step2: 将pulisher中相同的MessageConverter放入consumer 启动类中发送方与接收方必须相同
Bean
public MessageConverter jsonMessageConverter(){return new Jackson2JsonMessageConverter();
}step3: 定义一个消费者监听object.queue队列并消费消息 RabbitListener(queues object.queue)
public void listenObjectQueue(MapString,Object msg){System.out.println(消费者........接收到对象消息【 msg 】 LocalTime.now());
}
