RabbitMQ与SpringBoot整合详解_springboot rabbitmq

一、前言

  本文章总体简介，会采用Docker下载安装RabbitMQ，详细介绍RabbitMQ优缺点及其相关技术知识点，SpringBoot作为整合学习的基础架构。看完本片文章，你会了解消息队列RabbitMQ，你会使用消息队列RabbitMQ。

二、消息队列

  当第一次出现在我脑中这个词语时，我会问自己什么时消息队列？我做的项目为什么没有消息队列？我做的项目可以用消息队列么？如果可以用消息队列这要怎么使用消息队列？消息队列能解决项目什么问题？等等问题会出现在我的脑海…，然而我将会在面文章中对自己慢慢摸索和讲解和答疑。

• 什么是消息队列？
    来自百度百科，是这么说得“消息队列”是在消息的传输过程中保存消息的容器，四个字总结就是消息容器。我喜欢用生活中的水缸来形容这个消息队列，消息队列存放消息数据水缸存放水，家里的老婆只负责从水缸取水用水就是消费者，家里老公只负责往水缸里存水就是消息队列的生产者。水缸解决了家庭生产分配问题，也是家庭解耦分工的体现，而消息队列则是应用程序解耦的过程，为的是解决应用程序执行效率。
• 项目没有消息队列？
    也许很多伙伴刚开始毕业工作，接触到项目并没有采用相关技术，为什么没有用呢？一、项目不需要消息队列，项目业务中确实用不上消息队列，消息队列解决的问题就是应用程序解耦，什么程序不需要解耦呢？系统产生的数据量不大，业务比较单一，系统刚产生数据就被及时处理了，这样的应用程序是不需要消息队列的。就等同家里厨房接了自来水管，需要用多少水就用多少水，用完的水会被及时排放不存存储的概念。二、团队中技术处于劣势，并没有人知道者概念或者也不愿学习实践。
• 项目可以用消息队列么？
    项目可不可以用消息队列，我们得知道消息队列的作用和功能，查阅相关资料得到常见三大功能如下几点：
  1.应用解耦
    如上图就一个及其简版的商城系统，一个用户下单的操作设计到诸多复杂业务的处理，涉及到财务、积分、物流等等，各模块之间都被耦合了，相互受到影响，而引入消息队列可以解决这一问题。  各个模块之间相互独立却也有微弱的相关性，订单只需处理订单，处理之后的相关业务消息存入消息队列，引入消息队列后各个模块只需要负责自己订阅的消息处理相关业务即可，大大的对应用程序解耦，从而提高了每个模块的响应速度和程序效率。

2.异步消息
  应用程序的解耦也是运用的消息队列异步消息的功能，系统在没有引入消息队列时应用程序的数据流向会是用户请求（浏览器）->业务处理（订单，财务，积分等等）->数据存储(操作数据库)如下图，我们会发现大量不同的业务都会之间对数据库操作，整个应用程序的性能将受到数据量的影响，数据量过大时还会直接导致系统宕机。

  而加入了消息队列的系统是怎么样的呢？每个生产者业务都会把消息放入消息队列，每个消费者都会去消息队列取自己订阅的消息。保证了上游系统能一直正常运行，不会应为上游生产者系统产生大量业务数据而导致系统卡顿，上游大量的业务数据将会被存入消息队列，等待下游系统来消费，下游系统每次从消息队列取多少数据由下游系统性能决定，能处理2000就从消息队列取2000处理，这样减轻了数据库系统的操作，也保证了上游系统正常的业务。如下图：
3.流量削锋
  就目前技术而言，流量削锋的技术很多，分为两大类技术有损流量削峰和无损流量削峰。前者偏重与技术层面实现，后者侧重于物理和运行环境实现。
  有损流量削峰
  技术方案有消息队列（RabbitMQ、Kafka、ActiveMQ、ZeroMQ、MetaMQ、RocketMQ等）、答题（验证码、图片绘画，图片识别等）、过滤（技术层面层层过滤数据，从用户发起请求到存入数据库中所有环节对数据处理校验和筛选最终存入有效数据）。
  无损流量削峰
  负载均衡（可以用nginx反向代理等负载技术实现流量分发）等等技术。

三、RabbitMQ

  市场上的MQ种类繁多，常见消息队列有RabbitMQ，RocketMQ，ActiveMQ，ZeroMQ，Kafka等，各有优缺点自行查阅资料了解相对应的优缺点。主要学习RabbitMQ，其他相关知识自行了解。

• 1、RabbitMQ介绍
    详细介绍参照本篇博客文章介绍。这篇博客文章讲的比较细致全面。
• 2、docker搭建RabbitMQ
    docker使用和相关语法自行学习，下面直接使用docker搭建RabbitMQ，在linux系统或虚拟机（保证网络联通）下执行下面命令
  搜索镜像docker search rabbitmq
  
    上图中第一条就是我们需要下载的RabbitMQ镜像。
  下载RabbitMQ镜像docker pull rabbitmq
    等待下载完成，执行docker images查看镜像是否下载成功，如下图则下载成功。
  
  启动运行容器命令 docker run -dit --name Myrabbitmq -e RABBITMQ_DEFAULT_USER=admin -e RABBITMQ_DEFAULT_PASS=admin -p 15672:15672 -p 5672:5672 05a22b32da70
    RABBITMQ_DEFAULT_USER=admin 登陆RabbitMQ账号为admin
    RABBITMQ_DEFAULT_PASS=admin 登陆RabbitMQ密码为admin
    -p 15672:15672 Web 管理界面端口映射 -p 5672:5672 代码连接和客户端使用端口
    05a22b32da70 RabbitMQ镜像的ID
    --name Myrabbitmq 启动的RabbitMQ容器名称为RabbitMQ
     -it 以交互方式启动容器
     -d 在后台启动它们
    访问启动RabbitMQ容器，ip:15672出现登陆页面表示RabbitMQ容器登陆成功，输入账号admin密码admin进行登陆，登陆成功如下图：
  
    登陆成功后web页面顶部导航包括Overview(概述)、Connections(连接)、Channels(通道)、Exchanges(交换器)、Queues(队列)、Admin(用户管理)。如不能访问成功请尝试进入容器docker exec -it {rabbitmq容器名称或者id} /bin/bash，执行rabbitmq-plugins enable rabbitmq_management命令开启RabbitMQ的Web管理插件，重启rabbitMQ容器docker restart {rabbitmq容器名称或者id}。

四、SpringBoot整合RabbitMQ

  在上面所有教程中已经知道了什么是消息队列，RabbitMQRabbitMQ的核心概念（生产者（Producer）：发送消息的应用；消费者（Consumer）：接收消息的应用；队列（Queue）：存储消息的缓存；消息（Message）：由生产者通过RabbitMQ发送给消费者的信息；连接（Connection）：连接RabbitMQ和应用服务器的TCP连接；通道（Channel）：连接里的一个虚拟通道。当你通过消息队列发送或者接收消息时，这个操作都是通过通道进行的；交换机（Exchange）：交换机负责从生产者那里接收消息，并根据交换类型分发到对应的消息列队里。要实现消息的接收，一个队列必须到绑定一个交换机；绑定（Binding）：绑定是队列和交换机的一个关联连接；路由键（Routing Key）：路由键是供交换机查看并根据键来决定如何分发消息到列队的一个键。路由键可以说是消息的目的地址）。RabbitMQ五种消息发送模式（简单队列、工作队列、发布/订阅模式(Publish/Subcribe)、路由模式（Routing）、主题模式（Topic））。RabbitMQ四种交换机（Fanout exchange，Direct exchange、Topic exchange、Headers exchange）等相关知识，与RabbitMQ相关详细知识请自行查阅相关资料学习。推荐学习博客地址（https://blog.csdn.net/github_37130188/article/details/115289346）。

• 1、引入RabbitMQ
  在springBoot项目pom文件中引入依赖

		<dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
        </dependency>
1
2
3
4

  在application.properties配置文件中配置rabbitMQ相关配置信息

# rabbitmq配置信息
# rabbitmq启动的ip地址
spring.rabbitmq.host=192.168.147.135
# 端口
spring.rabbitmq.port=5672
# 用户名
spring.rabbitmq.username=admin
# 密码
spring.rabbitmq.password=admin
# 配置虚拟机
spring.rabbitmq.virtual-host=/
# 消息开启手动确认
spring.rabbitmq.listener.direct.acknowledge-mode=manual
1
2
3
4
5
6
7
8
9
10
11
12
13

以上配置整合已经完成了，接下来我们看看如何在SpringBoot中rabbitmq使用每种情况。

• 2、Direct定向/直连类型交换机的使用
  配置交换机类型为Direct的RabbitMQ的相关配置DirectRabbitmqConfig.java

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.boot.SpringBootConfiguration;
import org.springframework.context.annotation.Bean;

/**
 * @author 260497
 */
@SpringBootConfiguration
public class DirectRabbitmqConfig {


    /**
     * 队列
     * @return directQueue队列名称 true 持久化
     */
    @Bean
    public Queue directQueueTest(){
        return new Queue("directQueue",true);
    }

    /**
     * 新建Direct类型交换机
     * @return directExchangeTest交换机名称  true 持久化 false 不自动删除
     */
    @Bean
    public DirectExchange directExchangeTest(){
        return new DirectExchange("directExchangeTest",true,false);
    }

    /**
     * 将directQueueTest队列绑定到DirectExchange交换机
     * @return
     */
    @Bean
    public Binding bindingDirect(){
        /**
         * directQueueTest() Queue队列
         * directExchangeTest()  DirectExchange交换机
         * directRoutingKey  路由键
         */
        return BindingBuilder.bind(directQueueTest()).to(directExchangeTest()).with("directRoutingKey");
    }

    /**
     * 没有绑定任何队列的交换机Exchange
     * @return
     */
    @Bean
    DirectExchange lonelyDirectExchange() {
        return new DirectExchange("lonelyDirectExchange");
    }

}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58

书写接口生产数据调用生产者

import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;


/**
 * @author 260497
 */
@RestController
public class DirectRabbitmqController {
    @Autowired
    RabbitTemplate rabbitTemplate;

    /**
     * Direct测试
     * @return
     */
    @GetMapping("/sendDirectMessage")
    public String sendDirectMessage() {
        rabbitTemplate.convertAndSend("directExchangeTest", "directRoutingKey", "你好！大帅哥，我是交换机Direct类型所产生的消息");
        return "ok";
    }

}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

书写Direct类型交换机的消费者

import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.util.Map;

/**
 * @email： wf520200@163.com
 * @date: 2021/11/16 15:40
 * 功能：DirectExchange交换机类型生产者和消费者测试
 * 描述：
 * @author： 吴帆
 **/
@Component
@RabbitListener(queues = "directQueue") //
public class DirectReceiver {


    @RabbitHandler
    public void process(String msg){
        System.out.println("消费者接收到消息:"+msg);
    }

}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

启动SpringBoot项目，postman或浏览器直接访问ip:端口/sendDirectMessage地址，返回ok

查看控制台如下图，消费者消费交换机类型为derect直连类型消费消息成功

上面代码可以用下面流图表示，队列和交换机直通过路由键建立关系，不像topic模式可进行匹配。

• 3、Topic Exchange 主题交换机
  配置Topic类型的交换机并绑定相应的队列

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * @email： wf520200@163.com
 * @date: 2021/11/16 16:32
 * 功能：
 * 描述：
 * @author： 吴帆
 **/
@Configuration
public class TopicRabbitmqConfig {

    @Bean
    public Queue queue1(){
        //配置队列名称为topic.man的队列
        return new Queue("queue1");
    }


    @Bean
    public Queue queue2(){
        //配置队列名称为topic.woman的队列
        return new Queue("queue2");
    }

    /**
     * 新建TopicExchange交换机 起名topicExchange
     * @return
     */
    @Bean
    public TopicExchange topicExchange(){
        return new TopicExchange("topicExchange");
    }


    /**
     * 绑定一个manQueue()队列到topicExchange交换机 路由键routingKey=topic.man
     * @return
     */
    @Bean
    public Binding bindingManQueueToTopicExchange(){
        return BindingBuilder.bind(queue1()).to(topicExchange()).with("topic.man");
    }

    /**
     * womanQueue()topicExchange 路由键routingKey=topic.   这样只要是消息携带的路由键是以topic.开头,都会分发到该队列womanQueue
     * @return
     */
    @Bean
    public Binding bindingWoManQueueToTopicExchange(){
        // 这样只要是消息携带的路由键是以topic.开头,都会分发到该队列
        return BindingBuilder.bind(queue2()).to(topicExchange()).with("topic.#");
    }
}


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61

新建生产者1—指定路由键为topic.man（其实等同指定队列）

import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

/**
 * @email： wf520200@163.com
 * @date: 2021/11/16 17:04
 * 功能：
 * 描述：
 * @author： 吴帆
 **/
@RestController
public class TopicManRabbitmqController {

    @Autowired
    RabbitTemplate rabbitTemplate;

    /**
     * Direct测试
     * @return
     */
    @GetMapping("/sendTopicManMessage")
    public String sendTopicManMessage() {
        rabbitTemplate.convertAndSend("topicExchange", "topic.man", "你好！大帅哥，我是交换机Topic类型所产生的消息");
        return "ok";
    }

}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

新建生产者2—路由键为topic.woman

import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.Resource;

/**
 * @email： wf520200@163.com
 * @date: 2021/11/16 18:36
 * 功能：
 * 描述：
 * @author： 吴帆
 **/
@RestController
public class TopicWoManRabbitmqController {

    @Resource
    private RabbitTemplate rabbitTemplate;

    /**
     * Direct测试
     * @return
     */
    @GetMapping("/sendTopicWoManManMessage")
    public String sendTopicWoManManMessage() {
        rabbitTemplate.convertAndSend("topicExchange", "topic.woman", "你好！大帅哥，我是交换机Topic类型所产生的消息");
        return "ok";
    }

}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

新建消费者1—TopicManReceiver —queues = “queue1”

import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

/**
 * @email： wf520200@163.com
 * @date: 2021/11/16 17:09
 * 功能：
 * 描述：
 * @author： 吴帆
 **/

@Component
@RabbitListener(queues = "queue1")
public class TopicManReceiver {

    @RabbitHandler
    public void process(String msg){
        System.out.println("消费者接收到消息Man--Man--Man:"+msg);
    }

}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

新建消费者2–TopicWoManReceiver–queues = “queue2”

import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

/**
 * @email： wf520200@163.com
 * @date: 2021/11/16 17:26
 * 功能：
 * 描述：
 * @author： 吴帆
 **/
@Component
@RabbitListener(queues = "queue2")
public class TopicWoManReceiver {

    @RabbitHandler
    public void process(String msg){
        System.out.println("消费者接收到消息WoMan--WoMan--WoMan:"+msg);
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

  上面的代码就是如下图效果一致，将队列queue1绑定到交换机TopicExchange上路由键为topic.man、将队列queue绑定到交换机TopicExchange上路由键为topic.#（#代表任意一个或多个单词）。

  下图描述的就是在RabbitMQ中交换机类型为TopicExchange的整体流程图。生产者生产信息会携带交换机类型、路由键、消息三个信息去配置完的RabbitMQ根据对应关系把消息塞到队列queue1或queue2上。消费者监听自己指定的队列，每个队列绑定时都有自己的路由规则，路由规则符合自己的信息将会被该队列接受。

• 4、Fanout Exchang 扇型交换机（广播模式）
  配置fanout类型交换机绑定三个队列queueA、queueB、queueC，代码如下：

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.FanoutExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * @email： wf520200@163.com
 * @date: 2021/11/17 8:44
 * 功能：
 * 描述：
 * @author： 吴帆
 **/
@Configuration
public class FanoutRabbitConfig {

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

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

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

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

    /**
     * 名称为queueA的队列绑定到fanout类型的交换机
     * @return Binding
     */
    @Bean
    public Binding queueABingToExchange(){
        return BindingBuilder.bind(queueA()).to(fanoutExchange());
    }

    /**
     * 名称为queueB的队列绑定到fanout类型的交换机
     * @return Binding
     */
    @Bean
    public Binding queueBBingToExchange(){
        return BindingBuilder.bind(queueB()).to(fanoutExchange());
    }

    /**
     * 名称为queueC的队列绑定到fanout类型的交换机
     * @return Binding
     */
    @Bean
    public Binding queueCBingToExchange(){
        return BindingBuilder.bind(queueC()).to(fanoutExchange());
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66

新建一个消息发送，指定交换机类型为fanoutExchange，路由键为空。

import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

/**
 * @email： wf520200@163.com
 * @date: 2021/11/17 9:04
 * 功能：模拟生产者往队列queueA中推送消息
 * 描述：
 * @author： 吴帆
 **/
@RestController
public class FanoutQueueRabbitController {

    @Autowired
    private RabbitTemplate rabbitTemplate;


    @GetMapping("queue")
    public String queue(){
        rabbitTemplate.convertAndSend("fanoutExchange", null, "测试fanout类型的交换机消息发送");
        return "ok";
    }

}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

新建三个消费者监听queueA,queueA,queueC。

import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

/**
 * @email： wf520200@163.com
 * @date: 2021/11/17 9:22
 * 功能：队列queueA监听  消费者端
 * 描述：
 * @author： 吴帆
 **/
@Component
@RabbitListener(queues= "queueA")
public class QueueReceiverA {

    @RabbitHandler
    public void process(String msg){
        System.out.println("QueueReceiverA消费者接收到的消息："+msg);
    }

}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

/**
 * @email： wf520200@163.com
 * @date: 2021/11/17 9:22
 * 功能：
 * 描述：
 * @author： 吴帆
 **/
@Component
@RabbitListener(queues="queueB")
public class QueueReceiverB {

    @RabbitHandler
    public void process(String msg){
        System.out.println("QueueReceiverB消费者接收到的消息："+msg);
    }

}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

/**
 * @email： wf520200@163.com
 * @date: 2021/11/17 9:22
 * 功能：
 * 描述：
 * @author： 吴帆
 **/
@Component
@RabbitListener(queues="queueC")
public class QueueReceiverC {

    @RabbitHandler
    public void process(String msg){
        System.out.println("QueueReceiverC消费者接收到的消息："+msg);
    }

}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

重启项目访问ip:端口/queue地址，postman访问成功如下图：

查看控制台打印信息如下图：

上图得知，当交换机为fanout广播模式时，只要是与FanoutExchange交换机绑定的队列，路由键相同则所有队列都会收到消息。

RabbitMQ整合SpringBoot结束了，gitee代码地址https://gitee.com/wufanlove/mq.git，以上技术仅供学习参考，如有疑问请及时联系。

想要了解更多作者动态和最新知识、请关注扫码关注公众号：