一篇文章帶你使用 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 里的鍵值對：則無法匹配成功，佇列收不到該資訊