【Spring Boot 7】RabbitMQ基础知识总结

水深无声 2021-08-27 00:40 389阅读 0赞

# 一、RabbitMQ简介 #

RabbitMQ即一个消息队列，主要是用来实现应用程序的异步和解耦，同时也能起到消息缓冲，消息分发的作用。

消息中间件在互联网公司的使用中越来越多，消息中间件最主要的作用是解耦，中间件最标准的用法是生产者生产消息传送到队列，消费者从队列中拿取消息并处理，生产者不用关心是谁来消费，消费者不用关心谁在生产消息，从而达到解耦的目的。在分布式的系统中，消息队列也会被用在很多其它的方面，比如：分布式事务的支持，RPC的调用等等。

RabbitMQ是实现AMQP（高级消息队列）的消息中间件的一种，最初起源于金融系统，用于在分布式系统中存储转发消息，在易用性、扩展性、高可用性等方面表现不俗。RabbitMQ主要是为了实现系统之间的双向解耦而实现的。当生产者大量产生数据时，消费者无法快速消费，那么需要一个中间层保存这个数据。

AMQP，即advanced  message queuing protocol，高级消息队列协议，是应用层协议的一个开放标准，为面向消息的中间件设计。消息中间件主要用于组件之间的解耦，消息的发送者无需知道消息使用者的存在，反之亦然。AMQP的主要特征是面向消息、队列、路由（包括点对点和发布/订阅）、可靠安全。

RabbitMQ是一个开源的AMQP实现，服务器用Erlang语言编写，支持多种客户端，如 Python、Ruby、.NET、Java、JMS、C、PHP、ActionScript、XMPP、STOMP 等 ，支持AJAX。用于在分布式系统中存储转发消息，在易用性、扩展性、高可用性方面表现不俗。

在安装完rabbitMq后，输入[http://ip:15672/][http_ip_15672] ，是可以看到一个简单后台管理界面的。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2d1b3J1aV9qYXZh_size_16_color_FFFFFF_t_70][]

# 二、RabbitMQ的特点 #

1、被设计为一个通用的消息代理服务，支持点对点、请求-回复、发布-订阅等模式。

2、专注于消息的一致性、可靠性和稳定性。

3、对Java、.net、node.js、PHP等多语言有着非常好的支持，并且有丰富的插件来提供功能扩展，没有做不到只有想不到。

4、支持同步或异步消息通信，支持分布式部署和多节点集群部署，能提供2万/秒的消息分发能力，性能上不敌kafka。

# 三、RabbitMQ的相关概念 #

通常我们谈到队列服务，会有三个概念：发消息者、队列、收消息者，RabbitMQ在这个基本概念之上，多做了一层抽象，在发消息者和队列之间加入了交换器exchange，这样发消息者和队列就没有直接联系，转而编程发消息者把消息给交换器，交换器根据调度策略再把消息给队列。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2d1b3J1aV9qYXZh_size_16_color_FFFFFF_t_70 1][]

黄色的圈圈就是我们的消息推送服务，将消息推送到中间方框里面，也就是RabbitMQ服务器，然后经过服务器里面的交换机、队列等各种关系将数据处理入列，最终右边的蓝色圈圈消费者获取对应监听的消息。

那么，其中比较重要的概念就有四个，分别是：虚拟主机、交换机、队列和绑定。

*  虚拟主机：一个虚拟主机持有一组交换机、队列和绑定。为什么需要多个虚拟主机呢？很简单， RabbitMQ 当中，用户只能在虚拟主机的粒度进行权限控制。因此，如果需要禁止A组访问B组的交换机、队列、绑定，必须为A和B分别创建一个虚拟主机、每一个 RabbitMQ 服务器都有一个默认的虚拟主机“/”。
 *  交换机：Exchange用于转发消息，但是它不会做存储，如果没有queue bind到Exchange的话，它会直接丢掉Producer发送过来的消息。这里有一个比较重要的概念：路由键。消息到交换机的时候，交换机转发到对应的队列中，那么究竟转发到哪个队列，就要根据该路由键。
 *  绑定：也就是交换机需要和队列相绑定，这其中如上图所示，是多对多的关系。

## 交换机（Exchange） ##

交换机的功能主要是接收消息并转发到绑定的队列，交换机不存储消息，在启用ack模式后，交换机找不到队列会返回错误。交换机有四种类型：Direct 、Topic、Headers、Fanout

*  Direct ：direct类型的行为是“先匹配再投送”，即在绑定时设定一个routing\_key，消息的routing\_key匹配时，才会被交换机投送到绑定的队列中去。
 *  Topic：按规则转发消息（最灵活）
 *  Headers：设置header attribute参数类型的交换机（基本不用）
 *  Fanout：转发消息到所有绑定队列

### 1、Direct Exchange是RabbitMQ默认的交换机模型，也是最简单的模型，根据key全文匹配去寻找队列。 ###

![format_png][]

第一个 X - Q1 就有一个 binding key，名字为 orange； X - Q2 就有 2 个 binding key，名字为 black 和 green。当消息中的 路由键 和 这个 binding key 对应上的时候，那么就知道了该消息去到哪一个队列中。

Ps：为什么 X 到 Q2 要有 black，green，2个 binding key呢，一个不就行了吗？ - 这个主要是因为可能又有 Q3，而Q3只接受 black 的信息，而Q2不仅接受black 的信息，还接受 green 的信息。

### 2、Topic Exchange ###

Topic Exchange转发消息主要是根据通配符。在这种交换机下，队列和交换机的绑定会定义一种路由模式，那么，通配符就要在这种路由模式和路由键之间匹配后交换机才能转发消息。

在这种交换机模式下：

（1）路由键必须是一串字符，用句号隔开，比如说agreements.us，或者agreements.eu.stockholm等。

（2）路由模式必须包含一个星号（\*），主要用于匹配路由键指定位置的一个单词，比如说，一个路由模式是这样的： agreements..b.\*， 那么就只能匹配路由键是这样子的：第一个单词是agreements，第四个单词是b。井号就表示相当于一个或多个单词，例如一个匹配模式是 agreements.eu.berlin.\#，那么，以agreements.eu.berlin 开头的路由键都是可以的。

具体代码发送的时候还是一样，第一个参数表示交换机，第二个参数表示routing key，第三个参数即消息。如下：

rabbitTemplate.convertAndSend("testTopicExchange","key1.a.c.key2", " this is  RabbitMQ!");

topic 和 direct 类似, 只是匹配上支持了”模式”, 在”点分”的 routing\_key 形式中, 可以使用两个通配符:

*表示一个词
    #表示零个或多个词

### 3、Header Exchange  ###

headers 也是根据规则匹配, 相较于 direct 和 topic 固定地使用 routing\_key , headers 则是一个自定义匹配规则的类型. 在队列与交换器绑定时, 会设定一组键值对规则, 消息中也包括一组键值对( headers 属性), 当这些键值对有一对, 或全部匹配时, 消息被投送到对应队列。

### 4、 Fanout Exchange  ###

Fanout Exchange  消息广播的模式，不管路由键或者是路由模式，会把消息发给绑定给它的全部队列，如果配置了routing\_key会被忽略。

# 四、rabbitmq有哪些重要组件？ #

## 1、ConnectionFactory(连接管理器) ##

应用程序与RabbitMQ之间建立连接的管理器。

## 2、Channel(信道) ##

消息推送使用的通道。

## 3、RoutingKey(路由键) ##

用于把生产者的数据分配到交换器上。

## 4、Exchange(交换器) ##

用于接受、分配消息。

## 5、BindKey(绑定键) ##

用于把交换器的消息绑定到队列上

## 6、Queue(队列) ##

用于存储生产者的消息。

# 五、RabbitMQ的适用场景 #

## 1、异步处理 ##

场景说明：用户注册后，需要发送注册邮件和注册短信。

传统的做法有两种，①串行的方式；②并行的方式；

### （1）串行方式 ###

将注册信息写入数据库后，发送注册邮件，再发送注册短信，以上三个动作顺序完成后才返回给客户端。

这种方式的问题是发送注册邮件和发送注册短信不是必须的，它只是一个通知，而这种做法让客户端等待没有必要等待的东西，影响用户体验。

![20200817183446354.png][]

### （2）并行方式 ###

将注册信息写入数据库后，发送邮件的同时，发送短信，以上三种操作都完成后，返回给客户端，并行的方式能缩短处理的时间。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2d1b3J1aV9qYXZh_size_16_color_FFFFFF_t_70 2][]

假设三种业务节点分别使用50ms，串行方式就是150ms，并行方式就是100ms。

虽然并行已经缩短了处理时间，但是前面说过，邮件和短信不是必须的，没有必要让客户等待响应，应该是写入数据库就直接返回。

### （3） 消息队列RabbitMQ ###

引入消息队列后，用户的响应时间就等于写入数据库的时间+写入消息队列的时间（可以忽略不计），引入消息队列后，响应时间大幅度缩减。

## 2、应用解耦 ##

### （1）场景 ###

双11是购物狂欢节，用户下单后，订单系统需要通知库存系统，传统的做法就是订单系统调用库存系统接口。

![20200817185542404.png][]

### （2）缺点 ###

由于订单系统和库存系统的高耦合，当库存系统出现故障时，订单就会失败。

### （3）引入消息队列RabbitMQ ###

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2d1b3J1aV9qYXZh_size_16_color_FFFFFF_t_70 3][]

订单系统：用户下单后，订单系统完成持久化处理，将消息写入消息队列，返回用户订单下单成功。

库存系统：订阅下单的消息，获取下单消息，进行库操作。就算库存系统出现故障，消息队列也能保证订单消息的可靠传递，不会导致消息丢失。

## 3、流量削峰 ##

流量削峰一般在秒杀活动中应用广泛

场景：秒杀活动，一般因为流量过大，导致应用挂掉，为了解决这个问题，一般在应用前端加入消息队列。

### （1）作用 ###

① 可以控制活动人数，超过此阈值的订单直接丢弃。一般秒杀就是这个原理。

② 可以缓解短时间内的高流量压垮应用（应用程序按照自己的最大处理能力获取订单）。

![20200817205352971.png][]

### （2）用户的请求，服务器收到之后，首先写入消息队列，加入消息队列长度超过最大值，则直接抛弃用户请求或跳转到错误页面。 ###

### （3）秒杀业务根据消息队列中的请求信息，再做后续处理。 ###

# 六、springboot集成RabbitMQ #

Spring Boot 集成 RabbitMQ 非常简单，如果只是简单的使用配置非常少，Spring Boot 提供了spring-boot-starter-amqp项目对消息各种支持。

## 1、pom.xml ##

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

## 2、配置 RabbitMQ 的安装地址、端口以及账户信息 ##

spring.application.name=Spring-boot-rabbitmq
    spring.rabbitmq.host=192.168.0.86
    spring.rabbitmq.port=5672
    spring.rabbitmq.username=admin
    spring.rabbitmq.password=123456

## 3、队列配置 ##

@Configuration
    public class RabbitConfig {
    
        @Bean
        public Queue Queue() {
            return new Queue("hello");
        }
    }

## 4、发送者 ##

rabbitTemplate 是 Spring Boot 提供的默认实现

@component
    public class HelloSender {
    	@Autowired
    	private AmqpTemplate rabbitTemplate;
    
    	public void send() {
    		String context = "hello " + new Date();
    		System.out.println("Sender : " + context);
    		this.rabbitTemplate.convertAndSend("hello", context);
    	}
    }

## 5、接收者 ##

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

## 6、测试 ##

@RunWith(SpringRunner.class)
    @SpringBootTest
    public class RabbitMqHelloTest {
    
    	@Autowired
    	private HelloSender helloSender;
    
    	@Test
    	public void hello() throws Exception {
    		helloSender.send();
    	}
    }

发送者和接收者的queue name必须一致，否则不能接收

# 七、多对多使用 #

一个发送者，N个接收者或者N个发送者和N个接收者会出现什么情况呢？

## 1、一对多发送 ##

对上面的代码进行了小改造，接收端注册了两个 Receiver,Receiver1 和 Receiver2，发送端加入参数计数，接收端打印接收到的参数，下面是测试代码，发送一百条消息，来观察两个接收端的执行效果

@Test
    public void oneToMany() throws Exception {
    	for (int i=0;i<100;i++){
    		neoSender.send(i);
    	}
    }

结果如下：

Receiver 1: Spring boot neo queue ****** 11
    Receiver 2: Spring boot neo queue ****** 12
    Receiver 2: Spring boot neo queue ****** 14
    Receiver 1: Spring boot neo queue ****** 13
    Receiver 2: Spring boot neo queue ****** 15
    Receiver 1: Spring boot neo queue ****** 16
    Receiver 1: Spring boot neo queue ****** 18
    Receiver 2: Spring boot neo queue ****** 17
    Receiver 2: Spring boot neo queue ****** 19
    Receiver 1: Spring boot neo queue ****** 20

根据返回结果得到以下结论

一个发送者，N个接收者，经过测试会均匀的将消息发送到N个接收者中

## 2、多对多发送 ##

复制了一份发送者，加入标记，在一百个循环中相互交替发送

@Test
    public void manyToMany() throws Exception {
    	for (int i=0;i<100;i++){
    		neoSender.send(i);
    		neoSender2.send(i);
    	}
    }

结果如下：

Receiver 1: Spring boot neo queue ****** 20
    Receiver 2: Spring boot neo queue ****** 20
    Receiver 1: Spring boot neo queue ****** 21
    Receiver 2: Spring boot neo queue ****** 21
    Receiver 1: Spring boot neo queue ****** 22
    Receiver 2: Spring boot neo queue ****** 22
    Receiver 1: Spring boot neo queue ****** 23
    Receiver 2: Spring boot neo queue ****** 23
    Receiver 1: Spring boot neo queue ****** 24
    Receiver 2: Spring boot neo queue ****** 24
    Receiver 1: Spring boot neo queue ****** 25
    Receiver 2: Spring boot neo queue ****** 25

结论：和一对多一样，接收端仍然会均匀接收到消息。

# 八、高级使用 #

## 1、对象的支持 ##

springboot完美的支持对象的发送和接收，不需要额外的配置。

//发送者
    public void send(User user) {
    	System.out.println("Sender object: " + user.toString());
    	this.rabbitTemplate.convertAndSend("object", user);
    }
    
    ...
    
    //接收者
    @RabbitHandler
    public void process(User user) {
        System.out.println("Receiver object : " + user);
    }

结果如下：

Sender object: User{name='neo', pass='123456'}
    Receiver object : User{name='neo', pass='123456'}

## 2、Topic  Exchange ##

topic 是 RabbitMQ 中最灵活的一种方式，可以根据 routing\_key 自由的绑定不同的队列

首先对 topic 规则配置，这里使用两个队列来测试

@Configuration
    public class TopicRabbitConfig {
    
        final static String message = "topic.message";
        final static String messages = "topic.messages";
    
        @Bean
        public Queue queueMessage() {
            return new Queue(TopicRabbitConfig.message);
        }
    
        @Bean
        public Queue queueMessages() {
            return new Queue(TopicRabbitConfig.messages);
        }
    
        @Bean
        TopicExchange exchange() {
            return new TopicExchange("exchange");
        }
    
        @Bean
        Binding bindingExchangeMessage(Queue queueMessage, TopicExchange exchange) {
            return BindingBuilder.bind(queueMessage).to(exchange).with("topic.message");
        }
    
        @Bean
        Binding bindingExchangeMessages(Queue queueMessages, TopicExchange exchange) {
            return BindingBuilder.bind(queueMessages).to(exchange).with("topic.#");
        }
    }

使用 queueMessages 同时匹配两个队列，queueMessage 只匹配 “topic.message” 队列

public void send1() {
    	String context = "hi, i am message 1";
    	System.out.println("Sender : " + context);
    	this.rabbitTemplate.convertAndSend("exchange", "topic.message", context);
    }
    
    public void send2() {
    	String context = "hi, i am messages 2";
    	System.out.println("Sender : " + context);
    	this.rabbitTemplate.convertAndSend("exchange", "topic.messages", context);
    }

发送send1会匹配到topic.\#和topic.message 两个Receiver都可以收到消息，发送send2只有topic.\#可以匹配所有只有Receiver2监听到消

## 3、 Fanout  Exchange ##

Fanout 就是我们熟悉的广播模式或者订阅模式，给 Fanout 交换机发送消息，绑定了这个交换机的所有队列都收到这个消息。

Fanout 相关配置

@Configuration
    public class FanoutRabbitConfig {
    
        @Bean
        public Queue AMessage() {
            return new Queue("fanout.A");
        }
    
        @Bean
        public Queue BMessage() {
            return new Queue("fanout.B");
        }
    
        @Bean
        public Queue CMessage() {
            return new Queue("fanout.C");
        }
    
        @Bean
        FanoutExchange fanoutExchange() {
            return new FanoutExchange("fanoutExchange");
        }
    
        @Bean
        Binding bindingExchangeA(Queue AMessage,FanoutExchange fanoutExchange) {
            return BindingBuilder.bind(AMessage).to(fanoutExchange);
        }
    
        @Bean
        Binding bindingExchangeB(Queue BMessage, FanoutExchange fanoutExchange) {
            return BindingBuilder.bind(BMessage).to(fanoutExchange);
        }
    
        @Bean
        Binding bindingExchangeC(Queue CMessage, FanoutExchange fanoutExchange) {
            return BindingBuilder.bind(CMessage).to(fanoutExchange);
        }
    }

这里使用了 A、B、C 三个队列绑定到 Fanout 交换机上面，发送端的 routing\_key 写任何字符都会被忽略：

public void send() {
    	String context = "hi, fanout msg ";
    	System.out.println("Sender : " + context);
    	this.rabbitTemplate.convertAndSend("fanoutExchange","", context);
    }

结果如下：

Sender : hi, fanout msg 
    ...
    fanout Receiver B: hi, fanout msg 
    fanout Receiver A  : hi, fanout msg 
    fanout Receiver C: hi, fanout msg

结果说明，绑定到 fanout 交换机上面的队列都收到了消息。

上一篇：[【Spring Boot 6】自定义starter][Spring Boot 6_starter]

下一篇：[【Spring Boot 8】Okhttp实现GitHub第三方登录][Spring Boot 8_Okhttp_GitHub]

![20200817154406402.jpg][]

[http_ip_15672]: http://ip:15672/
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2d1b3J1aV9qYXZh_size_16_color_FFFFFF_t_70]: /images/20210813/63d995bd71dd4b8fa4481fdb13e88ee9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2d1b3J1aV9qYXZh_size_16_color_FFFFFF_t_70 1]: /images/20210813/7fa2d77660f24fd9ba780328c77d1265.png
[format_png]: /images/20210813/13422bbb767e4fa490bca4325846ea5d.png
[20200817183446354.png]: /images/20210813/2ac16f9a66ad4d94b247b403418c43d2.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2d1b3J1aV9qYXZh_size_16_color_FFFFFF_t_70 2]: /images/20210813/7e1ccc819a384efdb0ae342d7bc2f4fb.png
[20200817185542404.png]: /images/20210813/5f64fa5f68f14a7daf64c9f5000b7f31.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2d1b3J1aV9qYXZh_size_16_color_FFFFFF_t_70 3]: /images/20210813/b72c7dcf4e95441eb8d5e591d2f8e8a7.png
[20200817205352971.png]: /images/20210813/03a0e3463fa04da5b40904e06b8d2cd5.png
[Spring Boot 6_starter]: https://blog.csdn.net/guorui_java/article/details/106867215
[Spring Boot 8_Okhttp_GitHub]: https://blog.csdn.net/guorui_java/article/details/104328355
[20200817154406402.jpg]: /images/20210813/2f91703403f14dc9bedb14df1357b9dc.png