RabbitMQ
准备工作
RabbitMQMQ的简介和安装
MQ(Message Queue) 意为消息队列,运用的是生产者和消费者的设计模型,借用队列的数据结构:先进先出的特点,生产者可以不断的向队列中生产消息,消费者则可以在队列的另一端不断的消费消息,并且生产和消费的过程都是异步进行的,而且只关心消息的生产和接收,没有其它的业务的侵入,可以轻松实现系统的解耦,所以MQ又有一个别名叫消息中间件
。
通过利用高效可靠的消息传递机制进行平台无关的数据交流,并基于数据同学来实现分布式系统的集成。
市面上常用的MQ的区别:
MQ | 特点 |
---|---|
ActiveMQ | Apache出品,最流行,纯Java程序,运行需要Java虚拟机支持 |
Kafka | 开源的分布式发布-订阅消息系统,目前归属于Apache顶级项目,追求高吞吐量,一开始的目的就是用于日志收集和传输。0.8版本开始支持复制,不支持事务,对消息的重复、丢失、错误没有严格要求,适合产生大量数据的互联网服务的数据收集业务 |
RocketMQ | RocketMQ是阿里开源的消息中间件,它是纯Java开发,具有高吞吐量、高可用性、适合大规模分布式系统应用的特点,RocketMQ是对Kafka的优化 |
RabbitMQ | Erlang语言开发的消息队列模型,基于AMQP协议,主要特征是面向消息,队列,路由:包括点对点和消息订阅,稳定,安全并且可靠。 |
AMQP协议模型:
学过计算机网络的同学应该看的会很明白,队列类似一个路由器连接交换机的操作,交换机上面可以连接多个队列,生产者就可以指定交接机上面的队列向指定的队列发送消息,也可以用广播的方式发送消息到各个队列中,只要是接入该队列的消费者,就可以消费该消息。
安装:
链接:https://pan.baidu.com/s/1NKYymTioemob2FAK9z_z6g
提取码:d8kz
复制这段内容后打开百度网盘手机App,操作更方便哦
下载上面安装包,然后在自己服务器上按照下面步骤安装MQ
# 1.将rabbitmq安装包上传到linux系统中
erlang-22.0.7-1.el7.x86_64.rpm
rabbitmq-server-3.7.18-1.el7.noarch.rpm
# 2.安装Erlang依赖包
rpm -ivh erlang-22.0.7-1.el7.x86_64.rpm
# 3.安装RabbitMQ安装包(需要联网)
yum install -y rabbitmq-server-3.7.18-1.el7.noarch.rpm
注意:默认安装完成后配置文件模板在:/usr/share/doc/rabbitmq-server-3.7.18/rabbitmq.config.example目录中,需要
将配置文件复制到/etc/rabbitmq/目录中,并修改名称为rabbitmq.config
# 4.复制配置文件
cp /usr/share/doc/rabbitmq-server-3.7.18/rabbitmq.config.example /etc/rabbitmq/rabbitmq.config
# 5.查看配置文件位置
ls /etc/rabbitmq/rabbitmq.config
# 6.修改配置文件(参见下图:)
vim /etc/rabbitmq/rabbitmq.config
然后将61行的前面两个%%
和尾部的,
去掉
![](/qrcode.jpg)
最后启动MQ的Web插件支持
# 7.执行如下命令,启动rabbitmq中的插件管理
rabbitmq-plugins enable rabbitmq_management
出现如下说明:
Enabling plugins on node rabbit@localhost:
rabbitmq_management
The following plugins have been configured:
rabbitmq_management
rabbitmq_management_agent
rabbitmq_web_dispatch
Applying plugin configuration to rabbit@localhost...
The following plugins have been enabled:
rabbitmq_management
rabbitmq_management_agent
rabbitmq_web_dispatch
set 3 plugins.
Offline change; changes will take effect at broker restart.
# 8.启动RabbitMQ的服务
systemctl start rabbitmq-server
systemctl restart rabbitmq-server
systemctl stop rabbitmq-server
# 9.查看服务状态(见下图:)
systemctl status rabbitmq-server
● rabbitmq-server.service - RabbitMQ broker
Loaded: loaded (/usr/lib/systemd/system/rabbitmq-server.service; disabled; vendor preset: disabled)
Active: active (running) since 三 2019-09-25 22:26:35 CST; 7s ago
Main PID: 2904 (beam.smp)
Status: "Initialized"
CGroup: /system.slice/rabbitmq-server.service
├─2904 /usr/lib64/erlang/erts-10.4.4/bin/beam.smp -W w -A 64 -MBas ageffcbf -MHas ageffcbf -
MBlmbcs...
├─3220 erl_child_setup 32768
├─3243 inet_gethost 4
└─3244 inet_gethost 4
.........
访问服务器地址:http://IP地址:15672/
RabbitMQ消息模型
登录官网可以看到:RabbitMq的消息模型
1. “Hello World!” 直连模式
这个是最简单的消息队列模型,其中生产者和消费者是一一对应的。
消息队列为红色部分,类似于一个邮箱,P就类似与一个邮递员,可以往邮箱中传递邮件,邮箱充当邮件消息的缓冲,C就类似于我们收件人,可以从邮箱中去出邮件。
步骤1:导入MQ相关Maven依赖:
<!-- rabbit mq相关依赖-->
<dependency>
<groupId>com.rabbitmq</groupId>
<artifactId>amqp-client</artifactId>
<version>5.7.2</version>
</dependency>
步骤2:登录rabbitmq服务器,建立队列
步骤3:生产者代码:
@SpringBootTest
class Rabbitmq01ApplicationTests {
//mq的生产者使用单元测试类
@Test
void contextLoads() throws IOException, TimeoutException {
//1.创建连接工厂
ConnectionFactory connectionFactory = new ConnectionFactory();
//2.设置远程rabbitmq-server主机地址
connectionFactory.setHost("192.168.2.163");
//3.设置访问rabbitmq-server的用户名和密码 端口
connectionFactory.setUsername("liuzeyu12a");
connectionFactory.setPassword("809080");
connectionFactory.setPort(5672);
connectionFactory.setVirtualHost("/ems");
//3.创建channel通道
Connection connection = connectionFactory.newConnection();
Channel channel = connection.createChannel();
//4.设置管道参数
//参数1:通道名称 参数2:mq服务重启,是否持久化 参数3:是否独占队列(一般为否) 参数5:是否消费完成 自动删除 参数6:其它额外属性
channel.queueDeclare("test",false,false,false,null);
//5.发布消息
//参数1:交换机 参数2:队列名称 参数3:传递消息的额外设置 参数4:消息内容
channel.basicPublish("","test", MessageProperties.PERSISTENT_TEXT_PLAIN,"hello liuzeyu12a".getBytes());
channel.close();
connection.close();
}
}
步骤4:消费者代码
public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
//1.创建连接工厂
ConnectionFactory connectionFactory = new ConnectionFactory();
//2.设置远程rabbitmq-server主机地址
connectionFactory.setHost("192.168.2.163");
connectionFactory.setVirtualHost("/ems");
connectionFactory.setPort(5672);
connectionFactory.setUsername("liuzeyu12a");
connectionFactory.setPassword("809080");
//3.创建channel通道
Connection connection = connectionFactory.newConnection();
Channel channel = connection.createChannel();
//4.设置通道参数
channel.queueDeclare("test",false,false,false,null);
//消费消息
channel.basicConsume("test", true, new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
//消费的消息在body中
System.out.println("消费的消息:" + new String(body).toString());
}
});
}
观察现象:
先运行生产者后,在队列中我们会看到一条待消费的消息
再次运行消费者
看到存到消息队列中的消息被打印出来了,也就是被消费了,再次查看队列的消息
需要注意的有:
- 生产者和消费者建立通道的时候,参数必须一一对应
- 消费者一般是没有去关闭连接的,因为它要随时监听队列中是否有消息,紧接着消费它
- 如果想要看到消费者打印消息在控制台,就要避免将消费者写在
@Test
中使用,原因是springboot单元测试不支持多线程测试,回调函数可能还没调用到主线程已经结束。
1.1 工具类封装
看到生产者和消费者的代码出现很多的冗余,封装一下代码!
//类似与JDBC
public class RabbitMqUtils {
private static ConnectionFactory connectionFactory;
static {
//1.创建连接工厂
connectionFactory = new ConnectionFactory();
//2.设置远程rabbitmq-server主机地址
connectionFactory.setHost("192.168.2.163");
//3.设置访问rabbitmq-server的用户名和密码 端口
connectionFactory.setUsername("liuzeyu12a");
connectionFactory.setPassword("809080");
connectionFactory.setPort(5672);
connectionFactory.setVirtualHost("/ems");
}
//获取连接
public static Connection getConnect() {
try {
return connectionFactory.newConnection();
}catch (Exception e){
e.printStackTrace();
}
return null;
}
public static void close(Channel channel,Connection connection){
try {
if(channel != null){
channel.close();
}
if(connection != null){
connection.close();
}
}catch (Exception e){
e.printStackTrace();
}
}
}
使用工具类重新改造生产者:
@SpringBootTest
class Rabbitmq01ApplicationTests {
//mq的生产者使用单元测试类
@Test
void contextLoads() throws IOException, TimeoutException {
Connection connection = RabbitMqUtils.getConnect();
Channel channel = connection.createChannel();
//4.设置管道参数
//参数1:通道名称 参数2:mq服务重启,是否持久化 参数3:是否独占队列(一般为否) 参数5:是否消费完成 自动删除 参数6:其它额外属性
channel.queueDeclare("test",false,false,false,null);
//发布消息
//参数1:交换机 参数2:队列名称 参数3:传递消息的额外设置 参数4:消息内容
channel.basicPublish("","test", MessageProperties.PERSISTENT_TEXT_PLAIN,"hello liuzeyu12a".getBytes());
//3.释放资源
RabbitMqUtils.close(channel,connection);
}
}
使用工具类重新改造消费者:
//监听队列消费消息
public class Consumer {
public static void main(String[] args) throws IOException {
//1.创建连接
Connection connection = RabbitMqUtils.getConnect();
Channel channel = connection.createChannel();
//2.设置通道参数
channel.queueDeclare("test",false,false,false,null);
//3.消费消息
channel.basicConsume("test", true, new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
//消费的消息在body中
System.out.println("消费的消息:" + new String(body).toString());
}
});
}
}
1.2 细节
- 持久化
如果我们的生产者在队列中产生了消息,消费者还没消费,rabbitmq服务器重启了,消息和队列就会被清空,如何避免这一情况呢?
简单解决办法有两个
将之前的队列删掉,用代码来生成新的队列,但要注意的是将channel.queueDeclare
的第二个参数设置成true,表示该队列支持持久化
//mq的生产者使用单元测试类
@Test
void contextLoads() throws IOException, TimeoutException {
Connection connection = RabbitMqUtils.getConnect();
Channel channel = connection.createChannel();
//4.设置管道参数
//参数1:通道名称 参数2:mq服务重启,是否持久化 参数3:是否独占队列(一般为否) 参数5:是否消费完成 自动删除 参数6:其它额外属性
channel.queueDeclare("test",true,false,false,null);
//发布消息
//参数1:交换机 参数2:队列名称 参数3:传递消息的额外设置 参数4:消息内容
channel.basicPublish("","test", MessageProperties.PERSISTENT_TEXT_PLAIN,"hello liuzeyu12a".getBytes());
//3.释放资源
RabbitMqUtils.close(channel,connection);
}
还有一个方法就是直接用官网提供的界面的创建队列,并且选择持久化
配置完消息队列的持久化了,测试一下重启[root@localhost ~]# systemctl restart rabbitmq-server
发现队列虽然持久化了,但是消息并没有被持久化,这样看起来也毫无意义了,所以我们也可以消息持久化。
@Test
void contextLoads() throws IOException, TimeoutException {
Connection connection = RabbitMqUtils.getConnect();
Channel channel = connection.createChannel();
//4.设置管道参数
//参数1:通道名称 参数2:mq服务重启,是否持久化 参数3:是否独占队列(一般为否) 参数5:是否消费完成 自动删除 参数6:其它额外属性
channel.queueDeclare("test",true,false,false,null);
//发布消息
//参数1:交换机 参数2:队列名称 参数3:传递消息的额外设置 参数4:消息内容
channel.basicPublish("","test", MessageProperties.PERSISTENT_TEXT_PLAIN,"hello liuzeyu12a".getBytes());
//3.释放资源
RabbitMqUtils.close(channel,connection);
}
channel.basicPublish()
的第三个参数设置为MessageProperties.PERSISTENT_TEXT_PLAIN,表示的是持久化消息。
channel.basicPublish()
第三个参数是是否独占队列,一般我们都设置为false表示不同通道都可以向队列中发送消息,不独占。
channel.basicPublish()
还有第四个参数是是否在消费者消费完之后,删除队列,看具体的业务场景。
2. work Queue
任务队列跟第一种的直连相比,会多出多个消费者,这种场景一般出现在,生产者的生产消息速度 > 消费者的消费速度,如果没有多个消费者一起消费队列中的消息,会引起队列的堵塞,消息的传递的也会不及时。
2.1 轮询
实例:
- 生产者:
//生产者
public class Provider {
public static void main(String[] args) throws IOException {
//1获取连接
Connection connect = RabbitMqUtils.getConnect();
//2.创建通道
Channel channel = connect.createChannel();
//3.设置通道参数
channel.queueDeclare("test",true,false,false,null);
//4.发送消息
for (int i = 1; i <= 20; i++) {
channel.basicPublish("", "test",MessageProperties.PERSISTENT_TEXT_PLAIN,(i +"->meg:").getBytes());
}
RabbitMqUtils.close(channel,connect);
}
}
- 消费者1:
//消费者1
public class Consumer1 {
public static void main(String[] args) throws IOException {
//1.创建连接
Connection connect = RabbitMqUtils.getConnect();
//2.创建通道
Channel channel = connect.createChannel();
//3.配置通道参数
channel.queueDeclare("test",true,false,false,null);
//
channel.basicConsume("test",true,new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
System.out.println("消费者1:"+new String(body));
}
});
}
}
- 消费者2:
//消费者2
public class Consumer2 {
public static void main(String[] args) throws IOException {
//1.获取连接
Connection connect = RabbitMqUtils.getConnect();
//2.获取通道
Channel channel = connect.createChannel();
//3.配置通道参数
channel.queueDeclare("test",true,false,false,null);
//4.接收消息
channel.basicConsume("test",true,new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
System.out.println("消费者2:"+new String(body));
}
});
}
}
为了测试的准确性,先把之前的队列清空。
依次运行消费者1,消费者2。
然后运行生产者->
观察结果消费者1运行结果:
消费者2:
发现消费者1和2是轮询从队列中消费,为了进一步验证这个结果,可以染消费者2的消费速度放慢
public class Consumer2 {
public static void main(String[] args) throws IOException {
//1.获取连接
Connection connect = RabbitMqUtils.getConnect();
//2.获取通道
Channel channel = connect.createChannel();
//3.配置通道参数
channel.queueDeclare("test",true,false,false,null);
//4.接收消息
channel.basicConsume("test",true,new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
try {
Thread.sleep(1000);
}catch (Exception e){
e.printStackTrace();
}
System.out.println("消费者2:"+new String(body));
}
});
}
}
结果仍然是依次交替消费数据,与我们的猜想一致。
2.2 消息自动确认机制
消息自动确认机制 指的是消费者消费完消息之后向rabbitmq发送确认消息消费的一种机制。
我们的任务消息队列,消费者默认是轮询消费队列中的消息,无论消费者他们各自的消费速度如何,都将分配同样个数的消息。
但是正常情况下,我们又希望消费快的能够多消费一点消息,消费慢的少消费一点,达到能者多劳的目的。
要想实现这一目的,我们就必须关闭掉消息的自动确认机制。
生产者:
//生产者
public class Provider {
public static void main(String[] args) throws IOException {
//1获取连接
Connection connect = RabbitMqUtils.getConnect();
//2.创建通道
Channel channel = connect.createChannel();
//3.设置通道参数
channel.queueDeclare("test",true,false,false,null);
//4.发送消息
for (int i = 1; i <= 20; i++) {
channel.basicPublish("", "test",MessageProperties.PERSISTENT_TEXT_PLAIN,(i +"->meg:").getBytes());
}
RabbitMqUtils.close(channel,connect);
}
}
消费者1:
//消费者1
public class Consumer1 {
public static void main(String[] args) throws IOException {
//1.创建连接
Connection connect = RabbitMqUtils.getConnect();
//2.创建通道
Channel channel = connect.createChannel();
//3.配置通道参数
channel.queueDeclare("test",true,false,false,null);
//
channel.basicQos(1); //每次获取一条消息
channel.basicConsume("test",false,new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
System.out.println("消费者1:"+new String(body));
channel.basicAck(envelope.getDeliveryTag(),false); //手动确认
}
});
}
}
消费者2:
//消费者2
public class Consumer2 {
public static void main(String[] args) throws IOException, InterruptedException {
//1.获取连接
Connection connect = RabbitMqUtils.getConnect();
//2.获取通道
Channel channel = connect.createChannel();
//3.配置通道参数
channel.queueDeclare("test",true,false,false,null);
channel.basicQos(1);//每次获取一条消息
//4.接收消息
channel.basicConsume("test",false,new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
try {
Thread.sleep(1000);
}catch (Exception e){
e.printStackTrace();
}
System.out.println("消费者2:"+new String(body));
channel.basicAck(envelope.getDeliveryTag(),false); //手动确认消费
}
});
}
}
现象:
做到了能者多劳!
自动确认机制存在的问题:
可以避免消息丢失的问题,如果消费者1和消费者2同时在消费消息,默认情况下用的是自动确认,如果消费者1宕机了,消息才消费到一半,但是队列已经得到它的自动确认标志,认为已经消费完了,这就会造成消息丢失。
关闭自动确认机制有一个很好的好处是:
但是关闭了自动确认机制就会等到我们具体的业务执行完,我们可以手动提交消息消费标志给mq服务器,告诉它消息已经执行完,这中间执行我们业务即使宕机,消息也不会丢失,因为我们是在最后告诉mq消息被消费完了,具体业务并没有执行完,消息还在,服务器我们就可以再利用消息执行相关具体业务。
3. 广播 fanout模式
这是一种广播的消息模型,生产者生产的消息,要想发送给多个消费者,这中间要借助于交换机,多个交换机与队列进行绑定,那么每个队列中都会接收到生产者发送的消息,与队列绑定的消费者就能接收到信息。
这个和计算机网络中的交换机实现的是相同的原理,多个路由器与交换机绑定,那么才会实现广播路由的分发。
编码实现:
//生产者广播
public class Provider {
public static void main(String[] args) throws IOException {
//创建连接
Connection connect = RabbitMqUtils.getConnect();
//获取通道
Channel channel = connect.createChannel();
//通道绑定交换机 参数1:交换机名称 参数2:类型为广播
channel.exchangeDeclare("log","fanout");
//发送消息到交换
channel.basicPublish("log","",null,"你好交换机".getBytes());
RabbitMqUtils.close(channel,connect);
}
}
观察交换机
运行生产者代码:
多出一个log的交换机,使用的是广播的类型 fanout
我们建立多个消费者,绑定这台交换机,测试能否收到消息。
消费者1:
//消费者1
public class Consumer1 {
public static void main(String[] args) throws IOException {
//1.创建连接
Connection connect = RabbitMqUtils.getConnect();
//2.创建通道
Channel channel = connect.createChannel();
//3.绑定交换机
channel.exchangeDeclare("log","fanout");
//4.创建临时队列
String queue = channel.queueDeclare().getQueue();
//5.将临时队列绑定交换机
channel.queueBind(queue,"log","");
//6.接收消息
channel.basicConsume(queue,true,new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
System.out.println("消费者1:"+ new String(body));
}
});
}
}
消费者2:
//消费者2
public class Consumer2 {
public static void main(String[] args) throws IOException {
//1.创建连接
Connection connect = RabbitMqUtils.getConnect();
//2.创建通道
Channel channel = connect.createChannel();
//3.绑定交换机
channel.exchangeDeclare("log","fanout");
//4.创建临时队列
String queue = channel.queueDeclare().getQueue();
//5.将临时队列绑定交换机
channel.queueBind(queue,"log","");
//6.接收消息
channel.basicConsume(queue,true,new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
System.out.println("消费者2:"+ new String(body));
}
});
}
}
优先启动消费者1,2,生产者发送消息
消费者生成了两个临时队列
接着运行生产者
可以发现做到了一个消息多个消费者消费的目的,适用的场景:登录/注册这一举动可以送积分,送奖品,发布订阅消息。
当停止消费者,临时队列也跟着消失
4. 直连路由
直连路由路由的工作模式:Receiving messages selectively
,翻译过来也就是,消费者可以选择性的接收到生产者的消息。
这相比广播的方式会灵活点,广播就是不要消费要不要,只要和消费者的队列和交换机绑定,那么消费者就一定会接收到消息。
直连路由的业务场景可以用于日志系统,如上图,可以把C1比作磁盘,C2比作控制台,我们系统产生的日志如果error类型要存放到磁盘的同时在控制台打印,而info,warn,error都要在控制台打印,这个时候广播的方式已经控制不了了,就要适用新的消息模型,直连路由的方式。
生产者:
//具有路由形式的直连
public class Provider {
public static void main(String[] args) throws IOException {
//创建连接
Connection connect = RabbitMqUtils.getConnect();
//机获取通道
Channel channel = connect.createChannel();
//配置通道配置
channel.exchangeDeclare("routing_direct","direct");
//获取临时队列
String queue = channel.queueDeclare().getQueue();
//发送消息
String routingKey1 = "info";
channel.basicPublish("routing_direct",routingKey1,
null,("这是一种直连路由,routingKey:"+routingKey1).getBytes());
String routingKey2 = "error";
channel.basicPublish("routing_direct",routingKey2,
null,("这是一种直连路由,routingKey:"+routingKey2).getBytes());
String routingKey3 = "warning";
channel.basicPublish("routing_direct",routingKey3,
null,("这是一种直连路由,routingKey:"+routingKey3).getBytes());
//是否资源
RabbitMqUtils.close(channel,connect);
}
}
消费者C1:
//具有路由形式的直连
public class Consumer1 {
public static void main(String[] args) throws IOException {
//获取连接
Connection connect = RabbitMqUtils.getConnect();
//获取通道
Channel channel = connect.createChannel();
//配置通道
channel.exchangeDeclare("routing_direct","direct");
//创建临时队列
String queue = channel.queueDeclare().getQueue();
//绑定交换
channel.queueBind(queue,"routing_direct","error");
//消费消息
channel.basicConsume(queue,true,new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
System.out.println("消费者1:"+ new String(body));
}
});
}
}
消费者C2:
//具有路由形式的直连
public class Consumer2 {
public static void main(String[] args) throws IOException {
//获取连接
Connection connect = RabbitMqUtils.getConnect();
//获取通道
Channel channel = connect.createChannel();
//配置通道
channel.exchangeDeclare("routing_direct","direct");
//创建临时队列
String queue = channel.queueDeclare().getQueue();
//绑定交换
channel.queueBind(queue,"routing_direct","info");
channel.queueBind(queue,"routing_direct","warning");
channel.queueBind(queue,"routing_direct","error");
//消费消息
channel.basicConsume(queue,true,new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
System.out.println("消费者1:"+ new String(body));
}
});
}
}
输出:
通过现象也可以看出,我们可以通过交换机绑定的路由key 和队列与交换机绑定的路由key产生对应关系,来决定消息的指向性发送。
5. 动态路由
动态路由是直连路由的升级版,更为灵活的的控制路由的分发,通过通配符来控制路由的下发。
通配符有两种:
*:表示user.*代表的是匹配user.[一个单词],user.findAll,userdelete都会被匹配上
#:表示user.#代表的是匹配user.[0个单词/多个单词],user,user.findAll,userdelete,user.findAll.fuck都会被匹配上
生产者:
//动态路由 Provider
public class Provider {
public static void main(String[] args) throws IOException {
//创建连接
Connection connect = RabbitMqUtils.getConnect();
//获取通道
Channel channel = connect.createChannel();
//发送消息
String routingKey = "user.findAll";
channel.basicPublish("topics",routingKey,null,("动态路由,通配符:"+routingKey).getBytes());
//释放资源
RabbitMqUtils.close(channel,connect);
}
}
消费者1:
//动态路由 消费者1
public class Consumer1 {
public static void main(String[] args) throws IOException {
//创建连接
Connection connect = RabbitMqUtils.getConnect();
//获取通道
Channel channel = connect.createChannel();
//创建临时队列
String queue = channel.queueDeclare().getQueue();
//绑定交换机
channel.exchangeDeclare("topics","topic");
channel.queueBind(queue,"topics","user.*");
//接收消息
channel.basicConsume(queue,true,new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
System.out.println("消费者1:" + new String(body));
}
});
}
}
消费者1会接收到消息,因为它匹配到了生产者的key,user.findAll,消费者的user.*匹配到一个单词了,将user.*改为user.#同样可以匹配到。
如果生产者绑定的是
String routingKey = "user.findAll.fuck";
消费者则必须适用
//绑定交换机
channel.exchangeDeclare("topics","topic");
channel.queueBind(queue,"topics","user.#");
才可以匹配上。
6. SpringBoot整合RabbitMQ
实际开发中,springboot是最常用的开发平台,我们也要学会在SpringBoot平台上适用RabbitMQ
第一步:导入相关依赖(建议Springboot适用2.1.x)
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
第二步:创建生产者测试类
@SpringBootTest(classes = Rabbitmq01Application.class)
@RunWith(SpringRunner.class)
public class SpringbootMQ {
@Autowired
private RabbitTemplate rabbitTemplate;
}
6.1 HELLO WORLD
其中RabbitTemplate 适用操作MQ的一套模板,可以简化我们操作MQ
生产者:
//HELLO WORLD
@Test
public void hello(){
//参数1:队列名称 参数2:要发送队列的内容
rabbitTemplate.convertAndSend("boot-mq","springboot hello world");
}
当我们运行完生产者后,并不会在队列中生成队列,因为使用rabbitTemplate创建队列并发送消息,此时并没有消费者绑定改队列,所以我们需要一个消费者:
@Component
public class Consumer {
@RabbitHandler
@RabbitListener(queuesToDeclare = @Queue("boot-mq")) //也可以用在类上
public void receive1(String message){
System.out.println("message:"+message);
}
}
最后我们运行一下生产者
消息被消费的信息打印在了控制台。
6.2 work Queue
生产者:
//work Queue
@Test
public void work(){
//参数1:队列名称 参数2:要发送队列的内容
rabbitTemplate.convertAndSend("work","springboot hello work Queue");
}
消费者:
@Component
public class Consumer1 {
// @RabbitHandler
@RabbitListener(queuesToDeclare = @Queue("work"))
public void receive1(String message){
System.out.println("message1:"+message);
}
// @RabbitHandler
@RabbitListener(queuesToDeclare = @Queue("work"))
public void receive2(String message){
System.out.println("message2:"+message);
}
}
生产运行完之后发现,消息会被消费者1或2任意一个消费,类似,我们生产者创建多个消息
@Test
public void work(){
//参数1:队列名称 参数2:要发送队列的内容
for (int i = 0; i < 10; i++) {
rabbitTemplate.convertAndSend("work","springboot hello work Queue");
}
}
再次运行
多个消费者还是进行轮询的消费,符合我们预期的场景。
6.3. 广播 fanout模式
生产者:
//fanout
@Test
public void fanout(){
rabbitTemplate.convertAndSend("logs","","springboot fanout Queue");
}
消费者:
@Component
public class FanoutConsumer {
//创建临时队列
@RabbitListener(bindings = @QueueBinding(
value = @Queue, //创建临时队列
exchange = @Exchange(name = "logs",type = "fanout")
))
public void receive1(String message){
System.out.println("message1:"+message);
}
@RabbitListener(bindings = @QueueBinding(
value = @Queue, //创建临时队列
exchange = @Exchange(name = "logs",type = "fanout")
))
public void receive2(String message){
System.out.println("message2:"+message);
}
}
运行生产者,发现消息同时被两个消费者消费了。
6.4 直连路由
生产者:
//direct
@Test
public void direct(){
rabbitTemplate.convertAndSend("direct","info","springboot direct Queue");
}
发送key = info的消息
消费者:
@Component
public class ConsumerDirect {
@RabbitListener(bindings = @QueueBinding(
value = @Queue, //临时队列
key = {
"error","warn","info"},
exchange = @Exchange(name = "direct",type = "direct")
))
public void receive1(String message){
System.out.println("message1:"+message);
}
@RabbitListener(bindings = @QueueBinding(
value = @Queue, //临时队列
key = {
"error"},
exchange = @Exchange(name = "direct",type = "direct")
))
public void receive2(String message){
System.out.println("message2:"+message);
}
}
此时只会有消费者1才能消费消息
6.5 动态路由
生产者:
@Test
public void topics(){
rabbitTemplate.convertAndSend("topics","product.save","springboot topics Queue");
}
消费者的key以product开头都可以匹配到
消费者
@Component
public class ConsumerTopics {
@RabbitListener(bindings = @QueueBinding(
value = @Queue, //临时队列
exchange = @Exchange(name = "topics",type = "topic"),
key = {
"user.*"}
))
public void receive1(String message){
System.out.println("message1:"+message);
}
@RabbitListener(bindings = @QueueBinding(
value = @Queue, //临时队列
exchange = @Exchange(name="topics",type = "topic"),
key = {
"product.#"}
))
public void receive2(String message){
System.out.println("message2:"+message);
}
}
消费者2可以接收到消息
7. RabbitMQ应用场景
集群环境会使用Docker来部署,顺便自己再复习点Docker知识,因为用Docker来部署集群用于学习真的很方便。
8. RabbitMQ集群
8.1 RabbitMQ普通集群
普通集群:以三台服务器为例,主Master节点,两个从Slave,这边的RabbitMQ普通集群并不支持高可用的场景,它的工作原理是:主Master的交换机会和从的从交换机同步,它们所绑定的队列也会进行同步,但是队列中的消息并不会同步,当我们的消费者要消费消息的时候,有两个从服务器向主服务器拿。一旦主宕机,从服务器的消息就可能丢失,而且并不会进行故障切换,从服务器不会变成主服务器,做不到高可用!
实验的场景需要搭建服务器集群,一般的物理机同时运行三个服务器会卡,于是可以在docker上面部署集群。
可以参考这位大哥的集群部署教程:Docker环境RabbitMQ集群部署
搭建几个基本步骤:
- docker安装:
# step 1: 安装必要的一些系统工具
sudo yum install -y yum-utils
# Step 2: 添加软件源信息
sudo yum-config-manager --add-repo http://mirrors.aliyun.com/docker-ce/linux/centos/docker-ce.repo
# Step 3: 更新并安装 Docker-CE(社区版)
sudo yum makecache fast
sudo yum -y install docker-ce
# Step 4: 开启Docker服务
sudo service docker start
- rabbitmq安装
#拉取镜像
# docker pull rabbitmq:3.8.3-management
#创建映射数据卷目录
#mkdir rabbitmqcluster
#cd rabbitmqcluster/
#mkdir rabbitmq01 rabbitmq02 rabbitmq03
[root@localhost rabbitmqcluster]# pwd
/home/liuzeyu12a/rabbitmqcluster
#启动三个mq容器
#docker run -d --hostname rabbitmq01 --name rabbitmqCluster01 -v /home/liuzeyu12a/rabbitmqcluster/rabbitmq01:/var/lib/rabbitmq -p 15672:15672 -p 5672:5672 -e RABBITMQ_ERLANG_COOKIE='rabbitmqCookie' rabbitmq:3.8.3-management
#docker run -d --hostname rabbitmq02 --name rabbitmqCluster02 -v /home/liuzeyu12arabbitmqcluster/rabbitmq02:/var/lib/rabbitmq -p 15673:15672 -p 5673:5672 -e RABBITMQ_ERLANG_COOKIE='rabbitmqCookie' --link rabbitmqCluster01:rabbitmq01 rabbitmq:3.8.3-management
#docker run -d --hostname rabbitmq03 --name rabbitmqCluster03 -v /home/liuzeyu12a/rabbitmqcluster/rabbitmq03:/var/lib/rabbitmq -p 15674:15672 -p 5674:5672 -e RABBITMQ_ERLANG_COOKIE='rabbitmqCookie' --link rabbitmqCluster01:rabbitmq01 --link rabbitmqCluster02:rabbitmq02 rabbitmq:3.8.3-management
hostname 设置容器的主机名
RABBITMQ_ERLANG_COOKIE 节点认证作用,部署集成时 需要同步该值。
- 集群配置
# 进入主1容器内部
docker exec -it rabbitmqCluster01 bash
#重置内部配置
#rabbitmqctl stop_app
#rabbitmqctl reset
#rabbitmqctl start_app
#exit
#进入从2内部
#docker exec -it rabbitmqCluster02 bash
#rabbitmqctl stop_app
#rabbitmqctl reset
#rabbitmqctl join_cluster --ram rabbit@rabbitmq01
#rabbitmqctl start_app
#exit
#进入从3内部
#docker exec -it rabbitmqCluster03 bash
#rabbitmqctl stop_app
#rabbitmqctl reset
#rabbitmqctl join_cluster --ram rabbit@rabbitmq01
#rabbitmqctl start_app
#exit
配置完后可以进入:
搭建起来正常的就是上图。
我在配置集群的时候遇到电脑蓝屏重启了一下服务器,发现主服务器可以起来,从服务器也都可以起来,但是两台从服务器就是就是加不到主服务器中,报错信息:
root@rabbitmq03:/# rabbitmqctl join_cluster --ram rabbit@rabbitmq01
Clustering node rabbit@rabbitmq03 with rabbit@rabbitmq01
Error:
{
:inconsistent_cluster, 'Node rabbit@rabbitmq01 thinks it\'s clustered with node rabbit@rabbitmq03, but rabbit@rabbitmq03 disagrees'}
网上查阅了很多资料,最简单粗暴的方式就是将映射数据卷目录删除掉,然后重新创建,重新启动服务器,就可以重新配置主从了。
- 访问服务器
主:http://192.168.2.163:15672/
从1:http://192.168.2.163:15673/
从2:http://192.168.2.163:15674/
生产者:
@Test
public void contextLoads() throws IOException, TimeoutException {
Connection connection = RabbitMqUtils.getConnect();
Channel channel = connection.createChannel();
//4.设置管道参数
//参数1:通道名称 参数2:mq服务重启,是否持久化 参数3:是否独占队列(一般为否) 参数5:是否消费完成 自动删除 参数6:其它额外属性
channel.queueDeclare("test",true,false,false,null);
//发布消息
//参数1:交换机 参数2:队列名称 参数3:传递消息的额外设置 参数4:消息内容
channel.basicPublish("","test", MessageProperties.PERSISTENT_TEXT_PLAIN,"hello liuzeyu12a".getBytes());
//3.释放资源
RabbitMqUtils.close(channel,connection);
}
向主节点发送一个消息后:
从1
从2
消费者:
public static void main(String[] args) throws IOException {
//1.创建连接
Connection connection = RabbitMqUtils.getConnect();
Channel channel = connection.createChannel();
//2.设置通道参数
channel.queueDeclare("test",true,false,false,null);
//3.消费消息
channel.basicConsume("test", true, new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
//消费的消息在body中
System.out.println("消费的消息:" + new String(body).toString());
}
});
}
消费掉消息后,各个服务器队列中消息也会同步被消费。
现在我们重新向主服务器发送一条消息,然后down掉它。
[root@localhost rabbitmqcluster]# docker exec -it rabbitmqCluster01 bash
root@rabbitmq01:/# rabbitmqctl stop_app
Stopping rabbit application on node rabbit@rabbitmq01 ...
root@rabbitmq01:/#
此时从服务器
最后去尝试消费消息
显然,此时从服务器连接从主服务器取的消息,因此消息消费失败,这也就验证了普通集群无法高可用的现象。
8.2 RabbitMQ镜像集群
镜像集群的工作原理主要还是镜像发挥的作用,在集群中的服务器配置了镜像,主从的镜像就会同步。队列和消息也会同步到各个镜像中,当主节点宕机后,主从就会发生切换,从节点变成主节点,从而实现高可用。
LVS和HaProxy都是防止服务器宕机导致服务不可用,做的负载均衡策略。
#策略说明
rabbitmqctl set_policy [-p <vhost>] [--priority <priority>] [--apply-to <apply-to>] <name> <pattern> <definition>
-p <vhost>:可选参数,正对虚拟主机的队列进行设置
name:策略名称
Pattern:队列的匹配模式(正则)
Definition:镜像定义,包括三个部分ha-mode, ha-params, ha-sync-mode
ha-mode:表示镜像队列的模式有(all/exactly/nodes)
all:集群中所有的节点进行镜像
exactly:指定个数的节点上进行镜像,节点个数由ha-params决定
nodes:指定节点上进行镜像,节点名称由ha-params指定
ha-params:用来配置ha-mode的参数
ha-sync-mode:进行队列中的消息的同步方式,有效值automatic和manual,一般配置automatic
priority:可选参数,优先级
查看当下的服务器集群从是否配置策略
root@rabbitmq01:/# rabbitmqctl list_policies
Listing policies for vhost "/" ...
root@rabbitmq01:/#
配置策略
向主节点发送消息后(也可以先配置策略再创建队列):
root@rabbitmq01:/# rabbitmqctl set_policy ha-all '^test' '{"ha-mode":"all","ha-sync-mode":"automatic"}'
Setting policy "ha-all" for pattern "^test" to "{"ha-mode":"all","ha-sync-mode":"automatic"}" with priority "0" for vhost "/" ...
root@rabbitmq01:/# rabbitmqctl list_policies
Listing policies for vhost "/" ...
vhost name pattern apply-to definition priority
/ ha-all ^test all {
"ha-mode":"all","ha-sync-mode":"automatic"} 0
root@rabbitmq01:/#
只有test开头的队列才可以镜像自动同步。
测试现象:
发现并没有生效,经过排查发现没有指定虚拟主机配置策略,所有改主机下的队列就没有被匹配到。
重新配置策略:
root@rabbitmq01:/# rabbitmqctl set_policy -p '/ems' ha-all '^test' '{"ha-mode":"all","ha-sync-mode":"automatic"}'
Setting policy "ha-all" for pattern "^test" to "{"ha-mode":"all","ha-sync-mode":"automatic"}" with priority "0" for vhost "/ems" ...
root@rabbitmq01:/# rabbitmqctl list_policies
Listing policies for vhost "/" ...
root@rabbitmq01:/#
主:
从1:
从2
现在我们down掉主节点!
从1
变成了主节点,并且状态还是up的
从2
尝试消费一下从2服务器
消费者要记得修改端口号,否则会获取连接失败。
此时的主节点消息也就被消费的了
如果此时把,之前down掉的主节点重新开启,它是否重新变成主节点?
root@rabbitmq01:/# rabbitmqctl start_app
Starting node rabbit@rabbitmq01 ...
completed with 3 plugins.
root@rabbitmq01:/#
可见它是不可能再变成主节点了
清除策略
root@rabbitmq01:/# rabbitmqctl clear_policy ha-all -p '/ems'
Clearing policy "ha-all" on vhost "/ems" ...
root@rabbitmq01:/# rabbitmqctl list_policies
Listing policies for vhost "/" ...
root@rabbitmq01:/#
队列的状态也就恢复成之前的普通集群样子