目錄

• 1、簡單概述RabbitMQ重要作用
• 2、簡單概述RabbitMQ重要概念
• 3、Spring Boot整合RabbitMQ

前言
RabbitMQ是一個訊息佇列，主要是用來實現應用程式的非同步和解耦，同時也能起到訊息緩衝，訊息分發的作用。訊息中介軟體最主要的作用還是解耦，中介軟體最標準的用法是生產者生產訊息傳送到佇列，消費者從佇列中拿取訊息並處理，生產者不用關心是誰來消費，消費者不用關心誰在生產訊息，從而達到解耦的目的。在分散式的系統中，訊息佇列也會被用在很多其它的方面，比如：分散式事務的支援，RPC的呼叫等等。

@

1、簡單概述RabbitMQ重要作用

首先談談作用，你知道它有啥用才會有興趣去知道它的重要概念，走進它，親近它！上面已經提過了RabbitMQ主要是用來實現程式的非同步和解耦。這裡也主要講解它是如何做到非同步和解耦的。

1.1、非同步

對比一下使用訊息佇列實現非同步的好處：

1.2、解耦

至於解耦只能靠自己的對耦合的理解，這裡就以文字的形式概述：

以上面訊息佇列實現非同步場景分析：主執行緒依舊處理耗時低的入庫操作，然後把需要處理的訊息寫進訊息佇列中，這個寫入耗時可以忽略不計，非常快，然後，獨立的發郵件子系統，和獨立的發簡訊子系統，同時訂閱訊息佇列，進行單獨處理。處理好之後，向佇列傳送ACK確認，訊息佇列整條資料刪除。這個流程也是現在各大公司都在用的方式，以SOA服務化各個系統，把耗時操作，單獨交給獨立的業務系統，通過訊息佇列作為中介軟體，達到應用解耦的目的，並且消耗的資源很低，單臺伺服器能承受更大的併發請求。

到這裡，經過一段存文字的薰陶，估計各位已經一臉懵逼了，你們個個都是人才做到一臉懵逼的看完，阿姨都忍不住給你喊666....

2、簡單概述RabbitMQ重要概念

首先，RabbitMQ是訊息中介軟體的一種，類似的還有ActiveMQ、RocketMQ....總的來說這些訊息中介軟體都泛指的就是分散式系統中完成訊息的傳送和接收的基礎軟體。

接下來重點來了...

訊息中介軟體工作過程 == 生產者消費者模型

因此對於訊息佇列來說最重要的三個概念就是：生產者、訊息佇列、消費者

2.1、RabitMQ的工作流程

上面主要對訊息佇列做了一個共性分析，對於RabbitMQ訊息佇列來說,除了這三個重要概念以外,還有一個很重要的概念就是交換機(Exchange)。交換機使得生產者和訊息佇列之間產生了隔離,生產者將訊息傳送給交換機,而交換機則根據排程策略把相應的訊息轉發給對應的訊息佇列。
因此對於RabbitMQ來說最重要的四個概念就是：生產者、訊息佇列、消費者、交換機

總的來說，RabitMQ的工作流程如下所示:

而具體的交換機如下會講到。

2.2、交換機

對交換機的通俗易懂說法就是：

交換機好比快遞公司派發快遞，是百世快遞就給百世快遞小哥，是申通快遞就派發給申通快遞小哥，是圓通快遞就派發給圓通快遞小哥，而這些不同派發方式就是一種匹配規則，實際上交換機有四種類型,分別為Direct、topic、headers、Fanout，而這四種類型就好比四種不同的匹配規則，交換機就類似這種意思。

上面已經提到了交換機有四種類型Direct、topic、headers、Fanout，而這也是一個重點，下面簡單概述一下這四種類型：

Direct型別：【重點】
Direct是RabbitMQ預設的交換機模式,也是最簡單的模式.即建立訊息佇列的時候,指定一個BindingKey.當傳送者傳送訊息的時候,指定對應的Key.當Key和訊息佇列的BindingKey一致的時候,訊息將會被髮送到該訊息佇列中.

 

topic型別：【重點】

topic轉發資訊主要是依據萬用字元,佇列和交換機的繫結主要是依據一種模式(萬用字元+字串),而當傳送訊息的時候,只有指定的Key和該模式相匹配的時候,訊息才會被髮送到該訊息佇列中。比如*.news 或者#.news，其中比如#代表0到多個隨機字元。

 

Fanout型別：【重點】

Fanout是路由廣播的形式,將會把訊息發給繫結它的全部佇列,即便設定了key,也會被忽略.

 

headers型別：

headers也是根據一個規則進行匹配,在訊息佇列和交換機繫結的時候會指定一組鍵值對規則,而傳送訊息的時候也會指定一組鍵值對規則,當兩組鍵值對規則相匹配的時候,訊息會被髮送到匹配的訊息佇列中.

2.3、RabbitMQ關鍵概念總覽簡述

簡單來說，RabbitMQ關鍵概念如下：

1、生產者：傳送訊息的程式
2、消費者：監聽接收消費訊息的程式
3、訊息：一串二進位制資料流
4、佇列：訊息的暫存區/儲存區
5、交換機：訊息的中轉站，用於接收分發訊息。其中有 fanout、direct、topic、headers 四種類型
6、路由\鍵：相當於金鑰/第三者，與交換機繫結即可路由訊息到指定的佇列！

而有了上面的這些概念之後，再來了解了解訊息模型的演變歷程，當然，這一歷程在 RabbitMQ 官網也是可以得知的，RabbitMQ 官網：https://www.rabbitmq.com/getstarted.html

以下圖文轉截於https://blog.csdn.net/u013871100/article/details/82982235，於此同時，推薦各位去看看這篇文章，寫的真的不錯。


好了到這裡，RabbitMQ就概述的差不多了，接下來進行程式碼整合階段。

3、Spring Boot整合RabbitMQ

Spring Boot整合RabbitMQ原理分析

按照SpringBoot的常規套路，估計可能存在一個Rabbit的XXXAutoConfiguration，全域性檢索一下，毋庸置疑存在RabbitAutoConfiguration，那就來分析分析這個類。

1、首先映入眼簾的就是自動配置了連線工廠的ConnectionFactory
那麼，這個工程做了些什麼呢？進去看看其原始碼

public CachingConnectionFactory rabbitConnectionFactory(RabbitProperties properties,
                ObjectProvider<ConnectionNameStrategy> connectionNameStrategy) throws Exception {
            PropertyMapper map = PropertyMapper.get();
            CachingConnectionFactory factory = new CachingConnectionFactory(
                    getRabbitConnectionFactoryBean(properties).getObject());
            map.from(properties::determineAddresses).to(factory::setAddresses);
            map.from(properties::isPublisherReturns).to(factory::setPublisherReturns);
            map.from(properties::getPublisherConfirmType).whenNonNull().to(factory::setPublisherConfirmType);
            RabbitProperties.Cache.Channel channel = properties.getCache().getChannel();
            map.from(channel::getSize).whenNonNull().to(factory::setChannelCacheSize);
            map.from(channel::getCheckoutTimeout).whenNonNull().as(Duration::toMillis)
                    .to(factory::setChannelCheckoutTimeout);
            RabbitProperties.Cache.Connection connection = properties.getCache().getConnection();
            map.from(connection::getMode).whenNonNull().to(factory::setCacheMode);
            map.from(connection::getSize).whenNonNull().to(factory::setConnectionCacheSize);
            map.from(connectionNameStrategy::getIfUnique).whenNonNull().to(factory::setConnectionNameStrategy);
            return factory;
        }

裡面是一些連線配置的資訊操作，而這些配置資訊的來源正是其引數類RabbitProperties ，隨之看看RabbitProperties 這個配置類

既然是配置類，那裡面就是一些配置資訊，具體的可以通過配置檔案以spring.rabbitmq的方式進行配置。

2、RabbitTemplate類
再往下看就是RabbitTemplate類了，該類的具體的位置在於

看其RabbitTemplate關鍵原始碼：

    @Override
    public void convertAndSend(String exchange, String routingKey, final Object object) throws AmqpException {
        convertAndSend(exchange, routingKey, object, (CorrelationData) null);
    }
    ....
    @Override
    @Nullable
    public Object receiveAndConvert(String queueName) throws AmqpException {
        return receiveAndConvert(queueName, this.receiveTimeout);
    }

得知，RabbitTemplate主要是提供RabbitMQ傳送和接受訊息的功能；

3、AmqpAdmin
再接下來往下看就是AmqpAdmin介面了

@Bean
        @ConditionalOnSingleCandidate(ConnectionFactory.class)
        @ConditionalOnProperty(prefix = "spring.rabbitmq", name = "dynamic", matchIfMissing = true)
        @ConditionalOnMissingBean
        public AmqpAdmin amqpAdmin(ConnectionFactory connectionFactory) {
            return new RabbitAdmin(connectionFactory);
        }

點進AmqpAdmin 介面，觀察其具體關鍵程式碼：

boolean deleteQueue(String queueName);

    // Note that nowait option is not readily exposed in Rabbit Java API but is for Rabbit .NET API.

    /**
     * Delete a queue.
     * @param queueName the name of the queue.
     * @param unused true if the queue should be deleted only if not in use.
     * @param empty true if the queue should be deleted only if empty.
     */
    void deleteQueue(String queueName, boolean unused, boolean empty);

    /**
     * Purges the contents of the given queue.
     * @param queueName the name of the queue.
     * @param noWait true to not await completion of the purge.
     */
    void purgeQueue(String queueName, boolean noWait);

    /**
     * Purges the contents of the given queue.
     * @param queueName the name of the queue.
     * @return the number of messages purged.
     * @since 2.1
     */
    int purgeQueue(String queueName);

    // Binding operations

    /**
     * Declare a binding of a queue to an exchange.
     * @param binding a description of the binding to declare.
     */
    void declareBinding(Binding binding);

    /**
     * Remove a binding of a queue to an exchange. Note unbindQueue/removeBinding was not introduced until 0.9 of the
     * specification.
     * @param binding a description of the binding to remove.
     */
    void removeBinding(Binding binding);
    ...

發現AmqpAdmin是用來建立和刪除 Queue，Exchange，Binding等，起到管理元件的作用。

差不多，工廠裡的程式碼就分析到這裡。

3.1、整合前準備Rabbitmq環境

在進行整合RabbitMQ前，我們需要安裝好 RabbitMQ 及其後端控制檯應用，並在專案中配置一下 RabbitMQ 的相關引數以及相關 Bean 元件。我相信你們都安裝了，不然看個錘子整合文章啊....

RabbitMQ 安裝完成後，開啟後端控制檯應用：http://192.168.42.142:15672 輸入賬戶guest 密碼guest 之後登入，看到下圖即表示安裝成功

而且肯定要自己建立幾個Exchanges和Queues,並且自行Bing上關係

準備好之後就可以整合測試效果了。

3.2、搭建依賴環境

之後pom.xml的依賴如下：

  <dependency>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter-amqp</artifactId>
  </dependency>

3.3、配置 application.properties檔案

然後是專案配置檔案層面的配置 application.properties

spring.rabbitmq.host=192.168.42.142  
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest

這裡只是測試整合RabbitMQ，以上配置暫時足夠了。

3.4、編寫ApplicationTests測試程式碼

編寫程式碼之前，我們要先知道一點，springboot中提供了類似jdbcTemplate的模板，也就是RabbitTemplate ，作用理解起來就是一樣的，這裡先提一下，之後就可以使用其RabbitTemplate的強大功能方法。

package com.yichun.rabbitmq;

import org.junit.jupiter.api.Test;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;
import java.util.function.ObjIntConsumer;

@SpringBootTest
class RabbitmqApplicationTests {
   //依賴注入RabbitTemplate 
    @Autowired
    RabbitTemplate rabbitTemplate;

 //fanout：廣播方式傳送資料 ：與路由鍵key無關
    @Test
    void sendMsgs() {

        rabbitTemplate.convertAndSend("exchange.fanout","","你大爺還是你大爺");
    }

  //direct：單播方式傳送資料 ：與路由鍵key一一對應
    @Test
    void contextLoads() {
        Map<String, Object> map= new HashMap<>();
        map.put("1","hello，熊dei在嗎？");
        map.put("2", Arrays.asList("Tom",111));
        rabbitTemplate.convertAndSend("exchange.direct","yichun.news",map);
    }

  //Topic:按指定路由鍵Key規則方式傳送資料
    @Test
    void sendTopicMsgs() {
        rabbitTemplate.convertAndSend("exchange.topic","afdd.news","年輕人站起來嗷嗷嗷嗷...奧利給！");
    }

    //接收資料
    @Test
    void receive(){
        Object o = rabbitTemplate.receiveAndConvert("yichun.news");
        System.out.println(o.getClass());
        System.out.println(o);
    }
}

單個單個單元測試執行

1、fanout廣播方式傳送資料觀察

2、Topic指定路由鍵Key規則方式傳送資料

3、direct單播方式傳送資料

開啟資訊發現如下：

這是什麼情況呢？其實這種情況主要是因為Springboot在RabbitAutoConfiguration中預設使用的訊息轉換規則，要想看到想看到的資料格式，比如json格式，這個時候就要自定義轉換規則了。實際上面兩次測試也是出現這種情況。

3.5、自定義訊息轉換規則

首先要想自定義訊息轉換規則，我們就要改變預設的規則，首先開啟MessageConverter介面，如果是eclipse切換的快捷鍵，就可以直接F4，檢視它的實現繼承的關係類，如下圖

是的，分析發現確實有json的轉換規則，於是，開始編寫程式碼。首先編寫一個MyMQConfig類，具體實現如下：

@Configuration
public class MyMQConfig {

    @Bean
    public MessageConverter messageConverter(){
            return new Jackson2JsonMessageConverter(); //讓其返回一個json規則的轉換
    }
}

編寫完之後，再次執行之前的傳送訊息，效果如下：

3.6、AmqpAdmin 管理元件的使用

在之前我們整合前的那些交換機、佇列、繫結資訊都是自己手動新增的，實際上，Springboot中RabbitAutoConfiguration中的ConnectionFactory工廠存在的AmqpAdmin就可以建立和刪除 Queue，Exchange，Binding等，起到管理元件的作用。

==amqpAdmin中凡是declareXXX的方法都是用於建立元件，而以remove和delete開頭就是刪除元件，這是一個小技巧。接下來就以declareXXX的方法來建立各個元件進行分析，至於刪除就元件就直接amqpAdmin.元件名字就行了。==

1、建立Exchange

例項程式碼

    @Autowired
    AmqpAdmin amqpAdmin;

    @Test
    void clareExchange() {
        amqpAdmin.declareExchange(new DirectExchange("交換機名"));
    }

分析上面的程式碼，首先是關於交換機Exchange的，所以先全文檢索一下Exchange，是個介面，如下

其次既然是建立交換機Exchange，那肯定要寫入建立交換機的名字了，傳入的引數DirectExchange中，滑鼠點進去，發現

到這裡，分析的差不多了，可以看出上面建立Exchange的方法就是最簡單的建立一個叫“交換機名”的Exchange，執行測試一下，效果如下：

2、建立Queue

接下來同理建立Queue

    @Autowired
    AmqpAdmin amqpAdmin;

    @Test
    void clareExchange() {
        amqpAdmin.declareQueue(new Queue("測試佇列.queue",true)); //true代表是否持久化
    }

執行效果：

3、建立Binding

    @Autowired
    AmqpAdmin amqpAdmin;

    @Test
    void clareExchange() {
       amqpAdmin.declareBinding(new Binding("測試佇列.queue",Binding.DestinationType.QUEUE,"交換機名","路由鍵名",null));
    }

執行效果：

3.7、監聽訊息@EnableRabbit + @RabbitListener

我們實際開發中，常常有如下需求，下單通過訊息佇列之後庫存要隨之改變，也就是觸發監聽機制。

這種情況往往要通過：@EnableRabbit + @RabbitListener 監聽訊息佇列的內容
，特別注意@RabbitListener中的queues屬性是個陣列
,也就是說@RabbitListener可以同時監聽多個訊息
！

    String[] queues() default {};

當然這個監聽實現也很簡單，只需要在需要監聽的訊息業務程式碼上新增@RabbitListener註解，然後再Application主方法上新增@EnableRabbit 註解開啟基於註解的rabbit模式即可。例項程式碼如下：

@EnableRabbit  //開啟基於註解的rabbit模式
@SpringBootApplication
public class RabbitmqApplication {
    public static void main(String[] args) {
        SpringApplication.run(RabbitmqApplication.class, args);
    }
}

dao程式碼：

public class CatDao {
    private String name;
    private int age;
    
//    get、set、toString、構造方法....

}

service業務程式碼

@Service
public class CatService {

//    監聽方式一：直接將物件序列化輸出
    @RabbitListener(queues="yichun.news")
    public void receive(CatDao cat){  //這裡的CatDao就是一個普通的bean，這裡主要用於藉助觸發監聽到事件打印出其物件資訊
        System.out.println("監聽到的資訊："+ cat);
    }

//    監聽方式二：特殊需求需要訊息頭等資訊
    @RabbitListener(queues="yichun")
    public void  receiveMessage(Message message){ //千萬注意這裡的Message是 org.springframework.amqp.core.Message 別導錯包了
        System.out.println(message.getBody());
        System.out.println(message.getMessageProperties());
    }
}

單元Test測試程式碼：

 @Test
    void contextLoads() {
        rabbitTemplate.convertAndSend("exchange.direct","yichun.news",new CatDao("Tom貓",12)); //這裡的資料是object型別的
    }

監聽方式一：直接將物件序列化輸出列印結果：

監聽方式二特殊需求需要訊息頭等資訊列印結果：

如果本文對你有一點點幫助，那麼請點個讚唄，謝謝~

最後，若有不足或者不正之處，歡迎指正批評，感激不盡！如果有疑問歡迎留言，絕對第一時間回覆！

歡迎各位關注我的公眾號，裡面有一些java學習資料和一大波java電子書籍，比如說周志明老師的深入java虛擬機器、java程式設計思想、核心技術卷、大話設計模式、java併發程式設計實戰.....都是java的聖經，不說了快上Tomcat車，咋們走！最主要的是一起探討技術，嚮往技術，追求技術，說好了來了就是盆友喔...

相關推薦