RabbitMq详解+SpringBoot整合RabbitMq快速入门

1概述：

1.1.什么是MQ

消息队列（Message Queue，简称MQ），从字面意思上看，本质是个队列，FIFO先入先出，只不过队列中存放的内容是message而已。
其主要用途：不同进程Process/线程Thread之间通信。

1.2.背景

在系统的处理能力有限的情况下，使用队列的机制吧任务暂时存放起来，系统再依次处理掉排队的任务，提高系统的吞吐的能力，随着微服务愈发的流行，多个子系统之间的相互调用成为必须，其中有很多的方式，比如：webService的SOAP请求，js的jsonp形式的请求，httpClient模拟发送HTTP请求，这些都有些共性的缺点，就是紧耦合，同步。

1.3.同类型其他产品

ZeroMQ最快，storm中就使用ZeroMQ；RabbitMQ和Spring框架无缝集成，是Spring官方默认的MQ；ActiveMQ是Apache产品，使用广泛。MSMQ是微软的产品，应用于.net架构中。还有大数据的kafka（scala语言开发）

2.RabbitMQ

2.1.介绍

RabbitMQ是一个开源的，RabbitMQ是实现AMQP（高级消息队列协议）的消息中间件的一种，最初起源于金融系统，用于在分布式系统中存储转发消息，在易用性、扩展性、高可用性等方面表现不俗。消息中间件主要用于组件之间的解耦，消息的发送者无需知道消息使用者的存在，反之亦然。

支持主流的操作系统Linux、Windows、MacOX等。

多种开发语言支持java、Python、Ruby、.NET、PHP、C/C++、node.js等

2.2.AMQP协议

AMQP，即Advanced Message Queuing Protocol,一个提供统一消息服务的应用层标准高级消息队列协议,是应用层协议的一个开放标准,为面向消息的中间件设计。兼容了JMS，基于此协议的客户端与消息中间件可传递消息，并不受客户端/中间件不同产品，不同开发语言等条件的限制。Erlang中的实现有 RabbitMQ等

2.3.JMS

JMS即Java消息服务（Java Message Service）,Java消息服务是一个与具体平台无关的API，它类似于JDBC，提供同样与厂商无关的访问的API，以访问消息收发服务

	JMS	AMQP
定义	Java api	网络线级协议
跨语言	否	是
跨平台	否	是
Model	提供两种消息模型（1）Peer-2-Peer （2）Pub/sub	提供了五种消息模型：（1）、direct exchange （2）、fanout exchange （3）、topic change （4）、headers exchange （5）、system exchange 本质来讲，后四种和JMS的pub/sub模型没有太大差别，仅是在,路由机制上做了更详细的划分
支持消息类型	多种消息类型： TextMessage MapMessage BytesMessage ObjectMessage	byte[] 当实际应用时，有复杂的消息，可以将消息序列化后发送
综合评价	JMS 定义了JAVA API层面的标准；在java体系中，多个client均可以通过JMS进行交互，不需要应用修改代码，但是其对跨平台的支持较差；	AMQP定义了wire-level层的协议标准；天然具有跨平台、跨语言特性

2.4.核心概念

message：消息，消息是不具名的，它由消息头和消息体组成。消息体是不透明的，而消息头则由一系列的可选属性组成，这些属性包括routing-key（路由键）、priority（相对于其他消息的优先权）、delivery-mode（指出该消息可能需要持久性存储）等

publisher：消息的生产者，也是一个向交换器发布消息的客户端应用程序

exchange：交换器，用来接收生产者发送的消息并将这些消息路由给服务器中的队列。Exchange有4种类型：direct(默认)，fanout, topic, 和headers，不同类型的Exchange转发消息的策略有所区别

queue：消息队列，用来保存消息直到发送给消费者。它是消息的容器，也是消息的终点。一个消息可投入一个或多个队列。消息一直在队列里面，等待消费者连接到这个队列将其取走

binding：绑定，用于消息队列和交换器之间的关联。一个绑定就是基于路由键将交换器和消息队列连接起来的路由规则，所以可以将交换器理解成一个由绑定构成的路由表。Exchange 和Queue的绑定可以是多对多的关系

connection：网络连接，比如一个TCP连接

channel：信道，多路复用连接中的一条独立的双向数据流通道。信道是建立在真实的TCP连接内的虚拟连接，AMQP 命令都是通过信道发出去的，不管是发布消息、订阅队列还是接收消息，这些动作都是通过信道完成。因为对于操作系统来说建立和销毁 TCP 都是非常昂贵的开销，所以引入了信道的概念，以复用一条 TCP 连接

consumer：消息的消费者，表示一个从消息队列中取得消息的客户端应用程序

Virtual Host：虚拟主机，表示一批交换器、消息队列和相关对象。虚拟主机是共享相同的身份认证和加密环境的独立服务器域。每个 vhost 本质上就是一个 mini 版的 RabbitMQ 服务器，拥有自己的队列、交换器、绑定和权限机制。vhost 是 AMQP 概念的基础，必须在连接时指定，RabbitMQ 默认的 vhost 是 /

Broker：表示消息队列服务器实体

关系图

2.5.Exchange消息模型

Exchange分发消息时根据类型的不同分发策略有区别，目前共四种类型：direct、fanout、topic、headers 。headers 匹配 AMQP 消息的 header而不是路由键， headers 交换器和 direct 交换器完全一致，但性能差很多，目前几乎用不到了，所以直接看另外三种类型

2.5.1.Direct直连模式

消息中的路由键（routing key）如果和 Binding 中的 binding key 一致，交换器就将消息发到对应的队列中。路由键与队列名完全匹配，如果一个队列绑定到交换机要求路由键为“dog”，则只转发 routing key 标记为“dog”的消息，不会转发“dog.puppy”，也不会转发“dog.guard”等等。它是完全匹配、单播的模式

2.5.2.fanout广播模式

每个发到 fanout 类型交换器的消息都会分到所有绑定的队列上去。fanout 交换器不处理路由键，只是简单的将队列绑定到交换器上，每个发送到交换器的消息都会被转发到与该交换器绑定的所有队列上。很像子网广播，每台子网内的主机都获得了一份复制的消息。fanout 类型转发消息是最快的

2.5.3.topic主题模式

topic 交换器通过模式匹配分配消息的路由键属性，将路由键和某个模式进行匹配，此时队列需要绑定到一个模式上。它将路由键和绑定键的字符串切分成单词，这些单词之间用点隔开。它同样也会识别两个通配符：符号“#”和符号 “*”。

#匹配0个或多个单词，

*匹配一个单词

3.RabbitMq下载与介绍

3.1.环境准备

我这里是通过docker容器下的rabbitmq

下载镜像：docker pull rabbitmq

查看镜像：docker images

运行rabbitmq：docker run -d -p 5672:5672 -p 15672:15672 --name 别名容器ID

默认的账号密码：guest/guest

15672：管理界面的端口

5672：程序连接端口

3.2.操作界面介绍

在浏览器中输入地址查看：http://127.0.0.1:15672/

通过默认账号：guest/guest

3.2.1.添加用户

用户角色：

超级管理员(administrator) ： 可登陆管理控制台，可查看所有的信息，并且可以对用户，策略(policy)进行

操作监控者(monitoring) ： 可登陆管理控制台，同时可以查看rabbitmq节点的相关信息(进程数，内存使用情况，磁盘使用情况等)

策略制定者(policymaker) ： 可登陆管理控制台, 同时可以对policy进行管理。但无法查看节点的相关信息(上图红框标识的部分)

普通管理者(management) ： 仅可登陆管理控制台，无法看到节点信息，也无法对策略进行管理。

其他：无法登陆管理控制台，通常就是普通的生产者和消费者。

3.2.2.创建Virtual Hosts

3.3.3.创建Exchange，queue，以及exchange绑定queue

3.3.3.1创建exchange

3.3.3.2创建队列

3.3.3.3exchange绑定queue

绑定

3.3.3.界面功能介绍

4.与springboot整合入门

4.1.使用springboot创建向导创建项目

4.2.自动配置简介

查看rabbitmq的自动配置类RabbitAutoConfiguration

@Configuration(proxyBeanMethods = false
)
@ConditionalOnClass({RabbitTemplate.class, Channel.class})
@EnableConfigurationProperties({RabbitProperties.class})
@Import({RabbitAnnotationDrivenConfiguration.class})
public class RabbitAutoConfiguration {@Bean@ConditionalOnSingleCandidate(ConnectionFactory.class)@ConditionalOnMissingBean({RabbitOperations.class})//操作模板类，给rabbitmq发送和接受消息public RabbitTemplate rabbitTemplate(){....}@Bean@ConditionalOnSingleCandidate(ConnectionFactory.class)@ConditionalOnMissingBean//rabbitmq系统管理组件，和界面上的admin功能类似public AmqpAdmin amqpAdmin(ConnectionFactory connectionFactory) {...}
}
//属性配置类
@ConfigurationProperties(prefix = "spring.rabbitmq"
)
public class RabbitProperties {private String host = "localhost";private int port = 5672;//默认端口private String username = "guest";private String password = "guest";private final RabbitProperties.Ssl ssl = new RabbitProperties.Ssl();private String virtualHost;private String addresses;....
}

4.3.配置文件

spring.rabbitmq.host=192.168.32.101
#spring.rabbitmq.port=5672 默认配置
spring.rabbitmq.password=admin
spring.rabbitmq.username=admin
spring.rabbitmq.virtual-host=/test vhost #默认是/，这里是自定义的vhost

4.4测试

@SpringBootTest
class RabbitmqApplicationTests {@AutowiredRabbitTemplate rabbitTemplate;/*点对点*/@Testvoid directTest() {//message需要自己定义，包括消息体和消息头//rabbitTemplate.send(exchange,routeKey,message);User user = new User(1, "张三");rabbitTemplate.convertAndSend("test.exchage.direct", "test.news", user);}//测试广播模式，不需要指定路由key@Testpublic void fanout(){User user = new User(2, "李四");rabbitTemplate.convertAndSend("test.exchage.fanout", "", user);}//获取队列中的消息@Testvoid receive(){Object o = rabbitTemplate.receiveAndConvert("test.news");System.out.println(o.getClass());System.out.println(o);}
}

注意：User类要实现Serializable序列化接口，否则会消息转换报错

界面看到已经收到了一条消息，点击test.news名字，查看消息

发现序列化类型是这个，这是因为，RabbitTemplate默认是SimpleMessageConverter消息转换，在fromMessage方法中，有默认使用这个的方式

4.5.配置自定的消息转换，使用json格式

@Configuration
public class MessageConfig {@Beanpublic MessageConverter messageConverter(){return new Jackson2JsonMessageConverter();}
}

测试上面的广播模式，查看消息管理界面，发现绑定的队列都有收到消息

查看消息：json格式已正常

4.6.@EnableRabbit+@RabbitListener实现消息队列的监听

@Component
public class UserServiceImpl {//监听test.news消息队列，如果有消息，打印信息@RabbitListener(queues = "test.news")public void recevie(User user){System.out.println("收到消息"+user);}
}
//主启动类
@EnableRabbit//开启rabbitmq
@SpringBootApplication
public class RabbitmqApplication {public static void main(String[] args) {SpringApplication.run(RabbitmqApplication.class, args);}
}
//先启动主程序，然后执行测试方法，向test.news队列中添加消息，看控制台结果
收到消息com.ming.springboot.rabbitmq.entity.User@2fccd24