1.同步通訊和異步通訊

????????舉個例子，同步通訊就像是在打電話，因此它時效性較強(qiáng)，可以立即得到結(jié)果，但如果你正在和一個MM打電話，其他MM找你的話，你們之間是不能進(jìn)行消息的傳遞和響應(yīng)的

????????異步通訊就像是微信，你可以和多個MM進(jìn)行消息的傳遞，對方可以立即響應(yīng)或者有空了在”回“你

????????同步調(diào)用問題

????????微服務(wù)間基于Feign的調(diào)用就屬于同步方式，存在一些問題。

????????1.代碼耦合：如果后續(xù)需要添加業(yè)務(wù)，需要不斷修改支付服務(wù)的代碼

????????2.性能下降，吞吐量下降：支付服務(wù)一直在等待，訂單等服務(wù)的響應(yīng)，cpu資源一直在占用。

????????3.級聯(lián)失敗：如果服務(wù)提供者出現(xiàn)問題，所有調(diào)用方都會跟著出問題。

2.異步調(diào)用方案

????????異步調(diào)用常見的就是事件驅(qū)動模式：Broker是事件代理者，一旦用戶支付成功，這就是一個事件，這個事件就會被Broker管理，訂單服務(wù)、倉儲服務(wù)、短信服務(wù)。就會找Broker，這個叫做訂閱事件。一旦用戶支付成功之后，Broker就會通知被訂閱過事件的服務(wù)，支付服務(wù)完成事件發(fā)布之后，就結(jié)束了服務(wù)，返回給用戶

????????優(yōu)點(diǎn)：

????????解決代碼解耦：添加業(yè)務(wù)時不需要再更改支付服務(wù)的代碼，支付服務(wù)只需要發(fā)布事件就行。至于后面的業(yè)務(wù)支付服務(wù)可以不用考慮。

????????性能提升，吞吐量提供：相比較同步服務(wù)，業(yè)務(wù)處理時間的累加，支付服務(wù)還需要等待其他服務(wù)完成并響應(yīng)，通過異步的方式，支付服務(wù)發(fā)布時間之后就結(jié)束服務(wù)，無需等待其他服務(wù)響應(yīng)。提升了性能，和吞吐量

????????服務(wù)沒有依賴關(guān)系，不用擔(dān)心級聯(lián)失敗問題

????????流量削峰： 有多個事件發(fā)布，可以囤積到broker上，訂閱該事件的服務(wù)可以按自己的處理能力來穩(wěn)步進(jìn)行。broker起到緩沖作用

????????缺點(diǎn)：

????????依賴Broker的可靠性、安全性、吞吐能力

????????架構(gòu)復(fù)雜了，業(yè)務(wù)沒有明顯的流程線，不好追蹤管理

3.什么是MQ

MQ（MessageQueue），中文是消息隊(duì)列，字面來看就是存放消息的隊(duì)列。也就是事件驅(qū)動架構(gòu)中的Broker

常見的MQ

RabbitMQ，適用于中小型企業(yè)開發(fā)，如果對性能要求比較高的并且需要定制服務(wù)的大型企業(yè)推薦使用Kafka。下面會介紹RabbitMQ的使用。

4.RabbitMQ概述和安裝

RabbitMQ概述

RabbitMQ是基于Erlang語言開發(fā)的開源消息通信中間件，官網(wǎng)地址：https://www.rabbitmq.com

RabbitMQ的部署

環(huán)境：centos7，docker在線拉取的方式部署。

#拉取RabbitMQ鏡像

輸入：docker pull rabbitmq:3-management

#設(shè)置默認(rèn)用戶名密碼并啟動容器

docker run --name rabbitmq -e RABBITMQ_DEFAULT_USER=root -e RABBITMQ_DEFAULT_PASS=123 -p 15672:15672 -p 5672:5672 -d rabbitmq:3-management

端口15672是rabbitMQ的ui界面，5672是服務(wù)端口

ui界面展示：

RabbitMQ結(jié)構(gòu)和概念

PubLisher是我們的消息（事件）發(fā)送者，consumer是消息的消費(fèi)者，PubLisher將來會把消息發(fā)送到我們的exchange（交換機(jī)）上，交換機(jī)負(fù)責(zé)路由，并把消息投射到queue（隊(duì)列），queue負(fù)責(zé)暫存消息，consumer負(fù)責(zé)從queue里面獲取消息處理消息。

?

5.RabbitMQ的常見消息模型

一、基本消息隊(duì)列（BasicQueue）

HelloWorld是最基本的消息隊(duì)列模型，實(shí)現(xiàn)的話，包含三個角色

publisher：消息發(fā)布者，將消息發(fā)送到隊(duì)列queue

queue：消息隊(duì)列，負(fù)責(zé)接受并緩存消息

consumer：訂閱隊(duì)列，處理隊(duì)列中的消息

官方提供的編碼方式非常麻煩，下面我們介紹學(xué)習(xí)一下SpringAMQP，它可以大大簡化我們消息發(fā)送和接收API。

SpringAMQP簡介

AMQP：是用于在應(yīng)用程序或之間傳遞業(yè)務(wù)消息的開放標(biāo)準(zhǔn)，該協(xié)議與語言的平臺無關(guān)，更符合微服務(wù)中獨(dú)立性的要求。

Spring AMQP：是基于AMQP協(xié)議定義的一套API規(guī)范，提供了模板來發(fā)送和接收消息，包含兩部分，其中spring-amqp是基礎(chǔ)抽象，spring-rabbit是底層的默認(rèn)實(shí)現(xiàn)。

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

通過rabbitTemplate提供的convertAndSend就可以實(shí)現(xiàn)消息的發(fā)送。

引入相關(guān)依賴

<dependency><groupId>org.springframework.amqp</groupId><artifactId>spring-rabbit</artifactId><scope>test</scope>
</dependency>

publisher（消息發(fā)布者）的application.yml的配置

test的測試代碼：

@RunWith(SpringRunner.class)
@SpringBootTest
class PublisherApplicationTests {@Autowiredprivate RabbitTemplate rabbitTemplate;@Testvoid contextLoads() {String queuename="simple queue";String message="hello,spring AMQP";rabbitTemplate.convertAndSend(queuename,message);}}

通過rabbitTemplate實(shí)現(xiàn)對隊(duì)列消息的監(jiān)聽

引入依賴

<dependency><groupId>org.springframework.amqp</groupId><artifactId>spring-rabbit</artifactId><scope>test</scope>
</dependency>

配置consumer（消息接收者），的application.yml文件

spring:rabbitmq:host: 192.168.10.8 #主機(jī)名port: 5672         #端口username: root      #用戶名password: 123       #密碼virtual-host: /     #虛擬主機(jī)名

監(jiān)聽類的代碼

@Component
public class SimpleListener {@RabbitListener(queues = "simple queue")public void ListenSimpleQueue(String msg){System.out.println("消費(fèi)者接收到simple queue的消息："+msg);}
}

ps：消息一旦消費(fèi)就會從隊(duì)列中刪除，RabbitMQ沒有消息回溯功能。

二、工作消息隊(duì)列（WorkQueue）

工作隊(duì)列的結(jié)構(gòu)如下：

工作消息隊(duì)列的結(jié)構(gòu)，相比于基礎(chǔ)消息隊(duì)列多了個消費(fèi)者，因?yàn)閞abbitMQ閱后即焚的特性這兩個消費(fèi)者屬于共同工作的關(guān)系，如果有50個消息，他們兩個消費(fèi)者就會一人分一半，也就是一人25條，為啥會多一個消費(fèi)者？這是因?yàn)槿绻粋€消費(fèi)者每次處理40個消息，但是publisher一次發(fā)布50個消息，多出來的消息會存儲在queue里面，又因?yàn)閝ueue是占用內(nèi)存的假以時日，內(nèi)存就會爆滿，新的消息就存不進(jìn)去了，多一個消費(fèi)者每次就可以處理80條消息，可以有效解決這個問題。

work queue，工作隊(duì)列，可以提高消費(fèi)處理速度，避免隊(duì)列消息堆積。但是這里有一個消息預(yù)取機(jī)制 ，消費(fèi)者會提前把消息拿過來，因此消息是平局分配，并不是“能者多勞”的模式，通過設(shè)置prefetch的值來實(shí)現(xiàn)每次只能獲取一條消息，處理完成才能接取下一個消息。實(shí)現(xiàn)“能者多勞”的模式。

1. 發(fā)布訂閱（Publish、Subscribe）

基礎(chǔ)消息隊(duì)列和工作消息隊(duì)列都是一條信息只被一個消費(fèi)者消費(fèi)，消費(fèi)完就刪除，顯然不能實(shí)現(xiàn)我們之前預(yù)想的完成支付之后，通知倉儲、短信等服務(wù)。這就需要我們了解學(xué)習(xí)發(fā)布訂閱模式。發(fā)布訂閱模式與之前案例的區(qū)別就是允許同一消息發(fā)送給多個消費(fèi)者，實(shí)現(xiàn)方式是加入exchange（交換機(jī)），結(jié)構(gòu)如下：

publisher將消息發(fā)送給exchange（交換機(jī)），交換機(jī)把這個消息轉(zhuǎn)發(fā)給隊(duì)列，因此，發(fā)布者（publisher）并不需要知到轉(zhuǎn)發(fā)給了那個 隊(duì)列或多個隊(duì)列，轉(zhuǎn)發(fā)給多個隊(duì)列，這種方式就能實(shí)現(xiàn)被多個消費(fèi)者消費(fèi)，那么交換機(jī)到底是發(fā)給一個還是多個呢？這是由交換機(jī)類型來決定的。常見的exchange的類型包括：

廣播：Fanout

路由：Direct

主題：Topic

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

廣播-Fanout Exchange

Fanout Exchange 會將接收到的消息路由到每一個綁定的queue。

實(shí)現(xiàn)思路：

1.在consumer服務(wù)中，利用代碼聲明隊(duì)列、交換機(jī)、并將兩者綁定。

在consumer服務(wù)上添加@Configuration注解，并聲明FanoutExchange、Queue和關(guān)系對象Binding，代碼如下：

@Configuration
public class FanoutConfig {//聲明交換機(jī)對象@Beanpublic FanoutExchange fanoutExchange(){return new FanoutExchange("fanout");}//聲明隊(duì)列1@Beanpublic Queue fanoutqueue1(){return new Queue("fanoutqueue1");}//綁定隊(duì)列一到交換機(jī)@Beanpublic Binding fanoutBinding1(Queue fanoutqueue1,FanoutExchange fanoutExchange){return BindingBuilder.bind(fanoutqueue1).to(fanoutExchange);}//聲明隊(duì)列2@Beanpublic Queue fanoutqueue2(){return new Queue("fanoutqueue2");}//綁定隊(duì)列二到交換機(jī)@Beanpublic Binding fanoutBinding2(Queue fanoutqueue2,FanoutExchange fanoutExchange){return BindingBuilder.bind(fanoutqueue2).to(fanoutExchange);}
}

2.在consumer服務(wù)中，編寫兩個消費(fèi)者方法，分別監(jiān)聽fanout.queue1和fanout.queue2

@RabbitListener(queues = "fanoutqueue1")
public void ListenSimpleQueue3(String msg) throws InterruptedException {System.out.println("消費(fèi)者222接收到fanoutqueue1的消息："+msg);Thread.sleep(100);
}
@RabbitListener(queues = "fanoutqueue2")
public void ListenSimpleQueue4(String msg) throws InterruptedException {System.out.println("消費(fèi)者222接收到fanoutqueue2的消息："+msg);Thread.sleep(100);
}

3.在publisher中編寫測試方法，向fanout發(fā)送消息。

@Test
public void contectFonoutExchange()
{//交換機(jī)名稱String exchangeName="fanout";//消息內(nèi)容String message="hello everyone";//發(fā)送消息rabbitTemplate.convertAndSend(exchangeName,"",message);
}

路由-DirectExchange

Direct Exchange 會將接收到的消息根據(jù)規(guī)則路由到指定的Queue，因此稱為路由模式（routes）

每一個Queue都與Exchange設(shè)置一個BindingKey，一個隊(duì)列可以綁定多個BindingKey。

發(fā)布者發(fā)送消息時，指定消息的RoutingKey

Exchange（交換機(jī)）將消息路由到Bindingkey與消息RoutingKey一致的隊(duì)列

實(shí)現(xiàn)思路：

1.利用@RabbitListenner聲明Exchange、Queue、RountingKey

2.在consumer服務(wù)中編寫兩個消費(fèi)者方法，分別監(jiān)聽queue1和queue2

在我們的監(jiān)聽類里面增加兩個方法，用來聲明交換機(jī)、隊(duì)列和RountingKey

//發(fā)布訂閱DirectExchange
@RabbitListener(bindings = @QueueBinding(value = @Queue(name = "queue1"),exchange = @Exchange(name = "direct",type= ExchangeTypes.DIRECT),key = {"red","blue"}
))
public void listenerDirectqueue1(String msg)
{System.out.println("消費(fèi)者接收到direct.queue1的消息"+msg);
}@RabbitListener(bindings = @QueueBinding(value = @Queue(name = "queue2"),exchange = @Exchange(name = "direct",type= ExchangeTypes.DIRECT),key = {"red","yellow"}
))
public void listenerDirectqueue2(String msg)
{System.out.println("消費(fèi)者接收到direct.queue2的消息"+msg);
}

3.在publisher中編寫測試方法，向Exchange發(fā)送消息。

@Test
public void contextDirectExchange()
{String exchangeName="direct";String msgblue="hello blue";String msgRed="hello red";String msgYellow="hello yellow";rabbitTemplate.convertAndSend(exchangeName,"blue",msgblue);
}

@Test
public void contextDirectExchange()
{String exchangeName="direct";String msgblue="hello blue";String msgRed="hello red";String msgYellow="hello yellow";rabbitTemplate.convertAndSend(exchangeName,"red",msgRed);
}

話題-TopicExchange

TopicExchange與DirectExchange類似，區(qū)別在于rountingKey必須是多個單詞的列表，并且以"."分割。Queue與Exchange指定BindingKey可以使用通配符

#：代表0個或多個單詞

*：代表一個單詞

我們使用Direct的時候一個隊(duì)列如果綁定了很多key，會非常麻煩，通配符的引入就把key的綁定簡化許多，原來綁定多個key現(xiàn)在只需要綁定一個key。

實(shí)現(xiàn)思路：

1.利用@RabbitListenter聲明Exchange、Queue、RoutingKey

2.在consumer服務(wù)中，編寫兩個消費(fèi)者方法，分別監(jiān)聽topic.queue1和topic.queue2

//topic話題
@RabbitListener(bindings = @QueueBinding(value = @Queue(name = "topic.queue1"),exchange = @Exchange(name = "topic",type = ExchangeTypes.TOPIC),key ="china.#"
))
public void listennertopicqueue1(String msg)
{System.out.println("消費(fèi)者接收到topic.queue1的消息"+msg);
}
@RabbitListener(bindings = @QueueBinding(value = @Queue(name = "topic.queue2"),exchange = @Exchange(name = "topic",type = ExchangeTypes.TOPIC),key ="#.news"
))
public void listennertopicqueue2(String msg)
{System.out.println("消費(fèi)者接收到topic.queue2的消息"+msg);
}

3.在publisher中編寫測試方法，向交換機(jī)topic發(fā)送消息

@Test
public void contextTopicExchange()
{String exchangeName="topic";String msg="我是懶大王";rabbitTemplate.convertAndSend(exchangeName,"china.news",msg);
}

6.消息轉(zhuǎn)換器

在SpringAMQP的發(fā)送方法中，接收消息的類型是Object，也就是說我們可以發(fā)送任意對象類型的消息，SpringAMQP會幫我們序列化為字節(jié)后發(fā)送。

Spring的消息對象處理是由MessageConcerter來處理的，而默認(rèn)實(shí)現(xiàn)是SimpleMessageConverter，基于JDK的ObjectOutputStream完成序列化，這種序列化方式，比較浪費(fèi)內(nèi)存資源，如果需要修改，只需要定義一個MessageConverter類型的Bean即可。推薦用JSON方式序列化，步驟如下：

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

我們在publisher服務(wù)中聲明MessageConverter