SpringBoot项目开发(十一)：RabbitMQ消息队列

「爱情、让人受尽委屈。」 2022-05-17 08:46 232阅读 0赞

###### 1.为什么使用消息队列？ ######

*  **系统解耦**：在分布式环境中，可能有A、B、C、D…等多个系统，A直接调用B提供的服务接口，当A系统修改时，可能需要B也跟着修改，这样耦合性比较强。MQ可以解耦这样的场景
 *  **流量销峰**：在电商抢购活动中，商品可能几秒钟就被抢购完，但是系统或数据库无法在几秒内处理所有请求，大量请求或许会让数据库直接崩溃，把请求压到队列中，消费者有序的处理队列中的请求
 *  **稳定性**：队列有确认机制，当消费者进程异常时，也就是该消息还没有被确认消费，那么其他消费进程可再次处理该消息

###### 2.RabbitMQ 特点 ######

*  可靠性：RabbitMQ 使用一些机制来保证可靠性，如持久化、传输确认、发布确认
 *  消息集群：多个 RabbitMQ 服务器可以组成一个集群，形成一个逻辑 Broker
 *  高可用：队列可以在集群中的机器上进行镜像，使得在部分节点出问题的情况下队列仍然可用。
 *  多语言客户端：RabbitMQ 几乎支持所有常用语言，比如 Java、.NET、Ruby 等
 *  管理界面：RabbitMQ 提供了一个易用的用户界面，使得用户可以监控和管理消息 Broker 的许多方面

消息默认存储在磁盘上，磁盘的处理能力是有限，对于大流量、高并发场景，RabbitMQ有内存模式，内存可以高速处理消息，但是需要磁盘模式配合，防止消息丢失，关于RabbitMQ集群方面的，请参考 [RabbitMQ + Haproxy 高可用队列集群][RabbitMQ _ Haproxy]

###### 3.RabbitMQ 中的概念模型 ######

生产者（producer）创建消息，然后发布到队列（queue）中，消费者（consumer）订阅某个队列，最后将消息发送到监听的消费者，可以有多个消费者，消息会被平均分摊给多个消费者进行处理  
![这里写图片描述][70]

RabbitMQ有 4 个重要概念，分别为：虚拟主机，交换机，队列，和绑定。

*  Broker：rabbitmq队列服务器，里面可以有多个 虚拟主机 Virtual Host
 *  Virtual Host：为什么需要多个虚拟主机呢？很简单，RabbitMQ当中，用户只能在虚拟主机的粒度进行权限控制。 因此，如果需要禁止A组访问B组的交换机/队列/绑定，必须为A和B分别创建一个虚拟主机。每一个RabbitMQ服务器都有一个默认的虚拟主机“/”。
 *  Exchange：交换机用于转发消息，但是它不会做存储 ，如果没有 Queue bind 到 Exchange 的话，它会直接丢弃掉 Producer 发送过来的消息。  
    这里有一个比较重要的概念：路由键 。消息到交换机的时候，交互机会转发到对应的队列中，那么究竟转发到哪个队列，就要根据该路由键。
 *  Binding：交换机需要和绑定队列，这其中如上图所示，是多对多的关系。
 *  Quene：队列
 *  Connection：连接，建立一个tcp连接，使用多路复用方式提升性能，节约资源
 *  Channel：信道，一个连接可以多路复用打开多个信道，获取队列中的消息

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly96aHV5dS5ibG9nLmNzZG4ubmV0_size_16_color_FFFFFF_t_70]

RabbitMQ常用的交换机Exchange Type有四种：direct、fanout、topic、headers。

direct:点对点模式,一对一关系，把消息投递到那些binding key与routing key完全匹配的队列中。
      fanout:广播模式,一对多关系，转发消息到所有绑定队列。
      topic:发布订阅模式,一对多关系，按规则转发消息,将消息路由到binding key与routing key模式匹配的队列中。
      headers:设置header attribute参数类型的交换机，很少用

###### 4.SrpingBoot集成 RabbitMQ ######

1).添加引用依赖

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

2).添加配置，rabbitmq的安装地址、端口以及账户信息，没有rabbitmq环境的可以参考 [RabbitMQ安装教程][RabbitMQ]

spring.rabbitmq.host=192.168.68.128
    spring.rabbitmq.port=5672
    spring.rabbitmq.username=admin
    spring.rabbitmq.password=admin
    spring.rabbitmq.virtual-host=zy_host
    
    #下面的参数待测试
    #prefetch每次一次性从broker里面取的待消费的消息的个数
    spring.rabbitmq.listener.simple.prefetch=5
    #创建消费者的个数（并发消费）
    spring.rabbitmq.listener.simple.concurrency=5
    spring.rabbitmq.listener.simple.max-concurrency=10

3).队列配置

@Configuration
    public class RabbitConfig {
        //定义一个名称为hello的队列
        @Bean
        public Queue HelloQueue(){
            return new Queue("hello");
        }
    }

4).生产者，AmqpTemplate是springboot 提供的默认实现

@Component
    public class HelloSender {
    
        @Autowired
        private AmqpTemplate rabbitTemplate;
    
        public void send(String queueName, String message){
            rabbitTemplate.convertAndSend(queueName,message);
        }
    }

5).消费者，监听队列，处理消息

@Component
    @RabbitListener(queues = "hello")
    public class HelloReceiver {
    
        @RabbitHandler
        public void process(String context){
            System.out.println("Receiver : " + context);
        }
    }

6).测试，生产者与消费者的queue name必须一致，不然无法接收

@RunWith(SpringRunner.class)
    @SpringBootTest
    public class RabbitMQTest {
        @Autowired
        private HelloSender helloSender;
    
        @Test
        public void sendMessage(){
            helloSender.send("hello","你好，我要向队列发送消息1...");
        }
    }

运行测试方法时，可以在rabbitmq管理后台看到，hello队列会自动创建

![这里写图片描述][70 1]

![这里写图片描述][70 2]

还可以测试一对多，多对多关系，看队列是怎么消费的  
多个生成者

@Test
    public void sendManyMessage(){
    	for (int i=0;i<5;i++){
    		helloSender.send("hello","你好，我要向队列发送消息1..."+i);
    		helloSender.send("hello","你好，我要向队列发送消息2..."+i);
    	}
    }

多个消费者，监听 hello 队列  
![这里写图片描述][70 3]

运行，查看结果：多个消费者均匀的消费多个消息  
![这里写图片描述][70 4]

本篇是抛砖引玉，介绍了一些概念和简单使用，它的高级使用，请参考下面：  
[centos7下RabbitMQ安装教程][RabbitMQ]  
[RabbitMQ + Haproxy 高可用队列集群][RabbitMQ _ Haproxy]  
[RabbitMQ 延迟队列的实现][RabbitMQ 1]

下篇 [发送邮件][Link 1]

[RabbitMQ _ Haproxy]: https://blog.csdn.net/zhuyu19911016520/article/details/80206202
[70]: /images/20220517/887ea27ff20d4103997316584be46eab.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly96aHV5dS5ibG9nLmNzZG4ubmV0_size_16_color_FFFFFF_t_70]: /images/20220517/68eb78e3676e40b18a3b0122c07581d5.png
[RabbitMQ]: https://blog.csdn.net/zhuyu19911016520/article/details/79317579
[70 1]: /images/20220517/751119db979747b8b340a36da3dc376c.png
[70 2]: /images/20220517/2b107fc2a2a24668a5eeb70ca7c1d410.png
[70 3]: /images/20220517/c7f5a49447cc4d3eb5442d636794a970.png
[70 4]: /images/20220517/5013d21feaa14df3aa5a8c4410790e2d.png
[RabbitMQ 1]: https://blog.csdn.net/zhuyu19911016520/article/details/80656583
[Link 1]: https://blog.csdn.net/zhuyu19911016520/article/details/80020079