MQ

同步通信

異步通信

事件驅(qū)動(dòng)優(yōu)勢(shì)：

• 服務(wù)解耦

  

• 性能提升，吞吐量提高

  

• 服務(wù)沒有強(qiáng)依賴，不擔(dān)心級(jí)聯(lián)失敗問題

  

• 流量消峰

  

? 小結(jié): 大多情況對(duì)時(shí)效性要求較高，所有大多數(shù)時(shí)間用同步。而如果不需要對(duì)方的結(jié)果，且吞吐量，并發(fā)量較高則需要使用異步通信

MQ常見框架

MQ(MessageQueue)，消息隊(duì)列，字面來(lái)看就是存放消息的隊(duì)列，也就是事件驅(qū)動(dòng)架構(gòu)中的Broker

消息：就是事件，比如支付成功了這個(gè)事件，在MQ中就是一個(gè)消息

RabbitMQ,RocketMQ 適合處理業(yè)務(wù)（若需要優(yōu)化定制則選Rocket，因?yàn)橛肑ava寫的）

Kafka 適合處理日志（海量數(shù)據(jù)且對(duì)數(shù)據(jù)安全性要求不高的場(chǎng)景），ActiveMQ用的較少

RabbitMQ

RabbitMQ概述與安裝

RabbitMQ是基于Erlang語(yǔ)言（面向并發(fā)的語(yǔ)言，天生為分布式系統(tǒng)而設(shè)計(jì)的）開發(fā)的開源消息通信中間件，官網(wǎng)地址：https://www.rabbitmq.com/

參考課前資料(鏈接：https://pan.baidu.com/s/1JuVKKFpUXg8TFxa_FoV3Gg
提取碼：1468) 來(lái)安裝RabbitMQ

之后在瀏覽器輸入：http://192.168.83.130:15672/ 進(jìn)入RabbitMQ管理頁(yè)面，按docker run中設(shè)置的賬號(hào)密碼進(jìn)行登錄

結(jié)果如下

mq整體架構(gòu)

小結(jié)

常見消息模型

HelloWorld 案例

動(dòng)手實(shí)踐

案例： 完成官方Demo中的hello world案例(鏈接：https://pan.baidu.com/s/1JuVKKFpUXg8TFxa_FoV3Gg
提取碼：1468)

打開項(xiàng)目，將ip調(diào)成自己的rabbitmq使用虛擬機(jī)（或電腦）的ip，再運(yùn)行一次PublisherTest中的 testSendMessage() 方法

發(fā)送一條消息。再運(yùn)行ConsumerTest 中main方法來(lái)接收消息。

小結(jié)

SpringAMQP

AMOP(Advanced Message Queuing Protocol)高級(jí)消息隊(duì)列協(xié)議，大大簡(jiǎn)化消息發(fā)送和接收的代碼量，且與語(yǔ)言無(wú)關(guān)

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 # 主機(jī)名port: 5672 # 端口virtual-host: / # 虛擬主機(jī) username: itcast # 用戶名password: 123321 # 密碼

Basic Queue 簡(jiǎn)單隊(duì)列模型

案例：利用SpringAMQP實(shí)現(xiàn)HelloWorld中的基礎(chǔ)消息隊(duì)列功能

流程如下：

1.在父工程中引入spring-amqp的依賴，以及在publisher服務(wù)中編寫配置

2.在publisher服務(wù)中利用RabbitTemplate的convertAndSend方法，發(fā)送消息到simple.queue這個(gè)隊(duì)列

SpringAMQP發(fā)送消息步驟：引入依賴和設(shè)置配置---->利用RabbitTemplate的convertAndSend方法

3.在consumer中編寫代碼，接收消息

SpringAMQP接收消息步驟：引入依賴和設(shè)置配置—》定義類，添加Component注解，類中聲明方法添加@RabbitListener注解

Work Queue 工作隊(duì)列模型

Work queue，工作隊(duì)列，可以提高消息處理速度，避免隊(duì)列消息堆積

比如隊(duì)列 一秒來(lái)50條消息 一個(gè)消費(fèi)者一秒處理40條消息，那么需要兩個(gè)消費(fèi)者才能使得隊(duì)列中消息被處理不丟失

案例：實(shí)現(xiàn)一個(gè)隊(duì)列綁定多個(gè)消費(fèi)者

問題：rabbitMQ消息預(yù)取，會(huì)將50條消息平均分給消費(fèi)者1和消費(fèi)者2，但消費(fèi)者2處理速度慢，因此在1s內(nèi)處理不完publisher發(fā)過(guò)來(lái)的50條消息

解決方案：讓能者多勞，設(shè)置preFetch,控制預(yù)取消息的上限

小結(jié)

發(fā)布、訂閱模型-Fanout

注意：exchange負(fù)責(zé)消息路由，而不是存儲(chǔ)（queue負(fù)責(zé)存儲(chǔ)），路由失敗則消息丟失

Fanout Exchange 會(huì)將接收到的消息路由到每一個(gè)跟其綁定的queue(廣播)

案例：利用SpringAMQP演示FanoutExchange的使用

step1 在consumer服務(wù)中聲明Exchange、Queue、Binding（綁定關(guān)系）

step2 在consumer服務(wù)聲明兩個(gè)消費(fèi)者

在consumer服務(wù)的SpringRabbitListener類中，添加兩個(gè)方法，分別監(jiān)聽fanout.queue1和fanout.queue2：

@RabbitListener(queues = "fanout.queue1")
public void listenFanoutQueue1(String msg) {System.out.println("消費(fèi)者1接收到Fanout消息：【" + msg + "】");
}@RabbitListener(queues = "fanout.queue2") 
public void listenFanoutQueue2(String msg) {System.out.println("消費(fèi)者2接收到Fanout消息：【" + msg + "】");
}

step3 在publisher服務(wù)發(fā)送消息到FanoutExchange

在publisher服務(wù)的SpringAmqpTest類中添加測(cè)試方法：

@Test
public void testFanoutExchange() {// 隊(duì)列名稱  String exchangeName = "itcast.fanout"; // 消息String message = "hello, everyone!";// 發(fā)送消息，參數(shù)分別是：交互機(jī)名稱、RoutingKey（暫時(shí)為空）、消息		rabbitTemplate.convertAndSend(exchangeName, "", message);
}

小結(jié)

發(fā)布、訂閱模型-Direct

案例：利用SpringAMQP演示DirectExchange的使用

步驟一 在consumer服務(wù)聲明Exchange、Queue

1.在consumer服務(wù)中，編寫兩個(gè)消費(fèi)者方法，分別監(jiān)聽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("消費(fèi)者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("消費(fèi)者2........接收到消路由息：【" + msg + "】" + LocalTime.now());
}

步驟二 在publisher服務(wù)發(fā)送消息到DirectExchange

在publisher服務(wù)的SpringAmqpTest類中添加測(cè)試方法：

@Test
public void testDirectExchange() {//交換機(jī)名字String exchangeName = "itcast.direct";//消息String message = "紅色警報(bào)！日本亂排核廢水，導(dǎo)致海洋生物變異，驚現(xiàn)哥斯拉！";//發(fā)送消息，參數(shù)依次為：交換機(jī)名稱，RoutingKey，消息rabbitTemplate.convertAndSend(exchangeName, "blue", message);
}

從blue->yellow->red 運(yùn)行三次，得到結(jié)果如下

小結(jié)

發(fā)布、訂閱模型-Topic

案例 利用SpringAMQP演示TopicExchange的使用

步驟一：在consumer服務(wù)聲明Exchange、Queue

1.在consumer服務(wù)中，編寫兩個(gè)消費(fèi)者方法，分別監(jiān)聽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("消費(fèi)者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("消費(fèi)者2........接收到消路由息：【" + msg + "】" + LocalTime.now());
}

步驟二：在publisher服務(wù)發(fā)送消息到TopicExchange

在publisher服務(wù)的SpringAmqpTest類中添加測(cè)試方法：

@Test
public void testTopicExchange() {//交換機(jī)名字String exchangeName = "itcast.topic";//消息String message = "喜報(bào)！孫悟空大戰(zhàn)哥斯拉，勝!";//發(fā)送消息，參數(shù)依次為：交換機(jī)名稱，RoutingKey，消息rabbitTemplate.convertAndSend(exchangeName, "china.news", message);
}

小結(jié)

消息轉(zhuǎn)化器

Spring的對(duì)消息對(duì)象的處理是由org.springframework.amqp.support.converter.MessageConverter來(lái)處理的。而默認(rèn)實(shí)現(xiàn)是SimpleMessageConverter，基于JDK的ObjectOutputStream完成序列化。如果要修改只需要定義一個(gè)MessageConverter 類型的Bean即可。推薦用JSON方式序列化，步驟如下：

? 在publisher服務(wù)引入依賴

<dependency>   <groupId>com.fasterxml.jackson.core</groupId>   <artifactId>jackson-databind</artifactId>
</dependency>

? 在publisher服務(wù)聲明MessageConverter。（原本應(yīng)該放到配置類中，但啟動(dòng)類也是配置類，所以可以放啟動(dòng)類中）

@Bean
public MessageConverter jsonMessageConverter(){return new Jackson2JsonMessageConverter(); 
}

案例 測(cè)試發(fā)送Object類型消息

結(jié)果如下(沒有更改JDK序列化方式)

使用json序列化器之后

consumer接收消息過(guò)程

step1：加jackson依賴，依賴上面已經(jīng)放父工程中，就不用做了

step2: 將pulisher中相同的MessageConverter放入consumer 啟動(dòng)類中（發(fā)送方與接收方必須相同）

@Bean
public MessageConverter jsonMessageConverter(){return new Jackson2JsonMessageConverter(); 
}

step3: 定義一個(gè)消費(fèi)者，監(jiān)聽object.queue隊(duì)列并消費(fèi)消息

 @RabbitListener(queues = "object.queue")
public void listenObjectQueue(Map<String,Object> msg){System.out.println("消費(fèi)者........接收到對(duì)象消息：【" + msg + "】" + LocalTime.now());
}