什么是RabbitMQ
RabbitMQ是实现了高级消息队列协议（AMQP）的开源消息代理软件（亦称面向消息的中间件）。RabbitMQ服务器是用Erlang语言编写的，而集群和故障转移是构建在开放电信平台框架上的。所有主要的编程语言均有与代理接口通讯的客户端库。
RabbitMQ的安装
(1)、安装erlang
1、下载：https://packages.erlang-solutions.com/erlang/rpm/centos/6/x86_64/esl-erlang_21.0-1~centos~6_amd64.rpm
2、安装erlang：rpm -ivh esl-erlang_21.0-1_centos_6_amd64.rpm（结果失败）
3、安装erlang依赖：yum install unixODBC unixODBC-devel wxBase wxGTK SDL wxGTK-gl
4、再安装erlang：rpm -ivh esl-erlang_21.0-1_centos_6_amd64.rpm（结果成功）
5、测试：
(2)、安装RabbitMQ
1、下载：https://github.com/rabbitmq/rabbitmq-server/releases/download/v3.7.18/rabbitmq-server-3.7.18-1.el6.noarch.rpm
2、安装RabbitMQ：rpm -ivh rabbitmq-server-3.7.18-1.el6.noarch.rpm（结果失败）
3、安装RabbitMQ依赖：
rpm -ivh esl-erlang-compat-R14B-1.el6.noarch.rpm
rpm -ivh socat-1.7.3.2-1.el6.lux.x86_64.rpm
4、安装RabbitMQ：rpm -ivh rabbitmq-server-3.7.18-1.el6.noarch.rpm（结果成功）
5、测试： rabbitmq-server start
6、设置开机启动：chkconfig rabbitmq-server on
（3）、安装RabbitMQ管理界面
1、安装管理界面
rabbitmq-plugins enable rabbitmq_management
2、创建用户
rabbitmqctl add_user admin 1111
rabbitmqctl set_user_tags admin administrator
rabbitmqctl set_permissions -p "/" admin ".*" ".*" ".*"
3、测试：http://serviceIP:15672
特点：
开源的消息代理和队列服务器，在普通协议在不同的应用之间共享数据，使用Erlang编写（Erlang进行数据交换的性能很好，和原生socket一样好的延迟响应效果），遵守AMQP协议，AMQP主要特征是面向消息、队列、路由（点对点的发布/订阅）、可靠性、安全。适合对数据一致性、稳定性（100%消息投递）和可靠性要求很高的场景，对性能和吞吐量的要求其次。
优点：
能达到几万级别的吞吐量（和具体的机器配置有关），但是相比kafka和RocketMQ差了不少，因为采用Erlang，时延性是微秒级别，最重要的是有专门的可视化界面，很方便进行管理，社区非常活跃，功能很完备
缺点：
吞吐量的问题，因为是Erlang语言开发，很难去查看源码。如果存在什么问题，只能等官方去解决，对其掌控度很低，但是社区活跃度很高，这个影响也不大的
RabbitMQ基础概念
1 Provider
消息生产者，就是投递消息的程序。
2 Consumer
消息消费者，就是接受消息的程序。
3 没有使用消息队列时消息传递方式
4 使用消息队列后消息传递方式
5 什么是队列？
队列就像存放了商品的仓库或者商店，是生产商品的工厂和购买商品的用户之间的中转站
6队列里存储了什么？
在 rabbitMQ 中，信息流从你的应用程序出发，来到 Rabbitmq 的队列，所有信息可以只存储在一个队列中。队列可以存储很多信息，因为它基本上是一个无限制的缓冲区，前提是你的机器有足够的存储空间。
7队列和应用程序的关系？
多个生产者可以将消息发送到同一个队列中，多个消息者也可以只从同一个队列接收数
据
编写 RabbitMQ 的入门案例
1)新建工程，添加RabbitMQ启动器或者手动在pom文件中添加
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
2)修改全局配置文件application
#配置rabbitmq的连接信息
spring:
rabbitmq:
host: 192.168.1.19
port: 5672
username: admin
password: 1111
#自定义服务名称
application:
name: springcloud-rabbitmq
3)创建消息列队
import org.springframework.amqp.core.Queue;//注意，此处有很多重名类，不要导错包
@Configuration
public class QueueConfig {
@Bean
public Queue createQueue(){
return new Queue("hello-rabbitMQ");
}
}
4）消息接收者
@Component
public class Receiver{
/**
* @RabbitListener 监听消息RabbitMQ消息队列
* queues参数监听哪个消息队列，参数是个字符串数组，可以监听多个
*/
@RabbitListener(queues = "hello-rabbitMQ")
public void process(String msg){
System.out.println("从RabbitMQ中获取到信息："+msg);
}
}
5)消息发送者
@Component
public class Sender {
@Autowired
private AmqpTemplate amqpTemplate;//Spring中的内置对象，用于操作rabbitMQ
/**
* 自定义方法向详消息队列发送消息
*/
public void send(String msg){
/**
* 参数一、队列名称
* 参数二、发送的信息
*/
amqpTemplate.convertAndSend("hello-rabbitMQ",msg);
}
}
6）测试代码
@RunWith(SpringRunner.class)
@SpringBootTest(classes = RabbitMqApplication.class)
public class RabbitMqApplicationTests {
@Autowired
private Sender sender;
@Test
public void test01() throws InterruptedException {
while (true) {
Thread.sleep(1000);
sender.send("你好呀");
}
}
}
RabbitMQ 原理
原理图
RabbitMQ基础组件
1，Messge
消息，由消息头和消息体组成。消息体是不透明的，消息头由一些列可选属性组成，这些属性包括：routing-key（路由键）、priority（优先级）、delivery-mode（消息是否可持久性存储）。
2，Publisher
消息的生产者，也是一个向交换器发布消息的客户端应用程序。
3，Consumer
消息的消费者，表示一个从消息队列中取得消息的客户端程序。
4，Exchange
交换器，用来接收生产者发送的消息，并将这些消息路由给服务器中的队列。
三种常见的交换器：direct（发布与订阅，完全匹配）、fanout（广播）、topic（主题，规则匹配）
5，Binding
绑定，用于消息队列和交换器之间的关联。一个绑定就是基于路由键将交换器与消息队列连接起来的路由规则，所以可以将交换器理解成一个由绑定构成的路由表。
6，Queue
消息队列，用来保存消息直到发送给消费者。它是消息的容器，也是消息的终点。一个消息可以投入一个或多个队列。消息一直在队列里面，等待消费者连接到这个队列取走消息。
7，Routing-key
路由键，RabbitMQ决定消息该投递到哪个队列的规则。队列通过路由键绑定到交换器。消息发送到MQ服务器时，消息拥有一个路由键，即便是空的，RabbitMQ也会将其和绑定使用的路由键进行匹配。如果相匹配，消息会投递到该队列，如果不匹配，消息将会进入黑洞。
8，Connection
链接，指rabbitMQ服务器和服务建立的TCP连接。
9，Channel
信道，是TCP里面的虚拟连接。例如，电缆相当于TCP，信道是一个独立的光纤束，一条TCP连接上创建多条信道是没有问题的。TCP一旦打开，就会创建AMQP信道，无论发布消息、接受消息、订阅队列，这些动作都是通过信道完成的。
10，Virtual Host
虚拟主机，表示一批交换机，消息队列和相关对象。虚拟主机是共享相同的身份认证和加密环境的独立服务器。每个virtualhost本质上就是一个mini版的RabbitMQ服务器，拥有自己的队列、交换器、绑定和权限机制。vhost是AMQP概念的基础，必须在连接时指定，RabbitMQ默认的vhost是/。
11，Broker
表示消息队列服务器实体。
RabbitMQ组件相关知识
1，交换器和队列的关系
交换器是通过路由键和队列绑定在一起的，如果消息拥有的路由键跟队列和交换器路由键匹配，那么消息就会被路由到该绑定的队列中。
也就是说，消息到队列的过程，消息首先会经过交换器，接下来交换器再通过路由键分发消息到具体的队列中。
路由键可以理解为匹配的规则。
2，RabbitMQ使用信道而不是直接使用TCP通信的原因
TCP的创建和销毁开销特别大，创建需要3次握手，销毁需要4次挥手。如果不用信道，那么应用程序就会以TCP连接到RabbitMQ，高峰时每秒成千上万条连接会造成资源的巨大浪费，而且操作系统每秒处理TCP连接数也是有限制的，必定造成性能瓶颈。
信道的原理是一条线程一条信道，多条线程多条信道同用一条TCP连接。一条TCP连接可以容纳无线信道即使每秒成千上万的请求也不会成为性能的瓶颈。
RabbitMQ之Direct交换器模式开发
Dirtct交换器，即发布与订阅模式，匹配规则为完全匹配。
Provider--消息提供者
application配置文件
#设置rabbitmq的连接配置
spring.rabbitmq.host=192.168.1.19
spring.rabbitmq.port=5672
spring.rabbitmq.username=admin
spring.rabbitmq.password=1111
spring.application.name=springcloud-rabbitmq
#设置交换器的名称
mq.config.exchange=log.direct
#info路由键
mq.config.queue.info.routing.key=log.info.routing.key
#error路由键
mq.config.queue.error.routing.key=log.error.routing.key
消息发送类
@Component
public class Sender {
@Autowired
private AmqpTemplate amqpTemplate;
@Value("${mq.config.exchange}")
private String exchange;
@Value("${mq.config.queue.error.routing.key}")
private String error_routing_key;
@Value("${mq.config.queue.info.routing.key}")
private String info_routing_key;
public void send(String msg){
/**
* 参数一、交换器名称
* 参数二、路由键
* 参数三、信息
*/
//amqpTemplate.convertAndSend(exchange,error_routing_key,msg);
amqpTemplate.convertAndSend(exchange,info_routing_key,msg);
}
}
consumer--消息的消费者
application配置文件
#设置rabbitmq的连接配置
spring.rabbitmq.host=192.168.1.19
spring.rabbitmq.port=5672
spring.rabbitmq.username=admin
spring.rabbitmq.password=1111
spring.application.name=springcloud-rabbitmq
#设置交换器的名称
mq.config.exchange=log.direct
#info队列名称
mq.config.queue.info=log.info
#info路由键
mq.config.queue.info.routing.key=log.info.routing.key
#error队列名称
mq.config.queue.error=log.error
#error路由键
mq.config.queue.error.routing.key=log.error.routing.key
info消息列队的消费者类
/**
* 消息接收者
* @RabbitListener bindings:绑定队列
* @QueueBinding value:绑定队列的名称
* exchange:配置交换器
*
* @Queue value:配置队列名称
* autoDelete:是否是一个可删除的临时队列
*
* @Exchange value:为交换器起个名称
* type:指定具体的交换器类型
*/
@Component
@RabbitListener(
bindings = @QueueBinding(
value = @Queue(value = "${mq.config.queue.info}",autoDelete = "true"),
exchange = @Exchange(value = "${mq.config.exchange}",type = ExchangeTypes.DIRECT),
key = "${mq.config.queue.info.routing.key}"
)
)
public class InfoReceiver {
@RabbitHandler
public void process(String msg){
System.out.println("info.........："+msg);
}
}
error消息列队的消费者类
@Component
@RabbitListener(
bindings = @QueueBinding(
value = @Queue(value = "${mq.config.queue.error}",autoDelete = "true"),
exchange = @Exchange(value = "${mq.config.exchange}",type = ExchangeTypes.DIRECT),
key = "${mq.config.queue.error.routing.key}"
)
)
public class ErrorReceiver {
@RabbitHandler
public void process(String msg){
System.out.println("error.........："+msg);
}
}
测试
首先启动：信息消费者启动类
@SpringBootApplication
public class RabbitMqApplication {
public static void main(String[] args) {
SpringApplication.run(RabbitMqApplication.class, args);
}
}
然后运行：消息提供者测试类
@RunWith(SpringRunner.class)
@SpringBootTest(classes = RabbitMqApplication.class)
public class RabbitMqApplicationTests {
@Autowired
private Sender sender;
@Test
public void test01() throws InterruptedException {
while (true) {
Thread.sleep(1000);
sender.send("你好呀");
}
}
}
测试2：
将消息提供者类中的代码改为，其他不变，然后重启消息提供者测试类
amqpTemplate.convertAndSend(exchange,error_routing_key,msg);
//amqpTemplate.convertAndSend(exchange,info_routing_key,msg);
RabbitMQ之Topic交换器模式开发
Topic交换器，即主题模式，进行规则匹配。
Provider--消息提供者
spring.rabbitmq.host=192.168.1.19
spring.rabbitmq.port=5672
spring.rabbitmq.username=admin
spring.rabbitmq.password=1111
spring.application.name=springcloud-rabbitmq
#设置交换器的名称
mq.config.exchange=log.topic
消息提供者1、OrderSender
@Component
public class OrderSender {
@Autowired
private AmqpTemplate amqpTemplate;
@Value("${mq.config.exchange}")
private String exchange;
public void send(String msg) {
/**
* 参数一、交换器名称
* 参数二、路由键
* 参数三、信息
*/
this.amqpTemplate.convertAndSend(this.exchange, "order.log.debug", "order.log.debug....." + msg);
this.amqpTemplate.convertAndSend(this.exchange, "order.log.info", "order.log.info....." + msg);
this.amqpTemplate.convertAndSend(this.exchange, "order.log.warn", "order.log.warn....." + msg);
this.amqpTemplate.convertAndSend(this.exchange, "order.log.error", "order.log.error....." + msg);
}
}
消息提供者2、ProductSender
@Component
public class ProductSender {
@Autowired
private AmqpTemplate amqpTemplate;
@Value("${mq.config.exchange}")
private String exchange;
public void send(String msg) {
/**
* 参数一、交换器名称
* 参数二、路由键
* 参数三、信息
*/
this.amqpTemplate.convertAndSend(this.exchange, "product.log.debug", "product.log.debug....." + msg);
this.amqpTemplate.convertAndSend(this.exchange, "product.log.info", "product.log.info....." + msg);
this.amqpTemplate.convertAndSend(this.exchange, "product.log.warn", "product.log.warn....." + msg);
this.amqpTemplate.convertAndSend(this.exchange, "product.log.error", "product.log.error....." + msg);
}
}
消息提供者3、UserSender
@Component
public class UserSender {
@Autowired
private AmqpTemplate amqpTemplate;
@Value("${mq.config.exchange}")
private String exchange;
public void send(String msg) {
/**
* 参数一、交换器名称
* 参数二、路由键
* 参数三、信息
*/
this.amqpTemplate.convertAndSend(this.exchange, "user.log.debug", "user.log.debug....." + msg);
this.amqpTemplate.convertAndSend(this.exchange, "user.log.info", "user.log.info....." + msg);
this.amqpTemplate.convertAndSend(this.exchange, "user.log.warn", "user.log.warn....." + msg);
this.amqpTemplate.convertAndSend(this.exchange, "user.log.error", "user.log.error....." + msg);
}
}
consumer--消息消费者
application配置文件
spring.rabbitmq.host=192.168.1.19
spring.rabbitmq.port=5672
spring.rabbitmq.username=admin
spring.rabbitmq.password=1111
spring.application.name=springcloud-rabbitmq
#设置交换器的名称
mq.config.exchange=log.topic
#info队列名称
mq.config.queue.info=log.info
#error队列名称
mq.config.queue.error=log.error
#logs队列名称
mq.config.queue.logs=log.all
消费者1、ErrorReceiver,只接收消费以.log.error结尾的消息列队中的信息
@Component
@RabbitListener(
bindings = @QueueBinding(
value = @Queue(value = "${mq.config.queue.error}",autoDelete = "true"),
exchange = @Exchange(value = "${mq.config.exchange}",type = ExchangeTypes.TOPIC),
key = "*.log.error"
)
)
public class ErrorReceiver {
@RabbitHandler
public void process(String msg){
System.out.println("........error.........："+msg);
}
}
消费者2、InfoReceiver 只接收消费以.log.info结尾的消息列队中的信息
@Component
@RabbitListener(
bindings = @QueueBinding(
value = @Queue(value = "${mq.config.queue.info}",autoDelete = "true"),
exchange = @Exchange(value = "${mq.config.exchange}",type = ExchangeTypes.TOPIC),
key = "*.log.info"
)
)
public class InfoReceiver {
@RabbitHandler
public void process(String msg){
System.out.println("........info.........："+msg);
}
}
消费者3、LogsReceiver--接收消费包含.log.列队的消息
@Component
@RabbitListener(
bindings = @QueueBinding( //绑定队列
value = @Queue(value = "${mq.config.queue.logs}",autoDelete = "true"),
exchange = @Exchange(value = "${mq.config.exchange}",type = ExchangeTypes.TOPIC),
key = "*.log.*"
)
)
public class LogsReceiver {
@RabbitHandler
public void process(String msg){
System.out.println("........all.........："+msg);
}
}
测试：
启动消费者启动器类：
@SpringBootApplication
public class RabbitMqApplication {
public static void main(String[] args) {
SpringApplication.run(RabbitMqApplication.class, args);
}
}
启动生产者测试类：
@RunWith(SpringRunner.class)
@SpringBootTest(classes = RabbitMqApplication.class)
public class RabbitMqApplicationTests {
@Autowired
private UserSender userSender;
@Autowired
private ProductSender productSender;
@Autowired
private OrderSender orderSender;
@Test
public void test01() throws InterruptedException {
userSender.send("你好呀");
productSender.send("你好呀");
orderSender.send("你好呀");
}
}
RabbitMQ之 Fanout 交换器（广播）模式开发
![]()
provider端代码，配置文件
application配置文件
#设置rabbitmq的连接配置
spring.rabbitmq.host=192.168.1.19
spring.rabbitmq.port=5672
spring.rabbitmq.username=admin
spring.rabbitmq.password=1111
spring.application.name=springcloud-rabbitmq
#设置交换器的名称
mq.config.exchange=fanout
Sender类
@Component
public class Sender {
@Autowired
private AmqpTemplate amqpTemplate;
@Value("${mq.config.exchange}")
private String exchange;
public void send(String msg){
/**
* 参数一、交换器名称
* 参数二、路由键
* 参数三、信息
*/
amqpTemplate.convertAndSend(exchange,"",msg);
}
}
consumer端配置文件和代码
application配置文件
#设置rabbitmq的连接配置
spring.rabbitmq.host=192.168.1.19
spring.rabbitmq.port=5672
spring.rabbitmq.username=admin
spring.rabbitmq.password=1111
spring.application.name=springcloud-rabbitmq
#设置交换器的名称
mq.config.exchange=fanout
#sms队列名称
mq.config.queue.sms=queue.sms
#push
mq.config.queue.push=queue.push
PushReceiver
@Component
@RabbitListener(
bindings = @QueueBinding(
value = @Queue(value = "${mq.config.queue.push}",autoDelete = "true"),
exchange = @Exchange(value = "${mq.config.exchange}",type = ExchangeTypes.FANOUT)
)
)
public class PushReceiver {
@RabbitHandler
public void process(String msg){
System.out.println("push.........："+msg);
}
}
SmsReceiver
@Component
@RabbitListener(
bindings = @QueueBinding(
value = @Queue(value = "${mq.config.queue.sms}",autoDelete = "true"),
exchange = @Exchange(value = "${mq.config.exchange}",type = ExchangeTypes.FANOUT)
)
)
public class SmsReceiver {
@RabbitHandler
public void process(String msg){
System.out.println("sms.........："+msg);
}
}
测试：
首先启动consumer的启动类
@SpringBootApplication
public class RabbitMqApplication {
public static void main(String[] args) {
SpringApplication.run(RabbitMqApplication.class, args);
}
}
然后运行provider的测试类
@RunWith(SpringRunner.class)
@SpringBootTest(classes = RabbitMqApplication.class)
public class RabbitMqApplicationTests {
@Autowired
private Sender sender;
@Test
public void test01(){
sender.send("你好呀");
}
}
运行结果
![]()
RabbitMQ 的消息持久化处理
消息的可靠性是 RabbitMQ 的一大特色，那么 RabbitMQ 是如何保证消息可靠性的呢——消息持久化！
如何设置消息持久化：
配置注解中的autoDelete 属性
@RabbitListener(
bindings = @QueueBinding(
value = @Queue(value = "${mq.config.queue.error}",autoDelete = "false"),
exchange = @Exchange(value = "${mq.config.exchange}",type = ExchangeTypes.DIRECT),
key = "${mq.config.queue.error.routing.key}"
)
)
在@Queue注解中: 当所有消费客户端连接断开后，是否自动删除队列 true：删除； false：不删除
在@Exchange注解中：当所有绑定队列都不在使用时，是否自动删除交换器 true：删除； false：不删除
RabbitMQ的消息确认ACK机制
1、什么是消息确认ACK。
答：如果在处理消息的过程中，消费者的服务器在处理消息的时候出现异常，那么可能这条正在处理的消息就没有完成消息消费，数据就会丢失。为了确保数据不会丢失，RabbitMQ支持消息确定-ACK。
2、ACK的消息确认机制。
答：ACK机制是消费者从RabbitMQ收到消息并处理完成后，反馈给RabbitMQ，RabbitMQ收到反馈后才将此消息从队列中删除。
如果一个消费者在处理消息出现了网络不稳定、服务器异常等现象，那么就不会有ACK反馈，RabbitMQ会认为这个消息没有正常消费，会将消息重新放入队列中。
如果在集群的情况下，RabbitMQ会立即将这个消息推送给这个在线的其他消费者。这种机制保证了在消费者服务端故障的时候，不丢失任何消息和任务。
消息永远不会从RabbitMQ中删除，只有当消费者正确发送ACK反馈，RabbitMQ确认收到后，消息才会从RabbitMQ服务器的数据中删除。
消息的ACK确认机制默认是打开的。
3、ACK机制的开发注意事项。
答：如果忘记了ACK，那么后果很严重。当Consumer退出时候，Message会一直重新分发。然后RabbitMQ会占用越来越多的内容，由于RabbitMQ会长时间运行，因此这个"内存泄漏"是致命的。
手动设置重试规则
# 开启重试
spring.rabbitmq.listener.simple.retry.enabled=true
# 重试次数,默认为3次
spring.rabbitmq.listener.simple.retry.max-attempts=5
