spring-boot（八）RabbitMQ的使用

阿新 • • 發佈：2018-11-16

RabbitMQ 即一個訊息佇列，主要是用來實現應用程式的非同步和解耦，同時也能起到訊息緩衝，訊息分發的作用。

訊息中介軟體在網際網路公司的使用中越來越多，剛才還看到新聞阿里將RocketMQ捐獻給了apache，當然了今天的主角還是講RabbitMQ。訊息中介軟體最主要的作用是解耦，中介軟體最標準的用法是生產者生產訊息傳送到佇列，消費者從佇列中拿取訊息並處理，生產者不用關心是誰來消費，消費者不用關心誰在生產訊息，從而達到解耦的目的。在分散式的系統中，訊息佇列也會被用在很多其它的方面，比如：分散式事務的支援，RPC的呼叫等等。

以前一直使用的是ActiveMQ，在實際的生產使用中也出現了一些小問題，在網路查閱了很多的資料後，決定嘗試使用RabbitMQ來替換ActiveMQ，RabbitMQ的高可用性、高效能、靈活性等一些特點吸引了我們，查閱了一些資料整理出此文。

RabbitMQ介紹

RabbitMQ是實現AMQP（高階訊息佇列協議）的訊息中介軟體的一種，最初起源於金融系統，用於在分散式系統中儲存轉發訊息，在易用性、擴充套件性、高可用性等方面表現不俗。RabbitMQ主要是為了實現系統之間的雙向解耦而實現的。當生產者大量產生資料時，消費者無法快速消費，那麼需要一箇中間層。儲存這個資料。

AMQP，即Advanced Message Queuing Protocol，高階訊息佇列協議，是應用層協議的一個開放標準，為面向訊息的中介軟體設計。訊息中介軟體主要用於元件之間的解耦，訊息的傳送者無需知道訊息使用者的存在，反之亦然。AMQP的主要特徵是面向訊息、佇列、路由（包括點對點和釋出/訂閱）、可靠性、安全。

RabbitMQ是一個開源的AMQP實現，伺服器端用Erlang語言編寫，支援多種客戶端，如：Python、Ruby、.NET、Java、JMS、C、PHP、ActionScript、XMPP、STOMP等，支援AJAX。用於在分散式系統中儲存轉發訊息，在易用性、擴充套件性、高可用性等方面表現不俗。

交換機(Exchange)

交換機的功能主要是接收訊息並且轉發到繫結的佇列，交換機不儲存訊息，在啟用ack模式後，交換機找不到佇列會返回錯誤。交換機有四種類型：Direct, topic, Headers and Fanout

Direct：direct 型別的行為是"先匹配, 再投送". 即在繫結時設定一個 routing_key, 訊息的routing_key 匹配時, 才會被交換器投送到繫結的佇列中去.
Topic：按規則轉發訊息（最靈活）
Headers：設定header attribute引數型別的交換機
Fanout：轉發訊息到所有繫結佇列

Direct Exchange
Direct Exchange是RabbitMQ預設的交換機模式，也是最簡單的模式，根據key全文匹配去尋找佇列。

第一個 X - Q1 就有一個 binding key，名字為 orange； X - Q2 就有 2 個 binding key，名字為 black 和 green。當訊息中的路由鍵和這個 binding key 對應上的時候，那麼就知道了該訊息去到哪一個佇列中。

Ps：為什麼 X 到 Q2 要有 black，green，2個 binding key呢，一個不就行了嗎？ - 這個主要是因為可能又有 Q3，而Q3只接受 black 的資訊，而Q2不僅接受black 的資訊，還接受 green 的資訊。

Topic Exchange

Topic Exchange 轉發訊息主要是根據萬用字元。 在這種交換機下，佇列和交換機的繫結會定義一種路由模式，那麼，萬用字元就要在這種路由模式和路由鍵之間匹配後交換機才能轉發訊息。

在這種交換機模式下：

路由鍵必須是一串字元，用句號（.）隔開，比如說 agreements.us，或者 agreements.eu.stockholm 等。
路由模式必須包含一個星號（*），主要用於匹配路由鍵指定位置的一個單詞，比如說，一個路由模式是這樣子：agreements..b.*，那麼就只能匹配路由鍵是這樣子的：第一個單詞是 agreements，第四個單詞是 b。井號（#）就表示相當於一個或者多個單詞，例如一個匹配模式是agreements.eu.berlin.#，那麼，以agreements.eu.berlin開頭的路由鍵都是可以的。

具體程式碼傳送的時候還是一樣，第一個引數表示交換機，第二個引數表示routing key，第三個引數即訊息。如下：

rabbitTemplate.convertAndSend("testTopicExchange","key1.a.c.key2", " this is  RabbitMQ!");

topic 和 direct 類似, 只是匹配上支援了"模式", 在"點分"的 routing_key 形式中, 可以使用兩個萬用字元:

*表示一個詞.
#表示零個或多個詞.

Headers Exchange

headers 也是根據規則匹配, 相較於 direct 和 topic 固定地使用 routing_key , headers 則是一個自定義匹配規則的型別.
在佇列與交換器繫結時, 會設定一組鍵值對規則, 訊息中也包括一組鍵值對( headers 屬性), 當這些鍵值對有一對, 或全部匹配時, 訊息被投送到對應佇列.

Fanout Exchange

Fanout Exchange 訊息廣播的模式，不管路由鍵或者是路由模式，會把訊息發給繫結給它的全部佇列，如果配置了routing_key會被忽略。

springboot整合RabbitMQ

springboot整合RabbitMQ非常簡單，如果只是簡單的使用配置非常少，springboot提供了spring-boot-starter-amqp專案對訊息各種支援。

簡單使用

1、配置pom包，主要是新增spring-boot-starter-amqp的支援

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

2、配置檔案

配置rabbitmq的安裝地址、埠以及賬戶資訊

spring.application.name=spirng-boot-rabbitmq

spring.rabbitmq.host=192.168.0.86
spring.rabbitmq.port=5672
spring.rabbitmq.username=admin
spring.rabbitmq.password=123456

3、佇列配置

@Configuration
public class RabbitConfig {

    @Bean
    public Queue Queue() {
        return new Queue("hello");
    }

}

3、傳送者

rabbitTemplate是springboot 提供的預設實現

public class HelloSender {

    @Autowired
    private AmqpTemplate rabbitTemplate;

    public void send() {
        String context = "hello " + new Date();
        System.out.println("Sender : " + context);
        this.rabbitTemplate.convertAndSend("hello", context);
    }

}

4、接收者

@Component
@RabbitListener(queues = "hello")
public class HelloReceiver {

    @RabbitHandler
    public void process(String hello) {
        System.out.println("Receiver  : " + hello);
    }

}

5、測試

@RunWith(SpringRunner.class)
@SpringBootTest
public class RabbitMqHelloTest {

    @Autowired
    private HelloSender helloSender;

    @Test
    public void hello() throws Exception {
        helloSender.send();
    }

}

注意，傳送者和接收者的queue name必須一致，不然不能接收

多對多使用

一個傳送者，N個接收者或者N個傳送者和N個接收者會出現什麼情況呢？

一對多傳送
對上面的程式碼進行了小改造,接收端註冊了兩個Receiver,Receiver1和Receiver2，傳送端加入引數計數，接收端列印接收到的引數，下面是測試程式碼，傳送一百條訊息，來觀察兩個接收端的執行效果

@Test
public void oneToMany() throws Exception {
    for (int i=0;i<100;i++){
        neoSender.send(i);
    }
}

結果如下：

Receiver 1: spirng boot neo queue ****** 11
Receiver 2: spirng boot neo queue ****** 12
Receiver 2: spirng boot neo queue ****** 14
Receiver 1: spirng boot neo queue ****** 13
Receiver 2: spirng boot neo queue ****** 15
Receiver 1: spirng boot neo queue ****** 16
Receiver 1: spirng boot neo queue ****** 18
Receiver 2: spirng boot neo queue ****** 17
Receiver 2: spirng boot neo queue ****** 19
Receiver 1: spirng boot neo queue ****** 20

根據返回結果得到以下結論

一個傳送者，N個接受者,經過測試會均勻的將訊息傳送到N個接收者中

多對多傳送

複製了一份傳送者，加入標記，在一百個迴圈中相互交替傳送

@Test
    public void manyToMany() throws Exception {
        for (int i=0;i<100;i++){
            neoSender.send(i);
            neoSender2.send(i);
        }
}

結果如下：

Receiver 1: spirng boot neo queue ****** 20
Receiver 2: spirng boot neo queue ****** 20
Receiver 1: spirng boot neo queue ****** 21
Receiver 2: spirng boot neo queue ****** 21
Receiver 1: spirng boot neo queue ****** 22
Receiver 2: spirng boot neo queue ****** 22
Receiver 1: spirng boot neo queue ****** 23
Receiver 2: spirng boot neo queue ****** 23
Receiver 1: spirng boot neo queue ****** 24
Receiver 2: spirng boot neo queue ****** 24
Receiver 1: spirng boot neo queue ****** 25
Receiver 2: spirng boot neo queue ****** 25

結論：和一對多一樣，接收端仍然會均勻接收到訊息

高階使用

物件的支援

springboot以及完美的支援物件的傳送和接收，不需要格外的配置。

//傳送者
public void send(User user) {
    System.out.println("Sender object: " + user.toString());
    this.rabbitTemplate.convertAndSend("object", user);
}

...

//接受者
@RabbitHandler
public void process(User user) {
    System.out.println("Receiver object : " + user);
}

結果如下：

Sender object: User{name='neo', pass='123456'}
Receiver object : User{name='neo', pass='123456'}

Topic Exchange

topic 是RabbitMQ中最靈活的一種方式，可以根據routing_key自由的繫結不同的佇列

首先對topic規則配置，這裡使用兩個佇列來測試

@Configuration
public class TopicRabbitConfig {

    final static String message = "topic.message";
    final static String messages = "topic.messages";

    @Bean
    public Queue queueMessage() {
        return new Queue(TopicRabbitConfig.message);
    }

    @Bean
    public Queue queueMessages() {
        return new Queue(TopicRabbitConfig.messages);
    }

    @Bean
    TopicExchange exchange() {
        return new TopicExchange("exchange");
    }

    @Bean
    Binding bindingExchangeMessage(Queue queueMessage, TopicExchange exchange) {
        return BindingBuilder.bind(queueMessage).to(exchange).with("topic.message");
    }

    @Bean
    Binding bindingExchangeMessages(Queue queueMessages, TopicExchange exchange) {
        return BindingBuilder.bind(queueMessages).to(exchange).with("topic.#");
    }
}

使用queueMessages同時匹配兩個佇列，queueMessage只匹配"topic.message"佇列

public void send1() {
    String context = "hi, i am message 1";
    System.out.println("Sender : " + context);
    this.rabbitTemplate.convertAndSend("exchange", "topic.message", context);
}

public void send2() {
    String context = "hi, i am messages 2";
    System.out.println("Sender : " + context);
    this.rabbitTemplate.convertAndSend("exchange", "topic.messages", context);
}

傳送send1會匹配到topic.#和topic.message 兩個Receiver都可以收到訊息，傳送send2只有topic.#可以匹配所有隻有Receiver2監聽到訊息

Fanout Exchange

Fanout 就是我們熟悉的廣播模式或者訂閱模式，給Fanout交換機發送訊息，綁定了這個交換機的所有佇列都收到這個訊息。

Fanout 相關配置

@Configuration
public class FanoutRabbitConfig {

    @Bean
    public Queue AMessage() {
        return new Queue("fanout.A");
    }

    @Bean
    public Queue BMessage() {
        return new Queue("fanout.B");
    }

    @Bean
    public Queue CMessage() {
        return new Queue("fanout.C");
    }

    @Bean
    FanoutExchange fanoutExchange() {
        return new FanoutExchange("fanoutExchange");
    }

    @Bean
    Binding bindingExchangeA(Queue AMessage,FanoutExchange fanoutExchange) {
        return BindingBuilder.bind(AMessage).to(fanoutExchange);
    }

    @Bean
    Binding bindingExchangeB(Queue BMessage, FanoutExchange fanoutExchange) {
        return BindingBuilder.bind(BMessage).to(fanoutExchange);
    }

    @Bean
    Binding bindingExchangeC(Queue CMessage, FanoutExchange fanoutExchange) {
        return BindingBuilder.bind(CMessage).to(fanoutExchange);
    }

}

這裡使用了A、B、C三個佇列繫結到Fanout交換機上面，傳送端的routing_key寫任何字元都會被忽略：

public void send() {
        String context = "hi, fanout msg ";
        System.out.println("Sender : " + context);
        this.rabbitTemplate.convertAndSend("fanoutExchange","", context);
}

結果如下：

Sender : hi, fanout msg 
...
fanout Receiver B: hi, fanout msg 
fanout Receiver A  : hi, fanout msg 
fanout Receiver C: hi, fanout msg

結果說明，繫結到fanout交換機上面的佇列都收到了訊息

上面所有的例子，程式碼在這裡

參考

RabbitMQ 使用參考

RabbitMQ：Spring 整合 RabbitMQ 與其概念，訊息持久化，ACK機制

spring-boot（八）RabbitMQ的使用

RabbitMQ介紹

相關概念

交換機(Exchange)

springboot整合RabbitMQ

簡單使用

多對多使用

高階使用

參考

spring-boot（八）RabbitMQ的使用

Spring Boot（十三）RabbitMQ安裝與集成

Spring Boot （八）MyBatis + Docker + MongoDB 4.x

Spring Boot（十三）RabbitMQ安裝與整合

Spring Boot （八）： Mybatis 增強工具 MyBatis-Plus

Spring Boot （十三）： Spring Boot 整合 RabbitMQ

spring boot （三）：熱部署

Spring boot（二）—Web開發

Spring實戰（八）bean裝配的運行時值註入——屬性占位符和SpEL

Spring Cloud（八）：配置中心（服務化與高可用）【Finchley 版】

Spring Boot（一）Hello World

spring boot（一）

spring boot（二）websocket

spring boot（三）數據訪問

spring boot（四）事務與緩存

Spring Boot（一）—— Spring Boot 入門

Spring Boot（七）擴展分析

Spring Boot （1）

Spring boot（一）----入門篇

Spring Boot （七）MyBatis代碼自動生成和輔助插件

spring-boot（八）RabbitMQ的使用

RabbitMQ介紹

相關概念

交換機(Exchange)

springboot整合RabbitMQ

簡單使用

多對多使用

高階使用

參考

相關推薦