SpringBoot笔记：消息队列（RabbitMQ）

什么是消息队列

消息队列就是消息存储的容器,Java里面有两种

JMS：Sun公司出品，有两种模式，点对点和发布订阅。
AMQP：消息队列的一个协议，其实现有RabbitMQ，stormMQ等

我们会重点讲解RabbitMQ

消息队列的作用

异步通信

例如下面的案例，用户注册之后，我需要存储用户的信息，还要发邮件，发短信给用户。传统的方式呢，像第一个图，一步一步来，需要150ms

第二幅图呢好点了，用户信息存储到数据库之后，发送邮件和发送短信两个同时进行。需要100ms

第三幅图使用了消息队列，用户信息存储到数据库之后，直接存入消息队列，然后直接返回。是的，你没看错，直接返回了，这样用户等待的时间大概只需要55ms，真的是超级快了。那你可能会问了，消息队列的方式邮件和短信不计算时间了？肯定计算啊，但是你发邮件和发短信和用户有什么关系？用户注册之后提示注册成功就可以了，邮件和短信可以异步的去执行发送。

总结：

用户可以选择等待150ms看到注册成功的消息，有邮件和短信
用户可以选择等待100ms看到注册成功的消息，有邮件和短信
用户可以选择等待55ms看到成功的消息，邮件和短信稍后回来

应用解耦

比如一个商城系统，用户下了一个订单，我们肯定要去库存系统里面查一下，还有没有货了。这两个系统写在一起的话耦合度肯定是很高的。

现在采用消息队列的方式，订单系统和库存系统可以独立出来。用户下了订单，就存到消息队列里面，库存系统订阅了消息队列，里面一有内容就会去读取。

流量削峰

所谓的流量削峰，就类似于现在的1元秒杀活动。假如1万个人去秒杀一个1元商品，肯定不可能我1万的人一个一个的去判断，这样负载太大了。原理就是1万个用户请求进消息队列，我消息队列就一个位置，谁进去了谁就1元秒杀了，后面没进去的就不用判断了。秒杀业务处理就对消息队列里面的那一个幸运儿进行处理就可以了。所以，秒杀活动，你没进去，就别傻傻的在刷新了，因为你已经没有判断的资格了.......

RabbitMQ

RabbitMQ流程简介

讲一下，RabbitMQ的流程，如下图，首先是发布者发一个消息到RabbitMQ

我们可以看到RabbitMQ里面有很多的交换器路由和消息队列。很多。

交换器路由可以绑定多个消息队列，每个消息队列可以被多个交换器路由绑定

发布者发布的消息选择一个交换器路由，然后交换器路由会通过模式发给消息队列,这个模式下面讲。然后客户可以去获取队列的消息。

RabbitMQ的三种模式

上面讲了，交换器路由给队列发消息是通过模式筛选的，模式有三种

Direct：点对点模式，交换器路由只会给路由键为XXX的队列发消息
Topic：模糊匹配模式，# 匹配多个单词，* 匹配一个单词
Fanout：广播模式，交换器路由的每个绑定的队列都会收到消息

这个模糊匹配模式，我需要讲解一下，挺有意思的。

#是多个单词匹配，例如 Vae.#可以表示为 Vae.Music.com 。

而*是一个单词，Vae.就只能跟一个单词，例如 Vae.Music 后面不能再加了
假如我有一个交换器路由，绑定了4个队列，分别为 Vae.com，Vae.Music.com，shuyunquan.com，shuyunquan.Music.com
现在我的交换器路由发一个消息，路由键是Vae.#，消息内容是：哈哈哈。那么请问，这四个队列，哪几个可以收到消息呢？答案肯定是Vae.com，Vae.Music.com可以收到消息了，那么路由键是 #.com呢？又会是哪几个队列收到消息？简单吧

学了RabbitMQ的流程和三种模式，我们要开始实战一下了

安装RabbitMQ

我们还是使用Docker安装，你如果没学过Docker，你就不会知道Docker有多爽，我写的有Docker的博客，自己去翻阅学习。

先下载RabbitMQ，注意了，有latest版本和3-management版本的，两个版本，这个也是我的血泪史啊，latest没有后台网页，3-management版本有后台网页，别下错了

docker pull docker.io/rabbitmq:3-management

运行镜像，生成容器

docker run --name myrabbiymq -d -p 5672:5672 -p 15672:15672 rabbitmq:3-management

这里可以参考我的Docker一文的血泪史，还需要讲解一下，第一个 -p是5672，这个是RabbitMQ自己的端口，第二个 -p是15672，这个是给后台网页使用的

启动完成之后，我们打开浏览器，输入ip+15672，成功访问了

RabbitMQ交换器路由和队列的创建与绑定

我们来新建3个交换器路由，Direct，Fanout，Topic，新建如下图所示，访问你的RabbitMQ的网页，直接添加，type类型选择一致的，然后Durability要选择持久化的Durable，这样我们下次打开RabbitMQ的时候，交换器路由还是存在的

照葫芦画瓢，我3个交换器路由全部新建完毕了。接下来新建3个队列吧，如下图：

现在把交换器路由和队列进行绑定，Direct和Fanout的队列名和key都是一样的

但是Topic就不一样了，这个是模糊匹配，Vae.Music，Vae.com的key都写成Vae.#

RabbitMQ测试

绑定完成之后，发个消息测试一下，先来Direct的

果然是一对一，Direct一对一

再来试试Fanout

完美啊，加上我们Direct的一个，现在队列的消息是2 1 1了，我就不点进去看了

最后，试试我们的模糊匹配，我现在发一个Vae.JJ，看看效果咋样

3 2 1了，Nice啊！

RabbitMQ在Spring Boot中实现

引入RabbitMQ的Maven依赖

<!-- https://mvnrepository.com/artifact/org.springframework.boot/spring-boot-starter-amqp -->
<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId><version>2.1.3.RELEASE</version>
</dependency>

配置yml配置文件

指定一下我们RabbitMQ的服务器ip，用户名和密码都是默认的，我没改

spring:rabbitmq:host: 193.112.28.104username: guestpassword: guest

RabbitMQ Direct写入和读取

先写入RabbitMQ，我写的是Object类型的数据，默认是会序列化的

@AutowiredRabbitTemplate rabbitTemplate;@Testpublic  void testRabbitMQ(){//写入消息，Object会自动序列化Map<String,Object> map=new HashMap<>();map.put("msg","这是一个消息标题");map.put("data", Arrays.asList("许嵩",123,true));rabbitTemplate.convertAndSend("exchange.direct","Vae.com",map);}

可以看到，都被序列化了

这个时候，我们的Vae.com队列里面是有3个消息的，我们获取一下消息

@Testpublic void getRabbitMQ(){Object msg = rabbitTemplate.receiveAndConvert("Vae.com");System.out.println(msg.getClass());System.out.println(msg);}

你执行一下，会发现出来的消息不能看，因为Vae.com我们上面存了两个不是Object类型的数据，我们执行3次这个获取方法，可以发现，输出内容是

class java.util.HashMap
{msg=这是一个消息标题, data=[许嵩, 123, true]}

这个时候，你再去RabbitMQ的网页里面看Vae.com这个队列的消息数目，发现已经变成0了，说明只要获取方法一执行，队列里的消息就被读取了，就没了

RabbitMQ使用json序列化Object

上面的Object序列化总归不好看，为了好看，我们也可以使用json来序列化，替换Java本身的序列化就行，新建一个类，config文件夹下的MyRabbitMQConfig

package com.cache.config;import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;@Configuration
public class MyAMQPConfig {@Beanpublic MessageConverter messageConverter(){return  new Jackson2JsonMessageConverter();}
}

就这一个方法就可以了，我们再次执行上面的写入方法，然后来网页上看看

读取也是一样的，读出来也是json格式的，一目了然

自定义类类型上传

这次不使用Object了，我们自己定义一个类型，我新建一个类，叫Book

package com.cache.bean;public class Book {private String booName;private String author;public Book() {}public Book(String booName, String author) {this.booName = booName;this.author = author;}public String getBooName() {return booName;}public void setBooName(String booName) {this.booName = booName;}public String getAuthor() {return author;}public void setAuthor(String author) {this.author = author;}@Overridepublic String toString() {return "Book{" +"booName='" + booName + '\'' +", author='" + author + '\'' +'}';}
}

我们接下来就传这个类型的数据

@Testpublic  void testRabbitMQ(){rabbitTemplate.convertAndSend("exchange.direct","Vae.com",new Book("海上灵光","许嵩"));}

看看RabbitMQ的网页后台

很不错，类型Book都标出来了

RabbitMQ Fanout写入和读取

广播模式其实很简单，写入

@Testpublic  void FanoutWrite(){//广播的routingKey填不填都无所谓，没用rabbitTemplate.convertAndSend("exchange.fanout","",new Book("三国演义","罗贯中"));}

我就不截图了，RabbitMQ后台网页都是OK的

剩下的读取还有Topic都不讲了，都一样

发布者和订阅者的监听

发布者发布一个图书的消息，我的订阅者呢，可以立即的收到发布的消息，来写一下代码，发布者我们还使用上面的测试方法，订阅者写一个service

package com.cache.service;import com.cache.bean.Book;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Service;@Service
public class BookService {@RabbitListener(queues = "Vae.com")public void receive(Book book){System.out.println("收到消息了，发布的书籍是：" + book);}
}

还不行，还需要在主方法里面开启一下RabbitMQ的监听注解

@EnableRabbit

然后启动主方法，在启动一下发布的方法，你会看到主方法的输出框里已经有输出内容了

如果你想看消息的消息头信息，你可以这样写

//记住，Message是org.springframework.amqp.core.Message;包下的@RabbitListener(queues = "Vae.com")public void receiveHead(Message message){System.out.println(message.getBody());System.out.println(message.getMessageProperties());}

AmqpAdmin

AmqpAdmin就是使用代码来创建交换器路由和队列的，我们上面是自己在RabbitMQ的后台网页创建的，现在通过Amqp来使用代码进行创建

@AutowiredAmqpAdmin amqpAdmin;@Testpublic void amqp(){//创建交换器路由，还有很多参数，比如持久化我就不写了amqpAdmin.declareExchange(new DirectExchange("exchange.DirectTest"));//创建队列，可持久化为trueamqpAdmin.declareQueue(new Queue("Vae.Vae+",true));//绑定交换器路由和队列amqpAdmin.declareBinding(new Binding("Vae.Vae+",Binding.DestinationType.QUEUE,"exchange.DirectTest","Vae.Vae+",null));}