springBoot整合rabbitmq測(cè)試常用模型小結(jié)

更新時(shí)間：2022年01月08日 09:22:44 作者：穿條秋褲到處跑

這篇文章主要介紹了springBoot整合rabbitmq并測(cè)試五種常用模型,本文主要針對(duì)前五種常用模型，在spirngboot框架的基礎(chǔ)上整合rabbitmq并進(jìn)行測(cè)試使用，需要的朋友可以參考下

之前我們記錄了原生java代碼使用rabbitmq的方法，很簡(jiǎn)單，類(lèi)似于原生jdbc代碼一樣，將連接對(duì)象抽離出來(lái)作為工具類(lèi)，生產(chǎn)者和消費(fèi)者通過(guò)工具類(lèi)獲取連接對(duì)象，進(jìn)而獲取通道對(duì)象，再注冊(cè)交換機(jī)或者是隊(duì)列等，發(fā)送消息與接收消息。
在企業(yè)開(kāi)發(fā)中，我們更多的是使用spring框架來(lái)整合其它技術(shù)，springboot更是方便的提供了各種starter來(lái)快速添加依賴(lài)，完成整合，開(kāi)箱即用。

1.添加依賴(lài)

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

2.編寫(xiě)配置

配置信息包括ip，端口，虛擬主機(jī)，用戶(hù)名和密碼，和原生java代碼所需的配置信息一致。

spring:
  application:
    name: spirngboot-rabbitmq
  rabbitmq:
    host: 192.168.20.128
    port: 5672
    virtual-host: /vh
    username: wuwl
    password: 123456

3.編寫(xiě)并測(cè)試

本文主要針對(duì)前五種常用模型，在spirngboot框架的基礎(chǔ)上整合rabbitmq并進(jìn)行測(cè)試使用。

在這里插入圖片描述

(1) Hello World模型

這是一種簡(jiǎn)單的直連模型，生產(chǎn)者將消息直接發(fā)送至消息隊(duì)列，消費(fèi)者綁定消息隊(duì)列后直接獲取，一對(duì)一。
spring-boot-starter-amqp為我們提供了一個(gè)org.springframework.amqp.rabbit.core.RabbitTemplate類(lèi)來(lái)方便我們使用rabbitmq，自動(dòng)注入即可。

生產(chǎn)者測(cè)試類(lèi)：

@SpringBootTest(classes = RabbitmqDemoApplication.class)
@RunWith(SpringRunner.class)
public class RabbitmqDemoApplicationTests {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    @Test
    public void testHelloQueues(){
            rabbitTemplate.convertAndSend("hello","hello world");
    }
}

生產(chǎn)者向名為hello的隊(duì)列發(fā)送消息，但是，在沒(méi)有消費(fèi)者的情況下，生產(chǎn)者沒(méi)有任何意義。另外，convertAndSend方法的第一個(gè)參數(shù)并不是消息隊(duì)列的意思，而是routingKey，我們根據(jù)源碼找到最初定義的接口可以看到以下內(nèi)容：

/**
	 * Convert a Java object to an Amqp {@link Message} and send it to a default exchange
	 * with a specific routing key.
	 *
	 * @param routingKey the routing key
	 * @param message a message to send
	 * @throws AmqpException if there is a problem
	 */
	void convertAndSend(String routingKey, Object message) throws AmqpException;

第二個(gè)參數(shù)為Object類(lèi)型，也就是說(shuō)可以傳遞任意類(lèi)型的對(duì)象，該方法將對(duì)象轉(zhuǎn)換成一個(gè)Amqp消息并發(fā)送到一個(gè)默認(rèn)的交換機(jī)，并且routingKey為第一個(gè)參數(shù)的內(nèi)容，沒(méi)有提到消息隊(duì)列的信息，但我們可以分析到，這里的routingKey與queues應(yīng)該是同名的。

消費(fèi)者類(lèi)：

@Component
@RabbitListener(queuesToDeclare = @Queue("hello"))
public class HelloQueuesConsumer {
    @RabbitHandler
    public void consume(String msg){
        System.out.println("消費(fèi)消息：" + msg + " " + System.currentTimeMillis());
    }
}

上面的代碼等同于：

@Component
public class HelloQueuesConsumer {
    @RabbitListener(queuesToDeclare = @Queue("hello"))
    public void consume(String msg){
        System.out.println("消費(fèi)消息：" + msg + " " + System.currentTimeMillis());
    }
}

@RabbitListener 可以標(biāo)注在類(lèi)上面，需配合 @RabbitHandler 注解一起使用

@RabbitListener 標(biāo)注在類(lèi)上面表示當(dāng)有收到消息的時(shí)候，就交給 @RabbitHandler 的方法處理，具體使用哪個(gè)方法處理，根據(jù) MessageConverter 轉(zhuǎn)換后的參數(shù)類(lèi)型

直接啟動(dòng)測(cè)試方法，也就是生產(chǎn)者，可以看到：

在這里插入圖片描述

消費(fèi)者有接收到消息隊(duì)列中的信息并打印。

(2) work queues模型

生產(chǎn)者測(cè)試方法，類(lèi)與第一個(gè)模型一致

@Test
public void testWorkQueues(){
    for (int i = 0; i < 20; i++) {
        rabbitTemplate.convertAndSend("work","work index " + i);
    }
}

消費(fèi)者類(lèi)：

@Component
public class WorkQueuesConsumer {
    @RabbitListener(queuesToDeclare = @Queue("work"))
    public void consume1(String msg){
        System.out.println("consumer1消費(fèi)消息：" + msg);
    }
    @RabbitListener(queuesToDeclare = @Queue("work"))
    public void consume2(String msg){
        System.out.println("consumer2消費(fèi)消息：" + msg);
    }
}

啟動(dòng)生產(chǎn)者測(cè)試方法：

在這里插入圖片描述

消費(fèi)者一與消費(fèi)者二均勻分配了隊(duì)列中的消息任務(wù)，即使兩者執(zhí)行效率不一致，也同樣是均勻分配。

(3) Publish/Subscribe模型

生產(chǎn)者測(cè)試方法：

for (int i = 0; i < 20; i++) {
    rabbitTemplate.convertAndSend("amq.fanout","","fanout msg " + i);
}

消費(fèi)者類(lèi)：

@Component
public class FanoutQueuesConsumer {
    @RabbitListener(bindings = {
            @QueueBinding(value = @Queue,
                    exchange = @Exchange(
                            value = "amq.fanout",
                            type = "fanout"))})
    public void consume1(String msg) {
        System.out.println("consumer1消費(fèi)消息：" + msg);
    }
    @RabbitListener(bindings = {
            @QueueBinding(value = @Queue,
                    exchange = @Exchange(
                            value = "amq.fanout",
                            type = "fanout"))})
    public void consume2(String msg) {
        System.out.println("consumer2消費(fèi)消息：" + msg);
    }
}

注意此處的交換機(jī)信息

啟動(dòng)生產(chǎn)者測(cè)試方法：

在這里插入圖片描述

此處只粘貼了部分打印信息，兩個(gè)消費(fèi)者獲得了相同的消息，生產(chǎn)者將消息發(fā)送至交換機(jī)，由交換機(jī)發(fā)送至已注冊(cè)到交換機(jī)的所有臨時(shí)消息隊(duì)列，進(jìn)而消費(fèi)者獲取隊(duì)列中的消息。

(4) Routing模型

生產(chǎn)者測(cè)試方法：

@Test
public void testDirectQueues(){
    rabbitTemplate.convertAndSend("amq.direct","info","routingKey is info");
    rabbitTemplate.convertAndSend("amq.direct","warn","routingKey is warn");
    rabbitTemplate.convertAndSend("amq.direct","error","routingKey is error");
}

routing也成為fanout模型，對(duì)應(yīng)的交換機(jī)類(lèi)型為direct

消費(fèi)者類(lèi)：

@Component
public class DirectQueuesConsumer {
    @RabbitListener(bindings = {
            @QueueBinding(value = @Queue,
                    exchange = @Exchange(
                            value = "amq.direct",
                            type = "direct"),
                    key = {"info", "warn", "error"})})
    public void consume1(String msg) {
        System.out.println("consumer1消費(fèi)消息：" + msg);
    }

    @RabbitListener(bindings = {
            @QueueBinding(value = @Queue,
                    exchange = @Exchange(
                            value = "amq.direct",
                            type = "direct"),
                    key = "error")})
    public void consume2(String msg) {
        System.out.println("consumer2消費(fèi)消息：" + msg);
    }
}

啟動(dòng)生產(chǎn)者測(cè)試類(lèi)：
在這里插入圖片描述
消費(fèi)者一配置了三種類(lèi)型的routingKey，所以三種類(lèi)型的消息都能夠接收到，消費(fèi)者二只能接受到error類(lèi)型的消息。

(5) Topic模型

生產(chǎn)者測(cè)試方法：

@Test
public void testTopicQueues(){
    rabbitTemplate.convertAndSend("amq.topic","file.info","routingKey is info");
    rabbitTemplate.convertAndSend("amq.topic","file.warn","routingKey is warn");
    rabbitTemplate.convertAndSend("amq.topic","file.error","routingKey is error");
}

消費(fèi)者類(lèi)：

@Component
public class TopicQueuesConsumer {
    @RabbitListener(bindings = {
            @QueueBinding(value = @Queue,
                    exchange = @Exchange(
                            value = "amq.topic",
                            type = "topic"),
                    key = {"#"})})
    public void consume1(String msg) {
        System.out.println("consumer1消費(fèi)消息：" + msg);
    }

    @RabbitListener(bindings = {
            @QueueBinding(value = @Queue,
                    exchange = @Exchange(
                            value = "amq.topic",
                            type = "topic"),
                    key = "*.error")})
    public void consume2(String msg) {
        System.out.println("consumer2消費(fèi)消息：" + msg);
    }
}

啟動(dòng)生產(chǎn)者測(cè)試方法：

在這里插入圖片描述