-
消息队列的作用与使用场景
异步：批量数据异步处理（批量上传文件）
削峰：高负载任务负载均衡（电商秒杀抢购）
解耦：串行任务并行化（退货流程解耦）
广播：基于Pub/Sub实现一对多通信 -
多个消费者监听一个队列时，消息如何分发
轮询
默认的策略，消费者轮流，平均地接收消息
公平分发
根据消费者的能力来分发消息，给空闲的消费者发送更多消息
//当消费者有x条消息没有响应ACK时，不再给这个消费者发送消息
channel.basicQos(int x)
1
2
- 无法被路由的消息去了哪里
无设置的情况下，无法路由（Routing key错误）的消息会被直接丢弃
解决方案：
将mandatory设置为true，并配合ReturnListener，实现消息的回发
声明交换机时，指定备份的交换机
Map<String,Object> arguments = new HashMap<String,Object>();
arguments.put("alternate-exchange","备份交换机名");
1
2
-
消息在什么时候会变成死信
消息拒绝并且没有设置重新入队
消息过期
消息堆积，并且队列达到最大长度，先入队的消息会变成DL -
RabbitMQ如何实现延时队列
利用TTL（队列的消息存活时间或者消息存活时间），加上死信交换机
// 设置属性，消息10秒钟过期
AMQP.BasicProperties properties = new AMQP.BasicProperties.Builder()
.expiration("10000") // TTL
// 指定队列的死信交换机
Map<String,Object> arguments = new HashMap<String,Object>();
arguments.put("x-dead-letter-exchange","DLX_EXCHANGE");
1
2
3
4
5
6
7
- 如何保证消息的可靠性投递
发送方确认模式：
将信道设置成confirm模式（发送方确认模式），则所有在信道上发布的消息都会被指派一个唯一的ID。
一旦消息被投递到目的队列后，或者消息被写入磁盘后（可持久化的消息），信道会发送一个确认给生产者（包含消息唯一ID）。
如果RabbitMQ发生内部错误从而导致消息丢失，会发送一条nack（not acknowledged，未确认）消息。
发送方确认模式是异步的，生产者应用程序在等待确认的同时，可以继续发送消息。当确认消息到达生产者应用程序，生产者应用程序的回调方法就会被触发来处理确认消息。
接收方确认机制
接收方消息确认机制：消费者接收每一条消息后都必须进行确认（消息接收和消息确认是两个不同操作）。只有消费者确认了消息，RabbitMQ才能安全地把消息从队列中删除。
这里并没有用到超时机制，RabbitMQ仅通过Consumer的连接中断来确认是否需要重新发送消息。也就是说，只要连接不中断，RabbitMQ给了Consumer足够长的时间来处理消息。保证数据的最终一致性；
下面罗列几种特殊情况：
如果消费者接收到消息，在确认之前断开了连接或取消订阅，RabbitMQ会认为消息没有被分发，然后重新分发给下一个订阅的消费者。（可能存在消息重复消费的隐患，需要去重）
如果消费者接收到消息却没有确认消息，连接也未断开，则RabbitMQ认为该消费者繁忙，将不会给该消费者分发更多的消息。
- 消息幂等性
生产者方面：可以对每条消息生成一个msgID，以控制消息重复投递
AMQP.BasicProperties properties = new AMQP.BasicProperties.Builder()
porperties.messageId(String.valueOF(UUID.randomUUID()))
1
2
消费者方面：消息体中必须携带一个业务ID，如银行流水号，消费者可以根据业务ID去重，避免重复消费
- 消息如何被优先消费
//生产者
Map<String, Object> argss = new HashMap<String, Object>();
argss.put("x-max-priority",10);
//消费者
AMQP.BasicProperties properties = new AMQP.BasicProperties.Builder()
.priority(5) // 优先级，默认为5，配合队列的 x-max-priority 属性使用
1
2
3
4
5
6
7
-
如何保证消息的顺序性
一个队列只有一个消费者的情况下才能保证顺序，否则只能通过全局ID实现（每条消息都一个msgId，关联的消息拥有一个parentMsgId。可以在消费端实现前一条消息未消费，不处理下一条消息；也可以在生产端实现前一条消息未处理完毕，不发布下一条消息） -
RabbitMQ的集群模式和集群节点类型
普通模式：默认模式，以两个节点（rabbit01，rabbit02）为例来进行说明，对于Queue来说，消息实体只存在于其中一个节点rabbit01（或者rabbit02），rabbit01和rabbit02两个节点仅有相同的元数据，即队列结构。当消息进入rabbit01节点的Queue后，consumer从rabbit02节点消费时，RabbitMQ会临时在rabbit01，rabbit02间进行消息传输，把A中的消息实体取出并经过B发送给consumer，所以consumer应尽量连接每一个节点，从中取消息。即对于同一个逻辑队列，要在多个节点建立物理Queue。否则无论consumer连rabbit01或rabbit02，出口总在rabbit01，会产生瓶颈。当rabbit01节点故障后，rabbit02节点无法取到rabbit01节点中还未消费的消息实体。如果做了消息持久化，那么等到rabbit01节点恢复，然后才可被消费。如果没有消息持久化，就会产生消息丢失的现象。
镜像模式：把需要的队列做成镜像队列，存在与多个节点属于RabibitMQ的HA方案，该模式解决了普通模式中的问题，其实质和普通模式不同之处在于，消息体会主动在镜像节点间同步，而不是在客户端取数据时临时拉取，该模式带来的副作用也很明显，除了降低系统性能外，如果镜像队列数量过多，加之大量的消息进入，集群内部的网络带宽将会被这种同步通讯大大消耗掉，所以在对可靠性要求比较高的场合中适用
节点分为内存节点（保存状态到内存，但持久化的队列和消息还是会保存到磁盘），磁盘节点（保存状态到内存和磁盘），一个集群中至少需要一个磁盘节点
11.如何自动删除长时间没有消费的消息
// 通过队列属性设置消息过期时间
Map<String, Object> argss = new HashMap<String, Object>();
argss.put("x-message-ttl",6000);
// 对每条消息设置过期时间
AMQP.BasicProperties properties = new AMQP.BasicProperties.Builder()
.expiration("10000") // TTL
1
2
3
4
5
6
7
12.消息基于什么传输
RabbitMQ使用信道的方式来传输数据。信道是建立在真实的TCP连接内的虚拟连接，且每条TCP连接上的信道数量没有限制
13.如何确保消息不丢失
消息持久化，当然前提是队列必须持久化
RabbitMQ确保持久性消息能从服务器重启中恢复的方式是，将它们写入磁盘上的一个持久化日志文件，当发布一条持久性消息到持久交换器上时，Rabbit会在消息提交到日志文件后才发送响应。
一旦消费者从持久队列中消费了一条持久化消息，RabbitMQ会在持久化日志中把这条消息标记为等待垃圾收集。如果持久化消息在被消费之前RabbitMQ重启，那么Rabbit会自动重建交换器和队列（以及绑定），并重新发布持久化日志文件中的消息到合适的队列。
————————————————
版权声明：本文为CSDN博主「JAVA攻城狮_Nemo」的原创文章，遵循 CC 4.0 BY-SA 版权协议，转载请附上原文出处链接及本声明。
原文链接：https://blog.csdn.net/huijiangu4621/article/details/91347319
网友评论