RabbitMQ基础总结
Posted Serendipity sn
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了RabbitMQ基础总结相关的知识，希望对你有一定的参考价值。
RabbitMQ
目录
1.对MQ的介绍
-
说明是MQ
MQ(message queue)，从字面意思上看，本质是个队列，FIFO 先入先出，只不过队列中存放的内容是
message 而已，还是一种跨进程的通信机制，用于上下游传递消息。在互联网架构中，MQ 是一种非常常
见的上下游“逻辑解耦+物理解耦”的消息通信服务。使用了 MQ 之后，消息发送上游只需要依赖 MQ，不
用依赖其他服务。
-
MQ的好处
-
流量消峰
举个例子，如果订单系统最多能处理一万次订单，这个处理能力应付正常时段的下单时绰绰有余，正常时段我们下单一秒后就能返回结果。但是在高峰期，如果有两万次下单操作系统是处理不了的，只能限制订单超过一万后不允许用户下单。使用消息队列做缓冲，我们可以取消这个限制，把一秒内下的订单分散成一段时间来处理，这时有些用户可能在下单十几秒后才能收到下单成功的操作，但是比不能下单的体验要好。
-
应用解耦
以电商应用为例，应用中有订单系统、库存系统、物流系统、支付系统。用户创建订单后，如果耦合调用库存系统、物流系统、支付系统，任何一个子系统出了故障，都会造成下单操作异常。当转变成基于消息队列的方式后，系统间调用的问题会减少很多，比如物流系统因为发生故障，需要几分钟来修复。在这几分钟的时间里，物流系统要处理的内存被缓存在消息队列中，用户的下单操作可以正常完成。当物流系统恢复后，继续处理订单信息即可，用户感受不到物流系统的故障，提升系统的可用性。
-
-
异步处理
有些服务间调用是异步的，例如 A 调用 B，B 需要花费很长时间执行，但是 A 需要知道 B 什么时候可以执行完，以前一般有两种方式，A 过一段时间去调用 B 的查询 api 查询。或者 A 提供一个 callback api， B 执行完之后调用 api 通知 A 服务。这两种方式都不是很优雅，使用消息总线，可以很方便解决这个问题，A 调用 B 服务后，只需要监听 B 处理完成的消息，当 B 处理完成后，会发送一条消息给 MQ，MQ 会将此消息转发给 A 服务。这样 A 服务既不用循环调用 B 的查询 api，也不用提供 callback api。同样 B 服务也不用做这些操作。A 服务还能及时的得到异步处理成功的消息。
2.RabbitMQ的六种模式 及工作原理
工作模式
依次是：hello world ，工作模式，发布订阅模式，路由模式，主题模式，发布确认模式
工作原理
Binding：exchange 和 queue 之间的虚拟连接，binding 中可以包含 routing key，Binding 信息被保
存到 exchange 中的查询表中，用于 message 的分发依据
依赖
<!--rabbitmq 依赖客户端-->
<dependency>
<groupId>com.rabbitmq</groupId>
<artifactId>amqp-client</artifactId>
<version>5.8.0</version>
</dependency>
3.hello world队列
-
生产者
public class Producer
//建立队列
private static final String QUEUE_NAME="hello";
public static void main(String[] args)
//创建连接工场
ConnectionFactory factory=new ConnectionFactory();
factory.setHost("127.0.0.1");
factory.setUsername("guest");
factory.setUsername("guest");
try
//建立连接和信道
//channel 实现了自动 close 接口 自动关闭 不需要显示关闭
Connection connection=factory.newConnection();
Channel channel=connection.createChannel();
/**
* 生成一个队列,并将信道和队列连接
* 1.队列名称
* 2.队列里面的消息是否持久化 默认消息存储在内存中
* 3.该队列是否只供一个消费者进行消费 是否进行共享 true 可以多个消费者消费
* 4.是否自动删除 最后一个消费者端开连接以后 该队列是否自动删除 true 自动删除
* 5.其他参数
*/
channel.queueDeclare(QUEUE_NAME,false,false,false,null);
String message="hello world";
/**
* 发送一个消息
* 1.发送到那个交换机
* 2.路由的 key 是哪个
* 3.其他的参数信息
* 4.发送消息的消息体
*/
channel.basicPublish("",QUEUE_NAME,null,message.getBytes());
System.out.println("消息发送成功");
catch (Exception e)
e.printStackTrace();
-
消费者
public class Consumer
//定义队列名
private static final String QUEUE_NAME="hello";
public static void main(String[] args)
//建立连接和信道
try
ConnectionFactory factory=new ConnectionFactory();
factory.setHost("127.0.0.1");
factory.setUsername("guest");
factory.setPassword("guest");
Connection connection=factory.newConnection();
Channel channel=connection.createChannel();
System.out.println("等待接收消息");
/**
*1.同一个会话， consumerTag 是固定的 可以做此会话的名字， deliveryTag 每次接收消息+1，可以做此消息处理通道的名字。
*2.包含消息的字节形式的类
*/
DeliverCallback deliverCallback=(consumerTag,delivery)->
String message=new String(delivery.getBody());
System.out.println(message);
;
CancelCallback cancelCallback=(consumerTag)->
System.out.println("消息消费被取消");
;
/* 消费者消费消息
* 1.消费哪个队列
* 2.消费成功之后是否要自动应答 true 代表自动应答 false 手动应答
* 3.消费者未成功消费的回调
* 4.消费者取消消费的回调
*/
channel.basicConsume(QUEUE_NAME,true,deliverCallback,cancelCallback);
catch (Exception e)
e.printStackTrace();
4.工作队列模式
生产者
public class Producer
public static void main(String[] args) throws IOException, InterruptedException
Channel channel=RabbitMQChannelUtil.getChannel();
if(channel == null)
System.out.println("失败");
return;
channel.queueDeclare(RabbitMQChannelUtil.QUEUE_NAME,false,false,false,null);
int i=0;
while (true)
String message="消息"+i;
i++;
/**
* 发送一个消息
* 1.发送到那个交换机
* 2.路由的 key 是哪个
* 3.其他的参数信息
* 4.发送消息的消息体
*/
channel.basicPublish("",RabbitMQChannelUtil.QUEUE_NAME,null,message.getBytes());
System.out.println(message);
Thread.sleep(500);
消费者
public class Consumer
public static void main(String[] args)
Channel channel=RabbitMQChannelUtil.getChannel();
if(channel == null)
System.out.println("消费失败");
return;
DeliverCallback deliverCallback=(consumerTag, delivery)->
String message=new String(delivery.getBody());
System.out.println(Thread.currentThread().getName()+"消费了"+message);
;
CancelCallback cancelCallback=(consumerTag)->
System.out.println("消息消费被取消");
;
Thread[] threads=new Thread[5];
for (int i = 0; i <threads.length ; i++)
threads[i]=new Thread(()->
try
System.out.println(Thread.currentThread().getName()+"启动等待消费");
channel.basicConsume(RabbitMQChannelUtil.QUEUE_NAME,true,deliverCallback,cancelCallback);
catch (IOException e)
e.printStackTrace();
);
for (int i = 0; i <threads.length ; i++)
threads[i].start();
5.消息应答机制
认识
消费者处理消息时，可能在处理过程中挂掉，那么消息就会丢失为了保证消息在发送过程中不丢失，rabbitmq 引入消息应答机制，消息应答就是:消费者在接收到消息并且处理该消息之后，告诉 rabbitmq 它已经处理了rabbitmq 可以把该消息删除了。
自动应答
消息发送后立即被认为已经传送成功，这种模式需要在高吞吐量和数据传输安全性方面做权衡,因为这种模式如果消息在接收到之前，消费者那边出现连接或者 channel 关闭，那么消息就丢失了
手动应答
- Channel.basicAck(用于肯定确认)
RabbitMQ 已知道该消息并且成功的处理消息，可以将其丢弃了
- Channel.basicNack(用于否定确认)
- Channel.basicReject(用于否定确认)
与 Channel.basicNack 相比少一个参数,不处理该消息了直接拒绝，可以将其丢弃了
Channel.basicNack参数中Multiple(批量应答) 的解释
multiple 的 true 和 false 代表不同意思
- true 代表批量应答 channel 上未应答的消息
比如说 channel 上有传送 tag 的消息 5,6,7,8 当前 tag 是 8 那么此时
5-8 的这些还未应答的消息都会被确认收到消息应答
- false
只会应答 tag=8 的消息 5,6,7 这三个消息依然不会被确认收到消息应答
消息手动应答的代码
-
将手动应答开启
/* 消费者消费消息 * 1.消费哪个队列 * 2.消费成功之后是否要自动应答 true 代表自动应答 false 手动应答 * 3.当一个消息发送过来后的回调接口 * 4.消费者取消消费的回调 */ boolean ack=false; channel.basicConsume(QUEUE_NAME,ack,deliverCallback,cancelCallback); -
消息消费回调时，使用手动应答
/** * 消息发送过来后的回调接口 *1.同一个会话， consumerTag 是固定的 可以做此会话的名字， deliveryTag 每次接收消息+1，可以做此消息处理通道的名字。 *2.消息类 */ DeliverCallback deliverCallback=(consumerTag,delivery)-> String message=new String(delivery.getBody()); System.out.println(message); /** * 参数说明 * 1.消息的标记tag * 2.是否批量应答 */ channel.basicAck(delivery.getEnvelope().getDeliveryTag(),false); ;
消息自动进行重新入队
如果消费者由于某些原因失去连接(其通道已关闭，连接已关闭或 TCP 连接丢失)，导致消息未发送 ACK 确认，RabbitMQ 将了解到消息未完全处理，并将对其重新排队。如果此时其他消费者可以处理，它将很快将其重新分发给另一个消费者。这样，即使某个消费者偶尔死亡，也可以确保不会丢失任何消息。
6.RabbitMQ的持久化,不公平分发及预取值
概念
刚刚我们已经看到了如何处理任务不丢失的情况，但是如何保障当 RabbitMQ 服务停掉以后消
息生产者发送过来的消息不丢失。默认情况下 RabbitMQ 退出或由于某种原因崩溃时，它忽视队列
和消息，除非告知它不要这样做。确保消息不会丢失需要做两件事：我们需要将队列和消息都标
记为持久化。
队列持久化
- 之前我们创建的队列都是非持久化的，rabbitmq 如果重启的化，该队列就会被删除掉，如果
要队列实现持久化 需要在声明队列的时候把 durable(第二个) 参数设置为持久化
- 但是需要注意的就是如果之前声明的队列不是持久化的，需要把原先队列先删除，或者重新
创建一个持久化的队列，不然就会出现错误
-
channel.queueDeclare(RabbitMQChannelUtil.QUEUE_NAME,true,false,false,null);
这个就是持久化队列
消息持久化
- 要想让消息实现持久化需要在消息生产者修改代码，MessageProperties.PERSISTENT_TEXT_PLAIN 添
加这个属性。
队列持久化为false时:
channel.basicPublish("",RabbitMQChannelUtil.QUEUE_NAME,null,message.getBytes());
队列持久化为true时
channel.basicPublish("",RabbitMQChannelUtil.QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN,message.getBytes());
-
将消息标记为持久化并不能完全保证不会丢失消息。尽管它告诉 RabbitMQ 将消息保存到磁盘，但是
这里依然存在当消息刚准备存储在磁盘的时候 但是还没有存储完，消息还在缓存的一个间隔点。此时并没
有真正写入磁盘。持久性保证并不强.更强的持久化后面
发布确认会讲到
不公平分发
在最开始的时候我们学习到 RabbitMQ 分发消息采用的轮训分发，但是在某种场景下这种策略并不是
很好，比方说有两个消费者在处理任务，其中有个消费者 1 处理任务的速度非常快，而另外一个消费者 2
处理速度却很慢，这个时候我们还是采用轮训分发的化就会到这处理速度快的这个消费者很大一部分时间
处于空闲状态，而处理慢的那个消费者一直在干活，这种分配方式在这种情况下其实就不太好，但是
RabbitMQ 并不知道这种情况它依然很公平的进行分发。
为了避免这种情况，我们设置不公平分发：
channel.basicQos(1);
预取值
本身消息的发送就是异步发送的，所以在任何时候，channel 上肯定不止只有一个消息另外来自消费
者的手动确认本质上也是异步的。因此这里就存在一个未确认的消息缓冲区，因此希望开发人员能限制此
缓冲区的大小，以避免缓冲区里面无限制的未确认消息问题。这个时候就可以通过使用 basic.qos 方法设
置“预取计数”值来完成的。该值定义通道上允许的未确认消息的最大数量。一旦数量达到配置的数量，
RabbitMQ 将停止在通道上传递更多消息，除非至少有一个未处理的消息被确认
prefetch就是预取值数
7.发布确认
上文持久化中提到，当消息持久化存入RabbitMQ磁盘时，RabbitMQ突然宕机，则消息未成功存入，会发生消息丢失。所以发布确认即：在消息成功存入磁盘时，返还给生产者一个消息，确认已经存入磁盘
具体介绍
生产者将信道设置成 confirm 模式，一旦信道进入 confirm 模式，所有在该信道上面发布的
消息都将会被指派一个唯一的 ID(从 1 开始)，一旦消息被投递到所有匹配的队列之后，broker
就会发送一个确认给生产者(包含消息的唯一 ID)，这就使得生产者知道消息已经正确到达目的队
列了，如果消息和队列是可持久化的，那么确认消息会在将消息写入磁盘之后发出，broker 回传
给生产者的确认消息中 delivery-tag 域包含了确认消息的序列号，此外 broker 也可以设置
basic.ack 的 multiple 域，表示到这个序列号之前的所有消息都已经得到了处理。
为了保证消息不丢失：
- 开启队列持久化
- 开启消息持久化
- 开启信道的发布确认
开启发布确认的方法
channel.confirmSelect();
发布确认的模式
- 单个确认发布
public static void singleConfirm()
try
Channel channel=RabbitMQChannelUtil.getChannel();
if(channel == null)
System.out.println("信道建立失败");
return;
//开启发布确认
channel.confirmSelect();
long begin=System.currentTimeMillis();
for (int i = 0; i <MESSAGE_COUNT ; i++)
String message=i+"";
channel.basicPublish("",QUEUE_NAME,null,message.getBytes());
//可以加时间参数，当消息发送失败或超过参数时间没成功，则返回false
boolean flag=channel.waitForConfirms();
//如果失败可以重发
if(flag)
System.out.println(message+"发送成功");
else
//这里可以实现重发
System.out.println(message+"发送失败");
long end=System.currentTimeMillis();
System.out.println("发送"+MESSAGE_COUNT+"条消息，耗时"+(end-begin)+"ms");
catch (Exception e)
e.printStackTrace();
发布一个消息之后只有它被确认发布，后续的消息才能继续发布,waitForConfirmsOrDie(long)这个方法只有在消息被确认的时候才返回，如果在指定时间范围内这个消息没有被确认那么它将抛出异常。
缺点：速度太慢
- 批量发布确认模式
public static void batchConfirm()
try
Channel channel=RabbitMQChannelUtil.getChannel();
if(channel == null)
System.out.println("建立连接失败");
return;
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
//当100条消息发布成功时，再确认
int ackMessageCount=100;
//未确认的消息个数
int needAckMessageCount=0;
//开启发布确认
channel.confirmSelect();
long begin=System.currentTimeMillis();
for (int i = 0; i <MESSAGE_COUNT ; i++)
String message=i+"";
channel.basicPublish("",QUEUE_NAME,null,message.getBytes());
needAckMessageCount++;
if(needAckMessageCount == ackMessageCount)
//确认
channel.waitForConfirms();
needAckMessageCount=0;
//判断可能还有消息未发送，再发送依次
if(needAckMessageCount > 0)
channel.waitForConfirms();
long end= System.currentTimeMillis();
System.out.println("发送"+MESSAGE_COUNT+"条消息，耗时"+(end-begin)+"ms");
catch (Exception e)
e.printStackTrace();
缺点:当发生故障导致发布出现问题时，不知道是哪个消息出现问题
- 异步确认发布
原理
RabbitMQ基础总结