为什么要使用MQ

当面试官问出 “为什么要使用MQ” 时，其实也就是在问 MQ 有哪些作用？

1. 解耦

一个项目随着业务迭代会不断增加下游调用方，按照传统方式，那么每增加一个下游，就需要修改一次当前项目的代码，增加新的下游交互逻辑，工作量不可谓不大。

而如果在这里考虑接入 MQ 作为中间代理，那么只需要将数据写入到 MQ，而不用考虑复杂的交互通信。并且，下游增加时，基本也不用修改上游代码，只需要让下游监听对应的通道即可。甚至，配合上动态配置，可以实现上游零修改地增加下游服务。

2. 异步

如果需要提高高延时接口的响应速度。那么，就可以考虑使用 MQ 来实现异步执行，只需要将请求信息写入 MQ 就可以立即返回，实现高响应速度，至于剩余的工作交给 MQ 即可。

3. 流量削峰

如果业务中有高流量的场景，为了避免高流量一下子打到数据库层，造成数据库瘫痪，可以考虑将请求写入 MQ，即 MQ 作为一个缓冲区，然后在系统支持的消费速度下，通过下游服务不断地消费 MQ 中累积的消息。当高峰期过后，MQ 中的积压消息将慢慢被消费掉。

MQ优点和缺点

优点

如上所述：

1. 解耦
2. 异步
3. 流量削峰

缺点

1. 降低了系统的可用性。原先只有上下游，如今中间多了个 MQ。一旦 MQ 故障，整个系统将不可用。
2. 提高了系统的复杂性。多了 MQ 之后，开发过程中可能需要考虑 MQ 重复消费，消息丢失，如果保证消息顺序的问题。
3. 一致性问题。MQ 作为中间代理，多个下游执行过程中，某个或多个执行失败，那么势必造成数据的不一致。

不同的MQ区别

总结而言，如果是技术挑战不高，推荐使用 RabbitMQ（Erlang语言阻碍Java工程师深入研究掌握它）；如果研发能力强的，可以考虑 RocketMQ；如果是用于大数据领域，那么推荐 Kafka。

不同MQ特点

RabbitMQ

RocketMQ

参考 消息队列那些事

持久化方式

本地存储，commitLog + consumerQueue + indexFile
支持的刷盘方式：同步刷盘（吞吐量低，一致性高）、异步刷盘（吞吐量高、一致性低）

生产环境不建议自动创建主题

自动创建主题可能不会起到负载均衡的作用。

支持自动创建主题的 broker 启动时会注册默认主题到 nameServer，生产者发送不存在主题的消息时，由于从 nameServer 拉取不到 broker 信息，所以会往默认主题的 broker 发。默认主题的 broker 发现主题不存在就会创建这个主题，然后 30s 后同步到 nameServer，但生产者 30s 内不再用不存在主题发送消息，那么 30s 后 nameServer 同步主题信息后，上面可能只存在一个 broker。即自动创建主题可能不会起到负载均衡的作用。

推拉模式

推拉模式指的是 consumer 和 broker 交互的过程。

推模式，即 broker 一收到新的消息就推送到 consumer，消息实时性好，但如果推送速率大于消费速率，那么容易造成 consumer 高负载。为了解决速率不平衡问题，那么 consumer 消费不过来需要告诉 broker，broker 就需要维护每个 consumer 的消费状态，这就增加了 broker 的复杂度。
所以，总结而言，推模式适用于 consumer 消费能力强，要求消息实时性的场景。

拉模式，即 consumer 主动向 broker 拉取消息。这种情况下，consumer 可以根据自身情况来拉取消息，但循环拉取消息必定有时间间隔，比如2s，那么拉取消息必然出现消息延迟，并且如果一段时间内都没有消息，那么拉取一直在做无用功。
所以，总结而言，拉模式适用于 consumer 消费能力需要自适应，不要求消息实时性的场景。

RocketMQ 和 Kafka 默认采用拉模式。为了避免拉模式的缺陷，采用“长轮询”的机制。一句话说就是消费者和 Broker 相互配合，拉取消息请求不满足条件的时候 hold 住，避免了多次频繁的拉取动作，当消息一到就提醒返回。

Kafka

如何保证消息不重复消费（幂等）

1. 如果将数据写入到 Mysql，需要根据表主键判断行是否存在进行新增或更新即可；
2. 如果将数据写入到 redis，直接写入即可。不过，需要一个业务唯一ID作为 redis 的 key；
3. 如果没有使用到 Mysql 或 redis，那么还是需要使用其中一个来存储已经消费过的消息，处理方法同上。

如何保证消息不丢失（以RabbitMQ为例）

1. 保证生产者不丢数据

一是可以考虑开启 RabbitMQ 提供的事务操作。当发生消息丢失异常时，可以回滚事务，然后重新发送。这种方式下会极大地降低 MQ 吞吐量；

二是开启 channel 的 confirm 模式。生产者发送消息给 MQ 后，MQ 会回传一个 ack。如果生产者没有收到，就可以重试。该监听方法是异步的，不会阻塞，可以直接进行下一条的发送。

2. 保证 MQ 不丢数据

这里需要开启 MQ 的持久化。RabbitMQ 开启持久化分两步，一是开启 queue 的持久化，这样 MQ 会持久化 queue 的元数据；二是设置发送消息时的消息选项，是否持久化。

当 MQ 自己宕机，重启之后，也可以从磁盘中恢复 queue 和 queue 中的消息。

3. 保证消费者不丢数据

消费者默认有 AutoAck 的设置，即消费者会自动通知 RabbitMQ，这条消息已经消费完成。如果这条消息正在消费过程中，且消费者已经通知 MQ 消息消费成功，而此时消费者又发生故障，那么这条消息将丢失。

解决办法也很明显，就是关闭 AutoAck，消息消费完了才返回 ack。

如何保证消息有序消费

在某些场景下，需要保证 MQ 中的消息有序地消费。比如同一行记录的多个操作写入到 MQ（binlog 同步到 MQ），如果消费者不是有序消费的话，可能就会造成数据的不一致性。

解决办法也简单，RabbitMQ 就是一个 queue 对应一个 consumer。由于写入到一个 queue 的消息是有序的，一个 consumer 消费起来自然也是有序的。Kafka 则是一个 partition 对应一个 consumer。

如果一个 queue 对应一个消费者的消费速度达不到业务需求，可以根据业务 key，将消息通过 hash 算法，映射到多个内存队列，然后开启多个线程进行消费。对于同一个业务 key 的消息，它们的消费是顺序的。

延时消息

开源RocketMQ

rocketMQ 仅支持若干个级别的延时消息，通过在启动时，创建多个主题 schdule_topic_xx 对应不同延时级别。

延时消费流程：

1. 首先将消息存到commitLog（MQ的本地文件系统）。当判断出为延时消息后，将消息投递到 schdule_topic_xx 主题队列
2. delay service 读取 schdule_topic_xx 队列偏移量，然后根据主题队列的延时级别，不断创建延时迟任务（通过 timer 实现）
3. 当延时任务到期后，读取消息投递到目标队列，供消费者消费

RocketMQ本地存储指北

1. 文件类型
   （1）commitLog：存储RocketMQ的所有Topic的消息，称之为混合存储。默认每个commitLog文件为1G，文件以起始偏移量命名，所以通过消息中的offset就可以定位消息存储在哪个commitLog文件。
   （2）consumerQueue：消息消费队列，可以认为是commitLog中消息的索引。consumerQueue中的每条记录是固定大小的，分三部分：commitLog的物理偏移地址，消息长度，tag的哈希值。
   （3）IndexFile：索引文件，是提供的额外查询消息的手段，不影响主流程。通过keys或事件区间来查询消息，IndexFile文件名以创建时间戳命名，同样存储commitLog对应的索引。
2. 消息写入流程
   先将消息写入commitLog，然后异步写入到consumerQueue和IndexFile中。
3. 消息查询流程
   根据Topic、queueId和offsetTable.offset中记录的当前队列已经消费到的行数，可以从consumerQueue中得到消息在commitLog中的偏移量，然后再从commitLog中读出消息。参考消息查询流程

场景题

MQ 积压大量消息

1、为什么出现消息堆积，可能造成的原因有

• 生产者发送速率远大于消费者消费速率（高并发场景，比如促销，秒杀）
• 消费者出现问题，导致消费速率异常或无法消费

2、处理办法
（1）先处理 consumer，保证其能正常消费
（2）再建立原先 10 倍或 20 倍的队列，保证数据分发速率
（3）再启动 10 倍或 20 倍的机器来部署 consumer，将堆积的大量消息消费掉

大量消息积压，并设置了过期时间

这时的问题主要是消息过期导致的丢失。这时考虑对过期的消息进行批量重导。写程序将过期丢失的数据查询出来，再重新添加回 MQ，即补回丢失的数据