rabbitmq知識梳理

一.WorkQueues模型

Work queues，任務模型。簡單來說就是讓多個消費者綁定到一個隊列，共同消費隊列中的消息。

在這里插入圖片描述
當消息處理比較耗時的時候，可能生產消息的速度會遠遠大于消息的消費速度。長此以往，消息就會堆積越來越多，無法及時處理。
此時就可以使用work 模型，多個消費者共同處理消息處理，消息處理的速度就能大大提高了。
接下來，我們就來模擬這樣的場景。
首先，我們在控制臺創建一個新的隊列，命名為work.queue：
在這里插入圖片描述

1.消息發送

這次我們循環發送，模擬大量消息堆積現象。
在publisher服務中的SpringAmqpTest類中添加一個測試方法：

 /*** workQueue* 向隊列中不停發送消息，模擬消息堆積。*/@Testpublic void testWorkQueue() throws InterruptedException {// 隊列名稱String queueName = "work.queue";// 消息String message = "hello, message_";for (int i = 0; i < 50; i++) {// 發送消息，每20毫秒發送一次，相當于每秒發送50條消息rabbitTemplate.convertAndSend(queueName, message + i);Thread.sleep(20);}}

2.消息接收

要模擬多個消費者綁定同一個隊列，我們在consumer服務的SpringRabbitListener中添加2個新的方法：

@RabbitListener(queues = "work.queue")
public void listenWorkQueue1(String msg) throws InterruptedException {System.out.println("消費者1接收到消息：【" + msg + "】" + LocalTime.now());Thread.sleep(20);
}@RabbitListener(queues = "work.queue")
public void listenWorkQueue2(String msg) throws InterruptedException {System.err.println("消費者2........接收到消息：【" + msg + "】" + LocalTime.now());Thread.sleep(200);
}

注意到這兩消費者，都設置了`Thead.sleep`，模擬任務耗時：

消費者1 sleep了20毫秒，相當于每秒鐘處理50個消息
消費者2 sleep了200毫秒，相當于每秒處理5個消息

在這里插入圖片描述
可以看到消費者1和消費者2竟然每人消費了25條消息：

消費者1很快完成了自己的25條消息
消費者2卻在緩慢的處理自己的25條消息。

也就是說消息是平均分配給每個消費者，并沒有考慮到消費者的處理能力。導致1個消費者空閑，另一個消費者忙的不可開交。沒有充分利用每一個消費者的能力，最終消息處理的耗時遠遠超過了1秒。這樣顯然是有問題的。

3.能者多勞

在spring中有一個簡單的配置，可以解決這個問題。我們修改consumer服務的application.yml文件，添加配置：

spring:rabbitmq:listener:simple:prefetch: 1 # 每次只能獲取一條消息，處理完成才能獲取下一個消息

在這里插入圖片描述
可以發現，由于消費者1處理速度較快，所以處理了更多的消息；消費者2處理速度較慢，只處理了6條消息。而最終總的執行耗時也在1秒左右，大大提升。正所謂能者多勞，這樣充分利用了每一個消費者的處理能力，可以有效避免消息積壓問題。

4.總結

Work模型的使用：

多個消費者綁定到一個隊列，同一條消息只會被一個消費者處理
通過設置prefetch來控制消費者預取的消息數量

二.交換機類型

在之前的兩個測試案例中，都沒有交換機，生產者直接發送消息到隊列。而一旦引入交換機，消息發送的模式會有很大變化：
在這里插入圖片描述

可以看到，在訂閱模型中，多了一個exchange角色，而且過程略有變化：

Publisher：生產者，不再發送消息到隊列中，而是發給交換機
Exchange：交換機，一方面，接收生產者發送的消息。另一方面，知道如何處理消息，例如遞交給某個特別隊列、遞交給所有隊列、或是將消息丟棄。到底如何操作，取決于Exchange的類型。
Queue：消息隊列也與以前一樣，接收消息、緩存消息。不過隊列一定要與交換機綁定。
Consumer：消費者，與以前一樣，訂閱隊列，沒有變化

Exchange（交換機）只負責轉發消息，不具備存儲消息的能力，因此如果沒有任何隊列與Exchange綁定，或者沒有符合路由規則的隊列，那么消息會丟失！

交換機的類型有四種：

Fanout：廣播，將消息交給所有綁定到交換機的隊列。我們最早在控制臺使用的正是Fanout交換機
Direct：訂閱，基于RoutingKey（路由key）發送給訂閱了消息的隊列
Topic：通配符訂閱，與Direct類似，只不過RoutingKey可以使用通配符
Headers：頭匹配，基于MQ的消息頭匹配，用的較少。

本次記錄前面的三種交換機模式。

1.Fanout交換機

說明

Fanout，英文翻譯是扇出，我覺得在MQ中叫廣播更合適。
在廣播模式下，消息發送流程是這樣的：
在這里插入圖片描述

1）可以有多個隊列
2）每個隊列都要綁定到Exchange（交換機）
3）生產者發送的消息，只能發送到交換機
4）交換機把消息發送給綁定過的所有隊列
5）訂閱隊列的消費者都能拿到消息
我們的計劃是這樣的：
創建一個名為 hmall.fanout的交換機，類型是Fanout
創建兩個隊列fanout.queue1和fanout.queue2，綁定到交換機hmall.fanout

1.在控制臺增加兩個新的隊列

在這里插入圖片描述
然后再創建一個交換機：

然后綁定兩個隊列到交換機：

測試

1.消息發送

在publisher服務的SpringAmqpTest類中添加測試方法：

@Test
public void testFanoutExchange() {// 交換機名稱String exchangeName = "hmall.fanout";// 消息String message = "hello, everyone!";rabbitTemplate.convertAndSend(exchangeName, "", message);
}

2.消息接收

在consumer服務的SpringRabbitListener中添加兩個方法，作為消費者：

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

在這里插入圖片描述

3.總結

交換機的作用是什么？

接收publisher發送的消息
將消息按照規則路由到與之綁定的隊列
不能緩存消息，路由失敗，消息丟失
FanoutExchange的會將消息路由到每個綁定的隊列

2.Direct交換機

說明

在Fanout模式中，一條消息，會被所有訂閱的隊列都消費。但是，在某些場景下，我們希望不同的消息被不同的隊列消費。這時就要用到Direct類型的Exchange。
在這里插入圖片描述
在Direct模型下：

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

案例需求如圖：
在這里插入圖片描述

聲明一個名為hmall.direct的交換機
聲明隊列direct.queue1，綁定hmall.direct，bindingKey為blud和red
聲明隊列direct.queue2，綁定hmall.direct，bindingKey為yellow和red
在consumer服務中，編寫兩個消費者方法，分別監聽direct.queue1和direct.queue2
在publisher中編寫測試方法，向hmall.direct發送消息

聲明隊列和交換機

首先在控制臺聲明兩個隊列direct.queue1和direct.queue2
在這里插入圖片描述
然后聲明一個direct類型的交換機，命名為hmall.direct:

然后使用red和blue作為key，綁定direct.queue1到hmall.direct：
同理，使用red和yellow作為key，綁定direct.queue2到hmall.direct，步驟略，最終結果：

測試

1.消息發送

在publisher服務的SpringAmqpTest類中添加測試方法：

@Test
public void testSendDirectExchange() {// 交換機名稱String exchangeName = "hmall.direct";// 消息String message = "紅色警報！日本亂排核廢水，導致海洋生物變異，驚現哥斯拉！";// 發送消息rabbitTemplate.convertAndSend(exchangeName, "red", message);
}

2.消息接收

在consumer服務的SpringRabbitListener中添加方法：

@RabbitListener(queues = "direct.queue1")
public void listenDirectQueue1(String msg) {System.out.println("消費者1接收到direct.queue1的消息：【" + msg + "】");
}@RabbitListener(queues = "direct.queue2")
public void listenDirectQueue2(String msg) {System.out.println("消費者2接收到direct.queue2的消息：【" + msg + "】");
}

由于使用的red這個key，所以兩個消費者都收到了消息：
在這里插入圖片描述
我們再切換為blue這個key：

3.總結

描述下Direct交換機與Fanout交換機的差異？

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

3.Topic交換機

說明

Topic類型的Exchange與Direct相比，都是可以根據RoutingKey把消息路由到不同的隊列。
只不過Topic類型Exchange可以讓隊列在綁定RoutingKey 的時候使用通配符！

RoutingKey 一般都是有一個或多個單詞組成，多個單詞之間以.分割，例如： item.insert

通配符規則：

#：匹配一個或多個詞
*：匹配不多不少恰好1個詞

舉例：

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

在這里插入圖片描述

測試

假如此時publisher發送的消息使用的RoutingKey共有四種：

china.news 代表有中國的新聞消息；
china.weather 代表中國的天氣消息；
japan.news 則代表日本新聞
japan.weather 代表日本的天氣消息；

解釋：

topic.queue1：綁定的是china.# ，凡是以 china.開頭的routing key 都會被匹配到，包括：
- china.news
- china.weather
topic.queue2：綁定的是#.news ，凡是以 .news結尾的 routing key 都會被匹配。包括:
- china.news
- japan.news

接下來，我們就按照上圖所示，來演示一下Topic交換機的用法。
首先，在控制臺按照圖示例子創建隊列、交換機，并利用通配符綁定隊列和交換機。此處步驟略。最終結果如下：
在這里插入圖片描述

1.消息發送

在publisher服務的SpringAmqpTest類中添加測試方法：

/*** topicExchange*/
@Test
public void testSendTopicExchange() {// 交換機名稱String exchangeName = "hmall.topic";// 消息String message = "喜報！孫悟空大戰哥斯拉，勝!";// 發送消息rabbitTemplate.convertAndSend(exchangeName, "china.news", message);
}

2.消息接收

@RabbitListener(queues = "topic.queue1")
public void listenTopicQueue1(String msg){System.out.println("消費者1接收到topic.queue1的消息：【" + msg + "】");
}@RabbitListener(queues = "topic.queue2")
public void listenTopicQueue2(String msg){System.out.println("消費者2接收到topic.queue2的消息：【" + msg + "】");
}

在這里插入圖片描述

3.總結

描述下Direct交換機與Topic交換機的差異？

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

4.聲明隊列和交換機

在之前我們都是基于RabbitMQ控制臺來創建隊列、交換機。但是在實際開發時，隊列和交換機是程序員定義的，將來項目上線，又要交給運維去創建。那么程序員就需要把程序中運行的所有隊列和交換機都寫下來，交給運維。在這個過程中是很容易出現錯誤的。
因此推薦的做法是由程序啟動時檢查隊列和交換機是否存在，如果不存在自動創建。

1.基本API

SpringAMQP提供了一個Queue類，用來創建隊列
在這里插入圖片描述
SpringAMQP還提供了一個Exchange接口，來表示所有不同類型的交換機：

我們可以自己創建隊列和交換機，不過SpringAMQP還提供了ExchangeBuilder來簡化這個過程：

在這里插入圖片描述
而在綁定隊列和交換機時，則需要使用BindingBuilder來創建Binding對象：

1.fanout示例

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 {/*** 聲明交換機* @return Fanout類型交換機*/@Beanpublic FanoutExchange fanoutExchange(){return new FanoutExchange("hmall.fanout");}/*** 第1個隊列*/@Beanpublic Queue fanoutQueue1(){return new Queue("fanout.queue1");}/*** 綁定隊列和交換機*/@Beanpublic Binding bindingQueue1(Queue fanoutQueue1, FanoutExchange fanoutExchange){return BindingBuilder.bind(fanoutQueue1).to(fanoutExchange);}/*** 第2個隊列*/@Beanpublic Queue fanoutQueue2(){return new Queue("fanout.queue2");}/*** 綁定隊列和交換機*/@Beanpublic Binding bindingQueue2(Queue fanoutQueue2, FanoutExchange fanoutExchange){return BindingBuilder.bind(fanoutQueue2).to(fanoutExchange);}
}

2.direct示例

direct模式由于要綁定多個KEY，會非常麻煩，每一個Key都要編寫一個binding：

import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;@Configuration
public class DirectConfig {/*** 聲明交換機* @return Direct類型交換機*/@Beanpublic DirectExchange directExchange(){return ExchangeBuilder.directExchange("hmall.direct").build();}/*** 第1個隊列*/@Beanpublic Queue directQueue1(){return new Queue("direct.queue1");}/*** 綁定隊列和交換機*/@Beanpublic Binding bindingQueue1WithRed(Queue directQueue1, DirectExchange directExchange){return BindingBuilder.bind(directQueue1).to(directExchange).with("red");}/*** 綁定隊列和交換機*/@Beanpublic Binding bindingQueue1WithBlue(Queue directQueue1, DirectExchange directExchange){return BindingBuilder.bind(directQueue1).to(directExchange).with("blue");}/*** 第2個隊列*/@Beanpublic Queue directQueue2(){return new Queue("direct.queue2");}/*** 綁定隊列和交換機*/@Beanpublic Binding bindingQueue2WithRed(Queue directQueue2, DirectExchange directExchange){return BindingBuilder.bind(directQueue2).to(directExchange).with("red");}/*** 綁定隊列和交換機*/@Beanpublic Binding bindingQueue2WithYellow(Queue directQueue2, DirectExchange directExchange){return BindingBuilder.bind(directQueue2).to(directExchange).with("yellow");}
}

3.基于注解聲明

基于@Bean的方式聲明隊列和交換機比較麻煩，Spring還提供了基于注解方式來聲明。

例如，我們同樣聲明Direct模式的交換機和隊列：

@RabbitListener(bindings = @QueueBinding(value = @Queue(name = "direct.queue1"),exchange = @Exchange(name = "hmall.direct", type = ExchangeTypes.DIRECT),key = {"red", "blue"}
))
public void listenDirectQueue1(String msg){System.out.println("消費者1接收到direct.queue1的消息：【" + msg + "】");
}@RabbitListener(bindings = @QueueBinding(value = @Queue(name = "direct.queue2"),exchange = @Exchange(name = "hmall.direct", type = ExchangeTypes.DIRECT),key = {"red", "yellow"}
))
public void listenDirectQueue2(String msg){System.out.println("消費者2接收到direct.queue2的消息：【" + msg + "】");
}

再試試Topic模式：

@RabbitListener(bindings = @QueueBinding(value = @Queue(name = "topic.queue1"),exchange = @Exchange(name = "hmall.topic", type = ExchangeTypes.TOPIC),key = "china.#"
))
public void listenTopicQueue1(String msg){System.out.println("消費者1接收到topic.queue1的消息：【" + msg + "】");
}@RabbitListener(bindings = @QueueBinding(value = @Queue(name = "topic.queue2"),exchange = @Exchange(name = "hmall.topic", type = ExchangeTypes.TOPIC),key = "#.news"
))
public void listenTopicQueue2(String msg){System.out.println("消費者2接收到topic.queue2的消息：【" + msg + "】");
}

4.消息轉換器

Spring的消息發送代碼接收的消息體是一個Object：
在這里插入圖片描述
而在數據傳輸時，它會把你發送的消息序列化為字節發送給MQ，接收消息的時候，還會把字節反序列化為Java對象。只不過，默認情況下Spring采用的序列化方式是JDK序列化。眾所周知，JDK序列化存在下列問題：

數據體積過大
有安全漏洞
可讀性差
我們來測試一下。
1）創建測試隊列
首先，我們在consumer服務中聲明一個新的配置類：

import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;@Configuration
public class MessageConfig {@Beanpublic Queue objectQueue() {return new Queue("object.queue");}
}

注意，這里我們先不要給這個隊列添加消費者，我們要查看消息體的格式。

重啟consumer服務以后，該隊列就會被自動創建出來了：
在這里插入圖片描述
2）發送消息
我們在publisher模塊的SpringAmqpTest中新增一個消息發送的代碼，發送一個Map對象：

@Test
public void testSendMap() throws InterruptedException {// 準備消息Map<String,Object> msg = new HashMap<>();msg.put("name", "柳巖");msg.put("age", 21);// 發送消息rabbitTemplate.convertAndSend("object.queue", msg);
}

發送消息后查看控制臺：
在這里插入圖片描述
可以看到消息格式非常不友好。

1.配置JSON轉換器

顯然，JDK序列化方式并不合適。我們希望消息體的體積更小、可讀性更高，因此可以使用JSON方式來做序列化和反序列化。
在publisher和consumer兩個服務中都引入依賴：

<dependency><groupId>com.fasterxml.jackson.dataformat</groupId><artifactId>jackson-dataformat-xml</artifactId><version>2.9.10</version>
</dependency>

注意，如果項目中引入了spring-boot-starter-web依賴，則無需再次引入Jackson依賴。

配置消息轉換器，在publisher和consumer兩個服務的啟動類中添加一個Bean即可：

@Bean
public MessageConverter messageConverter(){// 1.定義消息轉換器Jackson2JsonMessageConverter jackson2JsonMessageConverter = new Jackson2JsonMessageConverter();// 2.配置自動創建消息id，用于識別不同消息，也可以在業務中基于ID判斷是否是重復消息jackson2JsonMessageConverter.setCreate	MessageIds(true);return jackson2JsonMessageConverter;
}

消息轉換器中添加的messageId可以便于我們將來做冪等性判斷。
此時，我們到MQ控制臺刪除object.queue中的舊的消息。然后再次執行剛才的消息發送的代碼，到MQ的控制臺查看消息結構：
在這里插入圖片描述

2.消費者接收Object

@RabbitListener(queues = "object.queue")
public void listenSimpleQueueMessage(Map<String, Object> msg) throws InterruptedException {System.out.println("消費者接收到object.queue消息：【" + msg + "】");
}