Spring Boot(十)：整合RabbitMQ消息中间件

野性酷女 2023-06-11 03:21 10阅读 0赞

### RabbitMQ背景 ###

消息队列中间件是分布式系统中重要的组件，主要解决应用耦合，异步消息，流量削锋等问题。实现高性能，高可用，可伸缩和最终一致性。

###### 消息中间件的分类 ######

①、ActiveMQ：是Apache软件基金会所研发的开放源代码消息中间件；由于ActiveMQ是一个纯Java程序，因此只需要操作系统支持Java虚拟机，ActiveMQ便可执行。  
②、RabbitMQ：是实现了高级消息队列协议(AMQP)的开源消息代理软件(亦称面向消息的中间件)。RabbitMQ服务器是用Erlang语言编写的，而集群和故障转移是构建在开放电信平台框架上的。所有主要的编程语言均有与代理接口通讯的客户端库。  
③、Kafka：是由Apache软件基金会开发的一个开源流处理平台，由Scala和Java编写。Kafka是一种高吞吐量的分布式发布订阅消息系统，它可以处理消费者在网站中的所有动作流数据。  
④、RocketMQ：是阿里巴巴集团自主研发的专业消息中间件。

> 上面每个消息中间件的解释大部分来自百度百科。

###### RabbitMQ的特点 ######

1、可靠性：使用一些机制来保证可靠性，如持久化、传输确认、发布确认。  
2、消息集群：多个RabbitMQ服务器可以组成一个集群，形成一个逻辑Broker(缓存代理)。  
3、高可用：队列可以在集群中的机器上进行镜像，使得在部分节点出问题的情况下队列仍然可用。  
4、多种协议：RabbitMQ支持多种消息队列协议，比如STOMP、MQTT等。  
5、多语言客户端：RabbitMQ几乎支持所有常用语言，比如Java、.NET、Ruby等。  
6、管理界面：RabbitMQ提供了一个易用的用户界面，使得用户可以监控和管理消息Broker的许多方面。  
7、插件机制：RabbitMQ提供了许多插件，来从多方面进行扩展，也可以编写自己的插件。  
8、跟踪机制：如果消息异常，RabbitMQ提供了消息跟踪机制。

### RabbitMQ架构图的介绍 ###

![图片引用于百度][format_png]  
ps: 图片来源于网络

根据这张架构图，可以对RabbitMQ中各个组件进行介绍：

**RabbitMQ Server**：也叫Broker Server，它是一种传输服务。 它的角色就是维护一条从Producer到Consumer的路线，保证数据能够按照指定的方式进行传输。

**Producer**：消息生产者，消息生产者连接RabbitMQ服务器然后将消息投递到Exchange。

**Consumer**：消息消费者，消息消费者订阅队列，RabbitMQ将Queue中的消息发送到消息消费者。

**Exchange**：生产者将消息发送到Exchange(交换器)，由Exchange将消息路由到一个或多个Queue中(或者丢弃)。Exchange并不存储消息。RabbitMQ中的Exchange有direct、fanout、topic、headers四种类型，每种类型对应不同的路由规则。

**Queue**：(队列)是RabbitMQ的内部对象，用于存储消息。消息消费者就是通过订阅队列来获取消息的，RabbitMQ中的消息都只能存储在Queue中，生产者生产消息并最终投递到Queue中，消费者可以从Queue中获取消息并消费。多个消费者可以订阅同一个Queue，这时Queue中的消息会被平均分摊给多个消费者进行处理，而不是每个消费者都收到所有的消息并处理。

**RoutingKey**：生产者在将消息发送给Exchange的时候，一般会指定一个Routing key，来指定这个消息的路由规则，而这个Routing key需要与Exchange Type及Binding key联合使用才能最终生效。在Exchange Type与Binding key固定的情况下(在正常使用时一般这些内容都是固定配置好的)，我们的生产者就可以在发送消息给Exchange时，通过指定Routing key来决定消息流向哪里。RabbitMQ为Routing key设定的长度限制为255bytes。

**Connection**：(连接)：Producer和Consumer都是通过TCP连接到RabbitMQ Server的。程序的起始处就是建立这个TCP连接。

**Channels**：(信道)：它建立在上述的TCP连接中。数据流动都是在Channel中进行的。也就是说，一般情况是程序起始建立TCP连接，第二步就是建立这个Channel。

**VirtualHost**：权限控制的基本单位，一个VirtualHost里面有若干Exchange和MessageQueue，以及指定被哪些user使用。

### RabbitMQ的安装 ###

至于RabbitMQ的安装，这里就不做说明了，网上也有很多安装教程，各位读者可以根据相关资料进行安装。博主这里使用的是本地Windows上安装的RabbitMQ。安装完成，通过命令安装管理界面：

rabbitmq‐plugins enable rabbitmq_management

然后输入地址：http://127.0.0.1:15672，就能看到如下界面。  
![format_png 1][]  
用户名和密码默认为: guest

### 三种模式的讲解与代码实现 ###

RabbitMQ有三种模式：直接模式(Direct)、分列模式(Fanout)、主题模式(Topic)。

###### 直接模式(Direct) ######

将消息发给唯一一个节点时使用这种模式。任何发送到Direct Exchange的消息都会被转发到RouteKey中指定的Queue。

**特点**：  
1、可以使用RabbitMQ自带的Exchange " " (该Exchange的名字为空字符串);  
2、这种模式下不需要将Exchange进行任何绑定操作;  
3、消息传递时需要一个"RouteKey"，可以理解为要发送到的队列名;  
4、如果vhost中不存在RouteKey中指定的队列名，则该消息会被抛弃。

**代码实现**：  
这里使用的是SpringBoot,需要引入RabbitMQ的pom依赖。

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

然后增加RabbitMQ的配置。

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

配置完成，先创建消息的生产者，这里使用SpringBoot中提供的测试类。

@RunWith(SpringRunner.class)
    @SpringBootTest
    public class MonitorApplicationTests { 
    
        @Autowired
        private RabbitTemplate rabbitTemplate;
    
        @Test
        public void direct(){ 
            rabbitTemplate.convertAndSend("qfcwx","direct模式");
        }
    }

注意，这里一定要先通过RabbitMQ的管控台创建好名为"qfcwx"的队列。不然启动会报错的。  
![format_png 2][]  
然后启动测试类，再通过管控台查看当前消息队列中是否存在消息。  
![format_png 3][]  
可以看到，此时队列中确实是有一条消息等待消费。

然后创建消息的消费者。

@Component
    @RabbitListener(queues = "qfcwx")
    public class DirectListener { 
    
        @RabbitHandler
        private void direct(String message){ 
            System.out.println("direct模式下收到的消息:" + message);
        }
    }

@RabbitListener：标注在类上面的时候，需要配合@RabbitHandler注解一起使用，queues指明监听哪个队列。

运行启动类，通过控制台看输出结果。  
![format_png 4][]  
查看管控台发现，队列中的消息也被消费了。

###### 分列模式(Fanout) ######

将消息一次发给多个队列时，需要使用这种模式。任何发送到Fanout Exchange的消息都会被转发到与该Exchange绑定的所有Queue上。

**特点**：  
1、这种模式不需要RouteKey。  
2、这种模式需要提前将Exchange与Queue进行绑定，一个Exchange可以绑定多个Queue，一个Queue可以同多个Exchange进行绑定。  
3、如果接受到消息的Exchange没有与任何Queue绑定，则消息会被丢弃。

**代码实现**：  
这次首先通过管控台创建Exchange与Queue。这里创建名为"qfcwx\_exchange"的交换器，然后创建名为"qfcwx\_a"、"qfcwx\_b"的两个队列。然后将两个队列绑定到交换器上。

![format_png 5][]

接着通过代码编写生产者：

@Test
        public void fanout(){ 
            rabbitTemplate.convertAndSend("qfcwx_exchange","","fanout模式");
        }

然后创建消费者：

@Component
    public class FanoutListener { 
    
        @RabbitListener(queues = "qfcwx_a")
        public void fanoutToQueueOne(String message) { 
            System.out.println("队列[qfcwx_a],收到了交换机[qfcwx_exchange]的消息:" + message);
        }
    
        @RabbitListener(queues = "qfcwx_b")
        public void fanoutToQueueTwo(String message) { 
            System.out.println("队列[qfcwx_b],收到了交换机[qfcwx_exchange]的消息:" + message);
        }
    }

直接启动测试类，通过控制台查看结果。  
![format_png 6][]

###### 主题模式(Topic) ######

任何发送到Topic Exchange的消息都会被转发到所有关心RouteKey中指定话题的Queue上,交换器是一个名称与队列绑定的列表。当消息发布到交换器时，实际上是由你所连接的信道，将消息路由键同交换器上绑定的列表进行比较，最后路由消息。

看到这里，相信很多读者会一脸懵逼，下面用表格展示一下。

<table> 
 <thead> 
  <tr> 
   <th align="left">routing key</th> 
   <th align="left">binding key</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td align="left">qfcwx.mobile</td> 
   <td align="left">qfcwx.#</td> 
  </tr> 
  <tr> 
   <td align="left">test.mail</td> 
   <td align="left">test.#</td> 
  </tr> 
 </tbody> 
</table>

表格只是大致说明这是一种匹配的规则。符号 \# 匹配一个或多个词，符号 \* 匹配不多不少一个词。这里也许博主说的比较模糊了。等会通过代码来讲解吧。

**特点**：  
1、每个队列都有其关心的主题，所有的消息都带有一个"标题"(RouteKey)，Exchange会将消息转发到所有关注主题能与RouteKey模糊匹配的队列。  
2、这种模式需要RouteKey，也需要绑定Exchange与Queue。  
3、在进行绑定时，要提供一个该队列关心的主题，如"\#.log.\#“表示该队列关心所有涉及log的消息。  
4、”\#“表示0个或若干个关键字，”"表示一个关键字。如"log."能与"log.warn"匹配，无法与"log.warn.timeout"匹配；但是"log.\#"能与上述两者匹配。  
5、如果Exchange没有发现能够与RouteKey匹配的Queue，则会抛弃此消息。

**代码实现**：

还是先操作管控台，创建一个交换机，命名为log\_exchange。  
![format_png 7][]  
然后，分别创建交互机下队列与routing的匹配规则。  
![format_png 8][]  
![format_png 9][]  
![format_png 10][]

接着创建生产者：

@Test
        public void topic() { 
            rabbitTemplate.convertAndSend("log_exchange", "log.error", "topic模式");
            rabbitTemplate.convertAndSend("log_exchange", "debug.log", "topic模式");
            rabbitTemplate.convertAndSend("log_exchange", "info.log", "topic模式");
        }

这里指定了交换机的名称和routingkey。

创建消费者。

@Component
    public class TopicListener { 
    
        @RabbitListener(queues = "qfcwx_a")
        public void topicToQueueOne(String message) { 
            System.out.println("队列[qfcwx_a],收到了交换机[log_exchange]的消息:" + message);
        }
    
        @RabbitListener(queues = "qfcwx_b")
        public void topicToQueueTwo(String message) { 
            System.out.println("队列[qfcwx_b],收到了交换机[log_exchange]的消息:" + message);
        }
    
        @RabbitListener(queues = "qfcwx_c")
        public void topicToQueueThree(String message) { 
            System.out.println("队列[qfcwx_c],收到了交换机[log_exchange]的消息:" + message);
        }
    }

启动程序，看控制太打印的结果。  
![format_png 11][]  
因为上面生产者的routingkey与我们先前在管控台配置的"qfcwx\_c"队列的routingkey不匹配，所以，队列"qfcwx\_c"并没有收到消息。这就是主题模式的使用方法。

到这里，关于RabbitMQ与SpringBoot整合的内容就讲完了。全文其实大部分都在讲解RabbitMQ的知识，特别是三种模式，一、二种模式好理解，第三种还需要结合代码来理解。如果文中存在什么问题或者错误的地方，也希望各位读者提出。

> 人都有以第一印象定好坏的习惯，以为一个人好时，就会爱屋及乌，以为一个人不好时，就会全盘否认。

[format_png]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly90aW1nc2EuYmFpZHUuY29tL3RpbWc_aW1hZ2UmcXVhbGl0eT04MCZzaXplPWI5OTk5XzEwMDAwJnNlYz0xNTcyMDg5NDM5MDc5JmRpPTkzMGE5OTgyOWE3OGQxNTViODE2YTk1Nzg3ZjU2YTM1JmltZ3R5cGU9anBnJnNyYz1odHRwJTNBJTJGJTJGaW1nNC5pbWd0bi5iZGltZy5jb20lMkZpdCUyRnUlM0QyMjYxNDY5OTE0JTJDNDEyMDczOTA3NCUyNmZtJTNEMjE0JTI2Z3AlM0QwLmpwZw?x-oss-process=image/format,png
[format_png 1]: /images/20230531/537a35efa9914fc4bde7615e9055960b.png
[format_png 2]: /images/20230531/98206ac6b9d9471b9771f4bb37e6d8c7.png
[format_png 3]: /images/20230531/4b915c63542d40ecbc27cd3b0f132450.png
[format_png 4]: /images/20230531/45396941571743bd9dcfc13fbb33dece.png
[format_png 5]: /images/20230531/58a55f2d40984bee88ccf4a2d27fd20d.png
[format_png 6]: /images/20230531/97dcbd763ca94b188edebac1fac1e3d3.png
[format_png 7]: /images/20230531/32c0a2e7b49e4a70829ff333a6db4712.png
[format_png 8]: /images/20230531/d6b905a0335f45f6ab14cbfd9c711b25.png
[format_png 9]: /images/20230531/99cf48ac611f42ad888ed2057daf1602.png
[format_png 10]: /images/20230531/4feee7d0992644e489d373f997ee665f.png
[format_png 11]: /images/20230531/68ac1dd7eea7429bac89c927a2d12f6f.png