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> <groupId>org.springframework.boot</groupId><artifactId>spring-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> <groupId>com.fasterxml.jackson.core</groupId> <artifactId>jackson-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(Map<String,Object> msg){System.out.println("消费者........接收到对象消息:【" + msg + "】" + LocalTime.now());
}