柚子快報(bào)邀請(qǐng)碼778899分享：微服務(wù)學(xué)習(xí)-RabbitMQ

http://yzkb.51969.com/

RabbitMQ

1.初識(shí)MQ

1.1.同步和異步通訊

微服務(wù)間通訊有同步和異步兩種方式： 同步通訊：就像打電話，需要實(shí)時(shí)響應(yīng)。

異步通訊就像發(fā)郵件，不需要馬上回復(fù)。 兩種方式各有優(yōu)劣，打電話可以立即得到響應(yīng)，但是你卻不能跟多個(gè)人同時(shí)通話。發(fā)送郵件可以同時(shí)與多個(gè)人收發(fā)郵件，但是往往響應(yīng)會(huì)有延遲。

1.1.1.同步通訊

我們之前學(xué)習(xí)的Feign調(diào)用就屬于同步方式，雖然調(diào)用可以實(shí)時(shí)得到結(jié)果，但存在下面的問(wèn)題：

總結(jié)：

同步調(diào)用的優(yōu)點(diǎn)：

時(shí)效性較強(qiáng)，可以立即得到結(jié)果

同步調(diào)用的問(wèn)題：

耦合度高性能和吞吐能力下降有額外的資源消耗有級(jí)聯(lián)失敗問(wèn)題

1.1.2.異步通訊

異步調(diào)用則可以避免上述問(wèn)題：

我們以購(gòu)買(mǎi)商品為例，用戶支付后需要調(diào)用訂單服務(wù)完成訂單狀態(tài)修改，調(diào)用物流服務(wù)，從倉(cāng)庫(kù)分配響應(yīng)的庫(kù)存并準(zhǔn)備發(fā)貨。

在事件模式中，支付服務(wù)是事件發(fā)布者（publisher），在支付完成后只需要發(fā)布一個(gè)支付成功的事件（event），事件中帶上訂單id。

訂單服務(wù)和物流服務(wù)是事件訂閱者（Consumer），訂閱支付成功的事件，監(jiān)聽(tīng)到事件后完成自己業(yè)務(wù)即可。

為了解除事件發(fā)布者與訂閱者之間的耦合，兩者并不是直接通信，而是有一個(gè)中間人（Broker）。發(fā)布者發(fā)布事件到Broker，不關(guān)心誰(shuí)來(lái)訂閱事件。訂閱者從Broker訂閱事件，不關(guān)心誰(shuí)發(fā)來(lái)的消息。

Broker 是一個(gè)像數(shù)據(jù)總線一樣的東西，所有的服務(wù)要接收數(shù)據(jù)和發(fā)送數(shù)據(jù)都發(fā)到這個(gè)總線上，這個(gè)總線就像協(xié)議一樣，讓服務(wù)間的通訊變得標(biāo)準(zhǔn)和可控。

好處：

吞吐量提升：無(wú)需等待訂閱者處理完成，響應(yīng)更快速 故障隔離：服務(wù)沒(méi)有直接調(diào)用，不存在級(jí)聯(lián)失敗問(wèn)題 調(diào)用間沒(méi)有阻塞，不會(huì)造成無(wú)效的資源占用 耦合度極低，每個(gè)服務(wù)都可以靈活插拔，可替換 流量削峰：不管發(fā)布事件的流量波動(dòng)多大，都由Broker接收，訂閱者可以按照自己的速度去處理事件

缺點(diǎn)：

架構(gòu)復(fù)雜了，業(yè)務(wù)沒(méi)有明顯的流程線，不好管理需要依賴于Broker的可靠、安全、性能

好在現(xiàn)在開(kāi)源軟件或云平臺(tái)上 Broker 的軟件是非常成熟的，比較常見(jiàn)的一種就是我們今天要學(xué)習(xí)的MQ技術(shù)。

1.2.技術(shù)對(duì)比：

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

比較常見(jiàn)的MQ實(shí)現(xiàn)：

ActiveMQRabbitMQRocketMQKafka

幾種常見(jiàn)MQ的對(duì)比：

RabbitMQActiveMQRocketMQKafka公司/社區(qū)RabbitApache阿里Apache開(kāi)發(fā)語(yǔ)言ErlangJavaJavaScala&Java協(xié)議支持AMQP，XMPP，SMTP，STOMPOpenWire,STOMP，REST,XMPP,AMQP自定義協(xié)議自定義協(xié)議可用性高一般高高單機(jī)吞吐量一般差高非常高消息延遲微秒級(jí)毫秒級(jí)毫秒級(jí)毫秒以內(nèi)消息可靠性高一般高一般

追求可用性：Kafka、 RocketMQ 、RabbitMQ

追求可靠性：RabbitMQ、RocketMQ

追求吞吐能力：RocketMQ、Kafka

追求消息低延遲：RabbitMQ、Kafka

2.RabbitMQ安裝

3.SpringAMQP

SpringAMQP是基于RabbitMQ封裝的一套模板，并且還利用SpringBoot對(duì)其實(shí)現(xiàn)了自動(dòng)裝配，使用起來(lái)非常方便。

SpringAmqp的官方地址：https://spring.io/projects/spring-amqp

SpringAMQP提供了三個(gè)功能：

自動(dòng)聲明隊(duì)列、交換機(jī)及其綁定關(guān)系基于注解的監(jiān)聽(tīng)器模式，異步接收消息封裝了RabbitTemplate工具，用于發(fā)送消息

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

在父工程mq-demo中引入依賴

org.springframework.boot

spring-boot-starter-amqp

3.1.1.消息發(fā)送

首先配置MQ地址，在publisher服務(wù)的application.yml中添加配置：

spring:

rabbitmq:

host: 192.168.150.101 # 主機(jī)名

port: 5672 # 端口

virtual-host: / # 虛擬主機(jī)

username: admin # 用戶名

password: 123321 # 密碼

用戶名密碼在安裝時(shí)指定

然后在publisher服務(wù)中編寫(xiě)測(cè)試類SpringAmqpTest，并利用RabbitTemplate實(shí)現(xiàn)消息發(fā)送：

package cn.itcast.mq.spring;

import org.junit.Test;

import org.junit.runner.RunWith;

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;

@RunWith(SpringRunner.class)

@SpringBootTest

public class SpringAmqpTest {

@Autowired

private RabbitTemplate rabbitTemplate;

@Test

public void testSimpleQueue() {

// 隊(duì)列名稱

String queueName = "simple.queue";

// 消息

String message = "hello, spring amqp!";

// 發(fā)送消息

rabbitTemplate.convertAndSend(queueName, message);

}

}

3.1.2.消息接收

首先配置MQ地址，在consumer服務(wù)的application.yml中添加配置：

spring:

rabbitmq:

host: 192.168.150.101 # 主機(jī)名

port: 5672 # 端口

virtual-host: / # 虛擬主機(jī)

username: itcast # 用戶名

password: 123321 # 密碼

然后在consumer服務(wù)的cn.itcast.mq.listener包中新建一個(gè)類SpringRabbitListener，代碼如下：

package cn.itcast.mq.listener;

import org.springframework.amqp.rabbit.annotation.RabbitListener;

import org.springframework.stereotype.Component;

@Component

public class SpringRabbitListener {

@RabbitListener(queues = "simple.queue")

public void listenSimpleQueueMessage(String msg) throws InterruptedException {

System.out.println("spring 消費(fèi)者接收到消息：【" + msg + "】");

}

}

3.1.3.測(cè)試

啟動(dòng)consumer服務(wù)，然后在publisher服務(wù)中運(yùn)行測(cè)試代碼，發(fā)送MQ消息

3.2.WorkQueue

Work queues，也被稱為（Task queues），任務(wù)模型。簡(jiǎn)單來(lái)說(shuō)就是讓多個(gè)消費(fèi)者綁定到一個(gè)隊(duì)列，共同消費(fèi)隊(duì)列中的消息。

當(dāng)消息處理比較耗時(shí)的時(shí)候，可能生產(chǎn)消息的速度會(huì)遠(yuǎn)遠(yuǎn)大于消息的消費(fèi)速度。長(zhǎng)此以往，消息就會(huì)堆積越來(lái)越多，無(wú)法及時(shí)處理。

此時(shí)就可以使用work 模型，多個(gè)消費(fèi)者共同處理消息處理，速度就能大大提高了。

3.2.1.消息發(fā)送

這次我們循環(huán)發(fā)送，模擬大量消息堆積現(xiàn)象。

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

/**

* workQueue

* 向隊(duì)列中不停發(fā)送消息，模擬消息堆積。

*/

@Test

public void testWorkQueue() throws InterruptedException {

// 隊(duì)列名稱

String queueName = "simple.queue";

// 消息

String message = "hello, message_";

for (int i = 0; i < 50; i++) {

// 發(fā)送消息

rabbitTemplate.convertAndSend(queueName, message + i);

Thread.sleep(20);

}

}

3.2.2.消息接收

要模擬多個(gè)消費(fèi)者綁定同一個(gè)隊(duì)列，我們?cè)赾onsumer服務(wù)的SpringRabbitListener中添加2個(gè)新的方法：

@RabbitListener(queues = "simple.queue")

public void listenWorkQueue1(String msg) throws InterruptedException {

System.out.println("消費(fèi)者1接收到消息：【" + msg + "】" + LocalTime.now());

Thread.sleep(20);

}

@RabbitListener(queues = "simple.queue")

public void listenWorkQueue2(String msg) throws InterruptedException {

System.err.println("消費(fèi)者2........接收到消息：【" + msg + "】" + LocalTime.now());

Thread.sleep(200);

}

注意到這個(gè)消費(fèi)者sleep了1000秒，模擬任務(wù)耗時(shí)。

3.2.3.測(cè)試

啟動(dòng)ConsumerApplication后，在執(zhí)行publisher服務(wù)中剛剛編寫(xiě)的發(fā)送測(cè)試方法testWorkQueue。

可以看到消費(fèi)者1很快完成了自己的25條消息。消費(fèi)者2卻在緩慢的處理自己的25條消息。

也就是說(shuō)消息是平均分配給每個(gè)消費(fèi)者，并沒(méi)有考慮到消費(fèi)者的處理能力。這樣顯然是有問(wèn)題的。

3.2.4.能者多勞

在spring中有一個(gè)簡(jiǎn)單的配置，可以解決這個(gè)問(wèn)題。我們修改consumer服務(wù)的application.yml文件，添加配置：

spring:

rabbitmq:

listener:

simple:

prefetch: 1 # 每次只能獲取一條消息，處理完成才能獲取下一個(gè)消息

3.2.5.總結(jié)

Work模型的使用：

多個(gè)消費(fèi)者綁定到一個(gè)隊(duì)列，同一條消息只會(huì)被一個(gè)消費(fèi)者處理通過(guò)設(shè)置prefetch來(lái)控制消費(fèi)者預(yù)取的消息數(shù)量

3.3.發(fā)布/訂閱

在訂閱模型中，多了一個(gè)exchange角色，而且過(guò)程略有變化：

Publisher：生產(chǎn)者，也就是要發(fā)送消息的程序，但是不再發(fā)送到隊(duì)列中，而是發(fā)給X（交換機(jī)）Exchange：交換機(jī)，圖中的X。一方面，接收生產(chǎn)者發(fā)送的消息。另一方面，知道如何處理消息，例如遞交給某個(gè)特別隊(duì)列、遞交給所有隊(duì)列、或是將消息丟棄。到底如何操作，取決于Exchange的類型。Exchange有以下3種類型：

Fanout：廣播，將消息交給所有綁定到交換機(jī)的隊(duì)列Direct：定向，把消息交給符合指定routing key 的隊(duì)列Topic：通配符，把消息交給符合routing pattern（路由模式） 的隊(duì)列 Consumer：消費(fèi)者，與以前一樣，訂閱隊(duì)列，沒(méi)有變化Queue：消息隊(duì)列也與以前一樣，接收消息、緩存消息。

Exchange（交換機(jī)）只負(fù)責(zé)轉(zhuǎn)發(fā)消息，不具備存儲(chǔ)消息的能力，因此如果沒(méi)有任何隊(duì)列與Exchange綁定，或者沒(méi)有符合路由規(guī)則的隊(duì)列，那么消息會(huì)丟失！

3.4.Fanout

Fanout，英文翻譯是扇出，我覺(jué)得在MQ中叫廣播更合適。

在廣播模式下，消息發(fā)送流程是這樣的：

1） 可以有多個(gè)隊(duì)列2） 每個(gè)隊(duì)列都要綁定到Exchange（交換機(jī)）3） 生產(chǎn)者發(fā)送的消息，只能發(fā)送到交換機(jī)，交換機(jī)來(lái)決定要發(fā)給哪個(gè)隊(duì)列，生產(chǎn)者無(wú)法決定4） 交換機(jī)把消息發(fā)送給綁定過(guò)的所有隊(duì)列5） 訂閱隊(duì)列的消費(fèi)者都能拿到消息

我們的計(jì)劃是這樣的：

創(chuàng)建一個(gè)交換機(jī) itcast.fanout，類型是Fanout創(chuàng)建兩個(gè)隊(duì)列fanout.queue1和fanout.queue2，綁定到交換機(jī)itcast.fanout

3.4.1.聲明隊(duì)列和交換機(jī)

Spring提供了一個(gè)接口Exchange，來(lái)表示所有不同類型的交換機(jī)：

在consumer中創(chuàng)建一個(gè)類，聲明隊(duì)列和交換機(jī)：

package cn.itcast.mq.config;

import org.springframework.amqp.core.Binding;

import org.springframework.amqp.core.BindingBuilder;

import org.springframework.amqp.core.FanoutExchange;

import org.springframework.amqp.core.Queue;

import org.springframework.context.annotation.Bean;

import org.springframework.context.annotation.Configuration;

@Configuration

public class FanoutConfig {

/**

* 聲明交換機(jī)

* @return Fanout類型交換機(jī)

*/

@Bean

public FanoutExchange fanoutExchange(){

return new FanoutExchange("itcast.fanout");

}

/**

* 第1個(gè)隊(duì)列

*/

@Bean

public Queue fanoutQueue1(){

return new Queue("fanout.queue1");

}

/**

* 綁定隊(duì)列和交換機(jī)

*/

@Bean

public Binding bindingQueue1(Queue fanoutQueue1, FanoutExchange fanoutExchange){

return BindingBuilder.bind(fanoutQueue1).to(fanoutExchange);

}

/**

* 第2個(gè)隊(duì)列

*/

@Bean

public Queue fanoutQueue2(){

return new Queue("fanout.queue2");

}

/**

* 綁定隊(duì)列和交換機(jī)

*/

@Bean

public Binding bindingQueue2(Queue fanoutQueue2, FanoutExchange fanoutExchange){

return BindingBuilder.bind(fanoutQueue2).to(fanoutExchange);

}

}

3.4.2.消息發(fā)送

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

@Test

public void testFanoutExchange() {

// 隊(duì)列名稱

String exchangeName = "itcast.fanout";

// 消息

String message = "hello, everyone!";

rabbitTemplate.convertAndSend(exchangeName, "", message);

}

3.4.3.消息接收

在consumer服務(wù)的SpringRabbitListener中添加兩個(gè)方法，作為消費(fèi)者：

@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 + "】");

}

3.4.4.總結(jié)

交換機(jī)的作用是什么？

接收publisher發(fā)送的消息將消息按照規(guī)則路由到與之綁定的隊(duì)列不能緩存消息，路由失敗，消息丟失FanoutExchange的會(huì)將消息路由到每個(gè)綁定的隊(duì)列

聲明隊(duì)列、交換機(jī)、綁定關(guān)系的Bean是什么？

QueueFanoutExchangeBinding

3.5.Direct

在Fanout模式中，一條消息，會(huì)被所有訂閱的隊(duì)列都消費(fèi)。但是，在某些場(chǎng)景下，我們希望不同的消息被不同的隊(duì)列消費(fèi)。這時(shí)就要用到Direct類型的Exchange。

在Direct模型下：

隊(duì)列與交換機(jī)的綁定，不能是任意綁定了，而是要指定一個(gè)RoutingKey（路由key）消息的發(fā)送方在 向 Exchange發(fā)送消息時(shí)，也必須指定消息的 RoutingKey。Exchange不再把消息交給每一個(gè)綁定的隊(duì)列，而是根據(jù)消息的Routing Key進(jìn)行判斷，只有隊(duì)列的Routingkey與消息的 Routing key完全一致，才會(huì)接收到消息

案例需求如下：

利用@RabbitListener聲明Exchange、Queue、RoutingKey 在consumer服務(wù)中，編寫(xiě)兩個(gè)消費(fèi)者方法，分別監(jiān)聽(tīng)direct.queue1和direct.queue2 在publisher中編寫(xiě)測(cè)試方法，向itcast. direct發(fā)送消息

3.5.1.基于注解聲明隊(duì)列和交換機(jī)

基于@Bean的方式聲明隊(duì)列和交換機(jī)比較麻煩，Spring還提供了基于注解方式來(lái)聲明。

在consumer的SpringRabbitListener中添加兩個(gè)消費(fèi)者，同時(shí)基于注解來(lái)聲明隊(duì)列和交換機(jī)：

@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)者接收到direct.queue1的消息：【" + msg + "】");

}

@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)者接收到direct.queue2的消息：【" + msg + "】");

}

3.5.2.消息發(fā)送

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

@Test

public void testSendDirectExchange() {

// 交換機(jī)名稱

String exchangeName = "itcast.direct";

// 消息

String message = "紅色警報(bào)！日本亂排核廢水，導(dǎo)致海洋生物變異，驚現(xiàn)哥斯拉！";

// 發(fā)送消息

rabbitTemplate.convertAndSend(exchangeName, "red", message);

}

3.5.3.總結(jié)

描述下Direct交換機(jī)與Fanout交換機(jī)的差異？

Fanout交換機(jī)將消息路由給每一個(gè)與之綁定的隊(duì)列Direct交換機(jī)根據(jù)RoutingKey判斷路由給哪個(gè)隊(duì)列如果多個(gè)隊(duì)列具有相同的RoutingKey，則與Fanout功能類似

基于@RabbitListener注解聲明隊(duì)列和交換機(jī)有哪些常見(jiàn)注解？

@Queue@Exchange

3.6.Topic

3.6.1.說(shuō)明

Topic類型的Exchange與Direct相比，都是可以根據(jù)RoutingKey把消息路由到不同的隊(duì)列。只不過(guò)Topic類型Exchange可以讓隊(duì)列在綁定Routing key 的時(shí)候使用通配符！

Routingkey 一般都是有一個(gè)或多個(gè)單詞組成，多個(gè)單詞之間以”.”分割，例如： item.insert

通配符規(guī)則：

#：匹配一個(gè)或多個(gè)詞

*：匹配不多不少恰好1個(gè)詞

舉例：

item.#：能夠匹配item.spu.insert 或者 item.spu

item.*：只能匹配item.spu

?

圖示：

解釋：

Queue1：綁定的是china.# ，因此凡是以 china.開(kāi)頭的routing key 都會(huì)被匹配到。包括china.news和china.weatherQueue2：綁定的是#.news ，因此凡是以 .news結(jié)尾的 routing key 都會(huì)被匹配。包括china.news和japan.news

案例需求：

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

并利用@RabbitListener聲明Exchange、Queue、RoutingKey 在consumer服務(wù)中，編寫(xiě)兩個(gè)消費(fèi)者方法，分別監(jiān)聽(tīng)topic.queue1和topic.queue2 在publisher中編寫(xiě)測(cè)試方法，向itcast. topic發(fā)送消息

3.6.2.消息發(fā)送

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

/**

* topicExchange

*/

@Test

public void testSendTopicExchange() {

// 交換機(jī)名稱

String exchangeName = "itcast.topic";

// 消息

String message = "喜報(bào)！孫悟空大戰(zhàn)哥斯拉，勝!";

// 發(fā)送消息

rabbitTemplate.convertAndSend(exchangeName, "china.news", message);

}

3.6.3.消息接收

在consumer服務(wù)的SpringRabbitListener中添加方法：

@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)者接收到topic.queue1的消息：【" + msg + "】");

}

@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)者接收到topic.queue2的消息：【" + msg + "】");

}

3.6.4.總結(jié)

描述下Direct交換機(jī)與Topic交換機(jī)的差異？

Topic交換機(jī)接收的消息RoutingKey必須是多個(gè)單詞，以 **.** 分割Topic交換機(jī)與隊(duì)列綁定時(shí)的bindingKey可以指定通配符#：代表0個(gè)或多個(gè)詞*：代表1個(gè)詞

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

之前說(shuō)過(guò)，Spring會(huì)把你發(fā)送的消息序列化為字節(jié)發(fā)送給MQ，接收消息的時(shí)候，還會(huì)把字節(jié)反序列化為Java對(duì)象。

只不過(guò)，默認(rèn)情況下Spring采用的序列化方式是JDK序列化。眾所周知，JDK序列化存在下列問(wèn)題：

數(shù)據(jù)體積過(guò)大有安全漏洞可讀性差

我們來(lái)測(cè)試一下。

3.7.1.測(cè)試默認(rèn)轉(zhuǎn)換器

我們修改消息發(fā)送的代碼，發(fā)送一個(gè)Map對(duì)象：

@Test

public void testSendMap() throws InterruptedException {

// 準(zhǔn)備消息

Map msg = new HashMap<>();

msg.put("name", "Jack");

msg.put("age", 21);

// 發(fā)送消息

rabbitTemplate.convertAndSend("simple.queue","", msg);

}

停止consumer服務(wù)

發(fā)送消息后查看控制臺(tái)：發(fā)現(xiàn)消息被序列化并且使用的是JDK默認(rèn)的序列化器，序列化后的消息長(zhǎng)度很長(zhǎng)，所以可以配置JSON轉(zhuǎn)換器。

3.7.2.配置JSON轉(zhuǎn)換器

顯然，JDK序列化方式并不合適。我們希望消息體的體積更小、可讀性更高，因此可以使用JSON方式來(lái)做序列化和反序列化。

在publisher和consumer兩個(gè)服務(wù)中都引入依賴：

com.fasterxml.jackson.dataformat

jackson-dataformat-xml

2.9.10

配置消息轉(zhuǎn)換器。

在啟動(dòng)類中添加一個(gè)Bean即可：

@Bean

public MessageConverter jsonMessageConverter(){

return new Jackson2JsonMessageConverter();

}

柚子快報(bào)邀請(qǐng)碼778899分享：微服務(wù)學(xué)習(xí)-RabbitMQ

http://yzkb.51969.com/

精彩內(nèi)容