SpringBoot整合RabbitMQ实现五种消息模型

叁歲伎倆 2022-10-10 12:51 113阅读 0赞

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQyMTA5NzQ2_size_16_color_FFFFFF_t_70][]

### 一、什么是消息队列？ ###

消息，可以理解为两个应用之间传递的数据，数据可以是基本数据类型，也可以是对象等。

消息队列，则是容器，生产者产生的消息存放在这个容器里面。

MQ的整个过程可以理解为生产者生产消息，然后放进消息队列，消费者从消息队列里面获取消息最后进行消费。

### 二、消息模型 ###

目前RabbitMQ的官网上介绍共有7种模型，最后两种RPC和Publisher Confirms本篇文章暂不讲，这两种目前实际应用很少，接下来我会结合官网的解释进行介绍和整合SpringBoot。

在进行整合前需要先安装RabbitMQ，可以根据我之前的一篇文章进行安装：

[RabbitMQ的安装教程][RabbitMQ]

模型，即RabbitMQ提供7种不同的模式，这7种模式可以应用于不同的场景。

PS：关于RabbitMQ的作用我这里就不多讲了，相信看到我这篇文章之前也大致知道是做什么的，无非就是解耦、异步、削峰。

### 三、搭建SpringBoot项目 ###

接下来就是熟悉的三部曲：导依赖、写配置、编代码

pom依赖：

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

application.yml：

spring:
      rabbitmq:
        host: 127.0.0.1
        port: 5672
        username: guest
        password: guest

注意端口号5672不要写成15672，这两个端口都有不同的用处，我这里并没有写错，把host改成自己安装RabbitMQ的服务器ip。

### 四、[Hello World!][Hello World] 模型 ###

Hello World模型也称点对点模型，也是最简单的模型，从图中很明显，P为生产者，红框为消息队列，C为消费者，生产者往消息队列发送消息，消费者从消息队列取消息。

![2021062816165862.png][]

接下来我们就以controller当成生产者，调用接口之后把一条文本消息放进消息队列，消费者自动从消息队列取消息。这时候就有疑问了，为什么它能自动取消息进行消费？因为它本质上是一个监听器，一旦有消息产生，那么就会被它监听到。

生产者：

@Autowired
        private RabbitTemplate rabbitTemplate; 
    
        /**
    	 * Hello World模型（点对点消费）
    	 * @return
    	 */
    	@GetMapping("/helloworld")
        public String helloworld() {
        	//第一个参数：发送的队列名称 
    		//第二个参数：发送的信息内容
            rabbitTemplate.convertAndSend("hello","hello world!!");
            return "helloworld---发送消息成功";
        }

消费者HelloConsumer：

package com.chen.consumers;
    
    import org.springframework.amqp.rabbit.annotation.Queue;
    import org.springframework.amqp.rabbit.annotation.RabbitHandler;
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.stereotype.Component;
    
    @Component
    @RabbitListener(queuesToDeclare = @Queue("hello")) // 声明队列名称为hello
    public class HelloConsumer {
    	
    	@RabbitHandler
        public void receivel(String message) {
            System.out.println("消费消息---"+message);
        }
    
    }

结果：

> 消费消息---hello world!!

### 五、[Work queues][] 模型 ###

work queues 模型，创建一个工作队列，用于在多个工作者之间分配耗时的任务。也就是说，多个消费者可以消费同一个消息队列里面的消息，并且是通过轮询的形式，比如说有1,2,3,4这四个消息，按这个模型平均分配的原则，C1和C2分别可以消费两条消息，且都是按顺序的，如：C1消费1和3，C2消费2和4。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQyMTA5NzQ2_size_16_color_FFFFFF_t_70 1][]

生产者：

/**
    	 * work queues模型
    	 * @return
    	 */
    	@GetMapping("/workqueue")
    	public String workqueue() {
    		for (int i = 0; i < 10; i++) {
                rabbitTemplate.convertAndSend("work", "hello world!!"+i);
            }
    		return "workqueue---发送消息成功";
    	}

消费者WorkConsumer：

package com.chen.consumers;
    
    import org.springframework.amqp.rabbit.annotation.Queue;
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.stereotype.Component;
    
    @Component
    public class WorkConsumer {
    
    	@RabbitListener(queuesToDeclare = @Queue("work"))
    	public void receivel(String message) {
    		System.out.println("消费者1  :" + message);
    	}
    
    	@RabbitListener(queuesToDeclare = @Queue("work"))
    	public void receivel2(String message) {
    		System.out.println("消费者2  :" + message);
    	}
    
    }

结果：

> 消费者2  :hello world!!1  
> 消费者1  :hello world!!0  
> 消费者1  :hello world!!2  
> 消费者2  :hello world!!3  
> 消费者1  :hello world!!4  
> 消费者2  :hello world!!5  
> 消费者1  :hello world!!6  
> 消费者2  :hello world!!7  
> 消费者1  :hello world!!8  
> 消费者2  :hello world!!9

###  六、[Publish/Subscribe][Publish_Subscribe] 模型  ###

Publish/Subscribe 模型，也称发布-订阅模型或扇出模型，即多个消费者可以同时接收到消息并进行消费，如下图，X代表交换机，生产者只能向一个交换机发送消息，而交换机再把消息推送到消息队列。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQyMTA5NzQ2_size_16_color_FFFFFF_t_70 2][]

生产者：

/**
    	 * Publish/Subscribe 模型
    	 * @return
    	 */
    	@GetMapping("/fanout")
    	public String fanout() {
    		// 第一个参数代表模型类型，第二个参数代表路由键，第三个参数代表消息内容
    		rabbitTemplate.convertAndSend("fanout", "","hello world!!");
    		return "fanout---发送消息成功";
    	}

消费者FanoutConsumer：

package com.chen.consumers;
    
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.stereotype.Component;
    import org.springframework.amqp.rabbit.annotation.Queue;
    import org.springframework.amqp.rabbit.annotation.QueueBinding;
    import org.springframework.amqp.rabbit.annotation.Exchange;
    
    @Component
    public class FanoutConsumer {
    
    	@RabbitListener(bindings = { @QueueBinding(value = @Queue, // 声明临时队列
    			exchange = @Exchange(value = "fanout", type = "fanout")) }) // value参数代表交换机名称，可以自定义命名，type参数代表模型类型
    	public void receivel(String message) {
    		System.out.println("C1  :" + message);
    	}
    
    	@RabbitListener(bindings = { @QueueBinding(value = @Queue, // 声明临时队列
    			exchange = @Exchange(value = "fanout", type = "fanout")) })
    	public void receivel2(String message) {
    		System.out.println("C2  :" + message);
    	}
    }

结果：

> C1  :hello world!!  
> C2  :hello world!!

### 七、[Routing][] 模型 ###

Routing 模型，也称直连模型，可以对消息指明路由键名称，消息队列绑定路由键名称，这样交换机就能根据路由键名称将消息匹配到相应的消息队列上。如下图，第一个消息队列能够接收路由键名称为“error”的消息，而第二个消息队列能够接收“info”，“error”，“warning”的消息。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQyMTA5NzQ2_size_16_color_FFFFFF_t_70 3][]

生产者：

/**
    	 * Routing 模型
    	 * @return
    	 */
    	@GetMapping("/routing")
    	public String routing() {
    		rabbitTemplate.convertAndSend("route", "info","hello world!!");
    		return "routing---发送消息成功";
    	}

消费者RoutingConsumer：

package com.chen.consumers;
    
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.amqp.rabbit.annotation.Queue;
    import org.springframework.amqp.rabbit.annotation.QueueBinding;
    import org.springframework.amqp.rabbit.annotation.Exchange;
    import org.springframework.stereotype.Component;
    
    @Component
    public class RoutingConsumer {
    
    	@RabbitListener(bindings = { @QueueBinding(value = @Queue, // 声明临时队列
    			exchange = @Exchange(value = "route", type = "direct"), key = { "error" }) }) // key参数代表路由键名称
    	public void receivel(String message) {
    		System.out.println("C1  :" + message);
    	}
    
    	@RabbitListener(bindings = { @QueueBinding(value = @Queue, // 声明临时队列
    			exchange = @Exchange(value = "route", type = "direct"), key = { "error", "info" }) })
    	public void receivel2(String message) {
    		System.out.println("C2  :" + message);
    	}
    
    }

结果：

> C2  :hello world!!

###  八、[Topics][] 模型 ###

Topics 模型，也称主题模型，在Routing原有模型的基础上进行了增强，可以使用通配符的形式来匹配路由键。

.\* 代表可以匹配一个字符

.\# 代表可以匹配多个字符

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQyMTA5NzQ2_size_16_color_FFFFFF_t_70 4][]

生产者：

/**
    	 * Topics 模型
    	 * @return
    	 */
    	@GetMapping("/topic")
    	public String topic() {
    		rabbitTemplate.convertAndSend("topic", "order.save","hello world!!");
    		return "topic---发送消息成功";
    	}

消费者TopicConsumer：

package com.chen.consumers;
    
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.stereotype.Component;
    import org.springframework.amqp.rabbit.annotation.Queue;
    import org.springframework.amqp.rabbit.annotation.QueueBinding;
    import org.springframework.amqp.rabbit.annotation.Exchange;
    
    @Component
    public class TopicConsumer {
    
    	@RabbitListener(bindings = { @QueueBinding(value = @Queue, // 声明临时队列
    			exchange = @Exchange(value = "topic", type = "topic"), key = { "order.*" }) })
    	public void receivel(String message) {
    		System.out.println("C1  :" + message);
    	}
    
    	@RabbitListener(bindings = { @QueueBinding(value = @Queue, // 声明临时队列
    			exchange = @Exchange(value = "topic", type = "topic"), key = { "order.#" }) })
    	public void receivel2(String message) {
    		System.out.println("C2  :" + message);
    	}
    
    }

结果：

> C2  :hello world!!

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQyMTA5NzQ2_size_16_color_FFFFFF_t_70]: /images/20221005/b14611f1b166482abd074e97aa869459.png
[RabbitMQ]: https://blog.csdn.net/qq_42109746/article/details/117821107
[Hello World]: https://www.rabbitmq.com/tutorials/tutorial-one-python.html
[2021062816165862.png]: /images/20221005/4671a758490549ce9a38ea727810495d.png
[Work queues]: https://www.rabbitmq.com/tutorials/tutorial-two-python.html
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQyMTA5NzQ2_size_16_color_FFFFFF_t_70 1]: /images/20221005/15bdc3b2265e48f7bffce2b694fd954b.png
[Publish_Subscribe]: https://www.rabbitmq.com/tutorials/tutorial-three-python.html
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQyMTA5NzQ2_size_16_color_FFFFFF_t_70 2]: /images/20221005/02e8cc4fd4fc46d4ad28eb1964729970.png
[Routing]: https://www.rabbitmq.com/tutorials/tutorial-four-python.html
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQyMTA5NzQ2_size_16_color_FFFFFF_t_70 3]: /images/20221005/930beae9655149be85620db856f8b56b.png
[Topics]: https://www.rabbitmq.com/tutorials/tutorial-five-python.html
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQyMTA5NzQ2_size_16_color_FFFFFF_t_70 4]: /images/20221005/c7ee848a824a4b96b449e88131cdf595.png