35 Spring Boot 與消息 （JMS、AMQP、RabbitMQ）

RabbitMQ教程 - 鳥哥的專欄 - CSDN博客

1、概述

1. 大多應用中，可經過消息服務中間件來提高系統異步通訊、擴展解耦能力
2. 消息服務中兩個重要概念：
   1. 消息代理（message broker)和目的地（destination)

   2. 當消息發送者發送消息之後，將由消息代理接管，消息代理保證消息傳遞到指定目
      的地。

      　
3. 消息隊列主要有兩種形式的目的地
   1. 隊列（queue):點對點消息通訊（point-to-point)
   2. 主題（topic):發佈（publish) /訂閱（subscribe)消息通訊

 

2、應用場景圖解：

 

 

 

4. 點對點式：

• 消息發送者發送消息，消息代理將其放入一個隊列中，消息接收者從隊列中獲取消息內容， 消息讀取後被移出隊列
• 消息只有惟一的_和接受者，但並非說只能有一個接收者

5. 發佈訂閱式：

• 發送者（發佈者）發送消息到主題，多個接收者（訂閱者）監聽（訂閱）這個主題，那麼就會在消息到達時同時收到消息

6. JMS (Java Message Service) JAVA消息服務：

• 基於JVM消息的規範 。ActiveMQ、HornetMQ是JMS實現

7. AMQP (Advanced Message Queuing Protocol)

• 高級消息隊列協議，也是一個消息代理的規範，兼容JMS
• RabbitMQ 是 AMQP 的實現

 

JMS & AMQP區別 

 

 

 

 8. Spring 支持

• spring-jms提供了JMS的支持
• spring-rabbit提供了AMQP的支持
• 須要ConnectionFactory的實現來實現代理
• 提供JmsTemplate、RabbitTemplate來發送消息
• @JmsListener（JMS） 、@RabbitListener （AMQP）註解在方法上監聽消息代理髮布的消息
• @EnableJms、@EnableRabbit開啓支持

9. Spring Boot 配置

• JmsAutoConfiguration
• RabbitAutoConfiguration

 

 

3、Rabbit MQ 簡介

 

rabbitMQ是一個由erlang開發的AMQP（Advanced Message Queue Protocol）的開源實現

1. Message

   消息，消息是不具名的，它由消息頭和消息體組成。消息體不是透明的，而消息體則是由一系列的可選屬性組成，這些屬性包括routing-key（路郵鍵）、priority（相對於其它消息的優先權）、delivery-mode（指出該消息可能須要持久性存儲）等。

 

2. Publisher

   消息的生產者，也是一個向交換器發佈消息的客戶端應用程序

3. Exchange

  交換器，用來接收生產者發送的消息並將這些消息路由給服務器中的隊列。有4種裏類型：direct(默認)、fanout、topic、headers，不一樣類型的Exchange轉發消息策略有所區別。

 

4. queue

消息隊列，用來保存消息，直到發送給消費者。它是消息容器，也是消息的終點。一個消息可投入一個或多個隊列。消息一直在隊列裏面，等待消費者鏈接到這個隊列將消息取走。

5. Binding

綁定，用於消息隊列和交換器：Exchange之間的關聯。一個綁定就是基於路由鍵將交換器和消息隊列鏈接起來的路由規則，因此能夠將交換器理解成一個由綁定構成的路由表。

Exchange和Binding的關係能夠是多對多

 

6. Connection

網絡鏈接，好比一個TCP鏈接

 

7. Channle

信道，多路複用鏈接中的一條獨立的雙向數據流通道。信道是創建在真實的TCP鏈接內的虛擬鏈接，AMQP命令都是經過信道發送出去的。不論是發佈消息，訂閱隊列、仍是接收消息，這些動做都是經過信道完成。由於對於操做系統來講創建和銷燬TCP都是很是昂貴的開銷，因此引入了信道概念，以複用一條TCP鏈接。

 

8. Consumer

消息的消費者，表示一個從消息隊列中取得消息的客戶端應用程序。

 

9. Virtual Host

虛擬主機，表示一批交換器、消息隊列和相關對象。虛擬主機是共享相同的身份認證和加密環境的獨立服務器域。每一個Vhost本質上就是一個mini的RabbitMQ服務器，擁有本身的隊列、交換器、綁定和權限機制。vhost是AMQP概念的基礎，必須在鏈接時指定，RabbitMQ默認的 vhost是/。

10. Broker

表示消息隊列服務器實體

 

 

 

4、RabbitMQ運行機制

AMQP中的消息路由

• AMQP中消息的路由過程和Java開發者熟悉的JMS存在一些差異，AMQP增長了Exchange和Binding的角色，生產者把消息發送到Exchange上，消息隊列最終到達隊列並被消費者接收，而Binding決定交換器的消息應該發送到哪一個隊列。

 

 Exchange模型

Exchange分發消息時根據類型的不一樣分發策略也有所區別，目前有四種類型：direct、fanout、topic、headers。Headers匹配AMQP消息的header而不是路由鍵，headers交換器和direct交換器徹底一致，但性能差不少，目前幾乎用不到了，因此直接看另外三種類型：

1. Direct Exchange 

消息中的路由鍵（routing key）若是和Binding中的bingding key一致，交換器就將消息發到對應的隊列中。路由鍵與隊列名徹底匹配，若是一個隊列綁定到交換機要求路由鍵爲dog，則只轉發routing key爲dog的消息，不會轉發dog.puppy，也不會轉發dog.guard等等。它是徹底匹配，單播的模式。

 

2.fanout Exchange

每一個發到fanout類型交換器的消息都會分到全部綁定的隊列上去。fanout交換器不處理路由鍵，只是簡單地將隊列綁定到交換器上去，每一個發送到交換器的消息都會被轉發到該交換器綁定的全部隊列上。很像子網廣播，每臺子網內的主機都得到了一份複製消息。fanout類型轉發消息是最快的。

 

 

3. Topic Exchange

Topic交換器經過模式匹配分配消息的路由鍵屬性，將路由鍵和某個模式匹配，此時隊列須要綁定到一個模式上。它將路由鍵和綁定鍵的字符串切分紅單詞，這些單詞之間用點隔開。一樣也會識別2個通配符：符號「#」和符號「*」。#匹配0個或多個單詞，*匹配一個單詞。

 

 

 

5、安裝啓動運行ActiveMQ

1. 官網下載ActiveMQ獨立運行包

wget rabbitmq-server-mac-standalone-3.7.17.tar.xz tar -zxvf rabbitmq-server-mac-standalone-3.7.17.tar.xz cd rabbitmq_server-3.7.17/sbin ./rabbitmq-server #啓動服務 ./rabbitmq-plugins list #查看全部插件 ./rabbitmq-plugins enable rabbitmq_management #啓動web端管理插件

 

2. 添加exchange

 

3. 添加隊列

4. exchange中向queue發佈消息

 

 

 

5. 獲取消息 

 

 

 

 

6、RabbitMQ 整合

1. 引入spring-boot-starter-amqp

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

 

2.application.yml配置

spring.rabbitmq.port=5672 spring.rabbitmq.username=guest spring.rabbitmq.password=guest spring.rabbitmq.host=localhost

 

3. 測試RabbitMQ

AmqpAdmin：管理組件

RabbitTemplate：消息發送處理組件

package com.everjiankang.amqp; 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; import java.util.HashMap; import java.util.Map; @RunWith(SpringRunner.class) @SpringBootTest public class AmqpApplicationTests { //private MessageConverter messageConverter = new SimpleMessageConverter();
 @Autowired RabbitTemplate template; @Test public void sendMsg() { /*Message須要本身構造一個；定義消息體內容和消息頭*/
        //template.send(exchange,routingKey,message);

        /* object默認被當成消息體，只須要傳入發送的對象，自動序列化發送給rabbitmq */
        //template.convertAndSend(exchange,routingKey,object);
 Map<String,Object> map = new HashMap(); map.put("name","xiaochao"); map.put("age",20); map.put("sex","男"); map.put("msg","我愛你中國"); //對象被默認序列化之後發送出去
        template.convertAndSend("exchange.direct","atguigu.news",map); } /** * 獲取消息 */ @Test public void receiveMsg() { Object o = template.receiveAndConvert("atguigu.news"); System.out.println(o.getClass()); System.out.println(o); } }

 

 

查看隊列：

get Message

 

 亂碼緣由：序列號採用了JDK自己的序列化機制

解決亂碼：更換SimpleMessageConverter ——> Jackson2JsonMessageConverter

 

 

@Configuration public class MyConfig { @Bean public MessageConverter getMessageConverter() { return new Jackson2JsonMessageConverter(); } }