一篇文章帶你使用 SpringBoot 整合 RabbitMQ

一、認識 RabbitMQ

RabbitMQ 簡介以 AMQP 協定：

（1）RabbitMQ 是開源的訊息代理和佇列伺服器，用來通過普通協定在完全不同的應用之間共用資料，RabbitMQ 底層是用了 Erlang 語言來編寫的，並且 RabbitMQ 是基於 AMQP 協定的.
（2）RabbitMQ 不僅僅可以使用 java 使用者端進行編寫，且可以使用其他的語言（python，php等…）,它提供了豐富的API

RabbitMQ 的優點：

（1）開源，效能優秀，穩定性保障
（2）與 SpringAMQP 完美的整合，API 豐富 （Spring基於 RabbitMQ 提供了一套框架，叫做AMQP 框架）這套框架不僅提呈了原生的 RabbitMQ，而且還提供了豐富可擴張的API幫助開發人員更好的去應用
（3）叢集模式豐富，表示式設定，HA模式，映象佇列模型
說明：（保證資料不丟失的提前做到高可靠性，可用性）普遍使用的映象佇列模式
（4）AMQP 全稱：Advanced Message Queuing Protocl AMQP 翻譯過來：高階訊息佇列協定

二、安裝 RabbitMQ

這裡為了方便在 docker 上安裝 RabbitMQ
（1）首先搜尋 RabbitMQ 的安裝命令：https://hub.docker.com/_/rabbitmq

 docker run -d --hostname my-rabbit --name some-rabbit rabbitmq:3-management

安裝成功：

對於對映的埠需要注意兩個：15672 和 5672，其中 15672 是管理的埠，5672 是通訊的埠

可以看到docker容器 15672 對應的主機對映埠是 32771，

預設的使用者名稱和密碼都是 guest

三、SpringBoot 整合 RabbitMQ

1. 加入依賴

2. 設定 application.properties

注意這裡的埠選擇 32781，對應rabbitmq 容器的埠是 5672，選擇通訊埠，切忌不用選擇管理埠

spring.rabbitmq.host=192.168.245.133
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest
spring.rabbitmq.port=32781

3. 直接交換模式

直接交換模式可以參考：https://blog.csdn.net/fakerswe/article/details/81508963

所謂「直接連線交換機」就是：Producer(生產者)投遞的訊息被DirectExchange (交換機)轉發到通過routingkey繫結到具體的某個Queue(佇列)，把訊息放入佇列,然後Consumer從Queue中訂閱訊息

這裡的具體某個佇列是由引數 routingKey 控制的，訊息是通過這個引數往佇列上發的

RabbitMQ訊息模型的核心思想（core idea）: 生產者會把訊息傳送給RabbitMQ的交換中心（Exchange），Exchange的一側是生產者，另一側則是一個或多個佇列，由Exchange決定一條訊息的生命週期–傳送給某些佇列，或者直接丟棄掉。

（1）設定 RabbitDirectConfig

@Configuration
public class RabbitDirectConfig {

    public final static String DIRECTNAME = "yolo-direct";

    // 訊息佇列
    @Bean
    Queue queue() {
        return new Queue("hello.yolo");
    }

    @Bean
    DirectExchange directExchange() {
        return new DirectExchange(DIRECTNAME, true, false);
    }
    // 將 queue 和 directExchange 繫結到一起
    @Bean
    Binding binding() {
        return BindingBuilder.bind(queue()).to(directExchange()).with("direct");
    }
}

（2）消費者

@Component
public class DirectReceiver {
    // 監聽佇列
    @RabbitListener(queues = "hello.yolo")
    public void handler1(String msg) {
        System.out.println("handler1>>>" + msg);
    }
}

（3）測試：生產者

 	@Autowired
    RabbitTemplate rabbitTemplate;
    @Test
    public void contextLoads() {
        //將訊息轉發到 routingKey 為 hello.yolo 的佇列，對應 DirectReceiver 的監聽佇列名
        rabbitTemplate.convertAndSend("hello.yolo", "hello yolo! ni hao!");
    }

4. 廣播模式

可以參考：https://blog.csdn.net/fakerswe/article/details/81455340

簡單的講，就是把交換機（Exchange）裡的訊息傳送給所有繫結該交換機的佇列，忽略routingKey。

由圖可知，生產者把訊息傳送到交換機後，由交換機傳送給消費者佇列。消費者佇列如果想要接收到交換機裡的訊息，那麼需要保證：佇列繫結的交換機名稱要和交換機一致，這個是廣播模式的關鍵，也是MQ後續所有模式最粗略的前提。

這裡訊息是通過生產者發往交換機的，然後交換機再傳送給繫結的佇列
（1）設定廣播模式

@Configuration
public class RabbitFanoutConfig {
    public static final String FANOUTNAME = "yolo-fanout";

    /**
     * 佇列1
     * @return
     */
    @Bean
    Queue queueOne() {
        return new Queue("queue-one");
    }

    /**
     * 佇列2
     * @return
     */
    @Bean
    Queue queueTwo() {
        return new Queue("queue-two");
    }

    /**
     * 交換機
     * @return
     */
    @Bean
    FanoutExchange fanoutExchange() {
        return new FanoutExchange(FANOUTNAME, true, false);
    }

    /**
     * 繫結佇列1
     * @return
     */
    @Bean
    org.springframework.amqp.core.Binding bindingOne() {
        return BindingBuilder.bind(queueOne()).to(fanoutExchange());
    }

    /**
     * 繫結佇列2
     * @return
     */
    @Bean
    Binding bindingTwo() {
        return BindingBuilder.bind(queueTwo()).to(fanoutExchange());
    }
}

（2）消費者

/**
 * 定義接收器：消費者
 */
@Component
public class FanoutReceiver {
    /**
     * 接收訊息佇列1
     * @param msg
     */
    @RabbitListener(queues = "queue-one")
    public void handler1(String msg) {
        System.out.println("FanoutReceiver:handler1:" + msg);
    }
    /**
     * 接收訊息佇列2
     * @param msg
     */
    @RabbitListener(queues = "queue-two")
    public void handler2(String msg) {
        System.out.println("FanoutReceiver:handler2:" + msg);
    }
}

（3）測試：生產者

  /**
     * 往交換機上傳送資訊
     */
    @Test
    public void test1() {
        rabbitTemplate.convertAndSend(RabbitFanoutConfig.FANOUTNAME,null,"hello fanout!");
    }

這裡需要注意，需要先啟動消費者，再啟動生產者，否則先啟動生產者，exchange接收到訊息後發現沒有佇列對其感興趣，就會將訊息丟掉，此時跟 routingKey 無關

佇列1和佇列2 均收到了訊息

5. 主題路由匹配模式

可參考：https://blog.csdn.net/weixin_43770545/article/details/90902788

假如你想在淘寶上買一雙運動鞋，那麼你是不是會在搜尋方塊中搜「XXX運動鞋」，這個時候系統將會模糊匹配的所有符合要求的運動鞋，然後展示給你。
所謂「主題路由匹配交換機」也是這樣一個道理，但是使用時也有一定的規則。

String routingkey = 「testTopic.#」;
String routingkey = 「testTopic.*」;

* 表示只匹配一個詞
#表示匹配多個詞

（1）設定 topic 模式

@Configuration
public class RabbitTopicConfig {
    public static final String TOPICNAME = "yolo-topic";

    @Bean
    TopicExchange topicExchange() {
        return new TopicExchange(TOPICNAME, true, false);
    }

    @Bean
    Queue xiaomi() {
        return new Queue("xiaomi");
    }

    @Bean
    Queue huawei() {
        return new Queue("huawei");
    }

    @Bean
    Queue phone() {
        return new Queue("phone");
    }

    @Bean
    Binding xiaomiBinding() {
        //xiaomi.# 表示如果路由的 routingKey 是以xiaomi 開頭就會傳送到 xiaomi 這個佇列上
        return BindingBuilder.bind(xiaomi()).to(topicExchange()).with("xiaomi.#");
    }

    @Bean
    Binding huaweiBinding() {
        //huawei.#
        return BindingBuilder.bind(huawei()).to(topicExchange()).with("huawei.#");
    }
    @Bean
    Binding phoneBinding() {
        // #.phone.# 表示routingKey 中包含 phone 就會被傳送到 phone 這個佇列上
        return BindingBuilder.bind(phone()).to(topicExchange()).with("#.phone.#");
    }
}

（2） 消費者

@Component
public class TopicReceiver {
    @RabbitListener(queues = "xiaomi")
    public void handler1(String msg) {
        System.out.println("TopicReceiver:handler1:" + msg);
    }

    @RabbitListener(queues = "huawei")
    public void handler2(String msg) {
        System.out.println("TopicReceiver:handler2:" + msg);
    }

    @RabbitListener(queues = "phone")
    public void handler3(String msg) {
        System.out.println("TopicReceiver:handler3:" + msg);
    }
}

（3）測試：生產者

  @Test
    public void test2() {
        //可以被小米的佇列收到
        rabbitTemplate.convertAndSend(RabbitTopicConfig.TOPICNAME, "xiaomi.news", "小米新聞");
        //可以被手機的佇列收到
        rabbitTemplate.convertAndSend(RabbitTopicConfig.TOPICNAME, "vivo.phone", "vivo 手機");
        //可以被華為和手機的佇列收到
        rabbitTemplate.convertAndSend(RabbitTopicConfig.TOPICNAME, "huawei.phone", "華為手機");
    }

6. header 模式

這種模式使用的是 header 中的 key/value (鍵值對) 匹配佇列，也和 routingKey 無關

（1）設定 config

@Configuration
public class RabbitHeaderConfig {
    public static final String HEADERNAME = "yolo-header";

    @Bean
    HeadersExchange headersExchange() {
        return new HeadersExchange(HEADERNAME, true, false);
    }

    @Bean
    Queue queueName() {
        return new Queue("name-queue");
    }

    @Bean
    Queue queueAge() {
        return new Queue("age-queue");
    }

    @Bean
    Binding bindingName() {
        Map<String, Object> map = new HashMap<>();
        //
        map.put("name", "yolo");
        //whereAny 表示訊息的header中只要有一個header匹配上map中的key,value,就把訊息傳送到對應的佇列上
        return BindingBuilder.bind(queueName()).to(headersExchange()).whereAny(map).match();
    }

    @Bean
    Binding bindingAge() {
        //只要有，age 這個欄位，就傳送到相應的佇列上去
        return BindingBuilder.bind(queueAge()).to(headersExchange()).where("age").exists();
    }
}

（2）消費者

@Component
public class HeaderReceiver {
    @RabbitListener(queues = "name-queue")
    public void handler1(byte[] msg) {
        System.out.println("HeaderReceiver:handler1:" + new String(msg, 0, msg.length));
    }

    @RabbitListener(queues = "age-queue")
    public void handler2(byte[] msg) {
        System.out.println("HeaderReceiver:handler2:" + new String(msg, 0, msg.length));
    }
}

（3）測試

  @Test
    public void test3() {
        //對應 RabbitHeaderConfig 中的map 的 key / value
        Message nameMsg = MessageBuilder.withBody("hello yolo !".getBytes()).setHeader("name","yolo").build();
        Message ageMsg = MessageBuilder.withBody("hello 99 !".getBytes()).setHeader("age","99").build();
        //此時傳送的訊息接收，跟 routingKey無關，跟訊息的 header 內容有關
        rabbitTemplate.send(RabbitHeaderConfig.HEADERNAME, null, ageMsg);
        rabbitTemplate.send(RabbitHeaderConfig.HEADERNAME, null, nameMsg);
    }

如果更改，header 裡的鍵值對：則無法匹配成功，佇列收不到該資訊