Java面试题消息队列

「爱情、让人受尽委屈。」 2024-03-22 11:48 97阅读 0赞

**消息队列**

**1****、为什么使用消息队列？**

消息队列常见的使用场景其实有很多，但是比较核心的有 3 个：解耦、异步、削峰。

**2****、消息队列有什么优点和缺点？**

优点就是在特殊场景下有其对应的好处，解耦、异步、削峰。 缺点有以下几个：

⚫ 系统可用性降低  
系统引入的外部依赖越多，越容易挂掉。本来你就是 A 系统调用 BCD 三个系统的接口就 好了，人 ABCD 四个系统好好的，没啥问题，你偏加个 MQ 进来，万一 MQ 挂了咋整， MQ 一挂，整套系统崩溃的，你不就完了？

⚫ 系统复杂度提高  
硬生生加个 MQ 进来，你怎么保证消息没有重复消费？怎么处理消息丢失的情况？怎么 保证消息传递的顺序性？头大头大，问题一大堆，痛苦不已。

⚫ 一致性问题  
A 系统处理完了直接返回成功了，人都以为你这个请求就成功了；但是问题是，要是

BCD 三个系统那里，BD 两个系统写库成功了，结果 C 系统写库失败了，咋整？你这数据就不一致了。

**3****、****Ka****fk****a****、****Act****i****ve****M****Q****、****R****a****b****b****i****t****MQ****、****R****o****c****k****e****t****MQ****都有什么区别，以及适合哪些场景？**![85255543e98d401d8c14da343cf0a1c8.png][]

**4****、如何保证消息不被重复消费？**

首先，比如 RabbitMQ、RocketMQ、Kafka，都有可能会出现消息重复消费的问题，正常。因为这问题通常不是 MQ 自己保证的，是由我们开发来保证的。

Kafka 实际上有个 offset 的概念，就是每个消息写进去，都有一个 offset，代表消息的序号，然后 consumer 消费了数据之后，每隔一段时间（定时定期），会把自己消费过的消息的 offset 提交一下，表示“我已经消费过了，下次我要是重启啥的，你就让我继续从上次消费到的 offset 来继续消费吧”。

但是凡事总有意外，比如我们之前生产经常遇到的，就是你有时候重启系统，看你怎么重启了，如果碰到点着急的，直接 kill 进程了，再重启。这会导致 consumer 有些消息处理了，但是没来得及提交 offset，尴尬了。重启之后，少数消息会再次消费一次。

**5****、如何保证消息消费的幂等性？**

其实重复消费不可怕，可怕的是你没考虑到重复消费之后，怎么保证幂等性。一条数据重复出现两次，数据库里就只有一条数据，这就保证了系统的幂等性。

幂等性，通俗点说，就一个数据，或者一个请求，给你重复来多次，你得确保对应的数据是不会改变的，不能出错。

其实还是得结合业务来思考：

⚫ 比如你拿个数据要写库，你先根据主键查一下，如果这数据都有了，你就别插入了，

update 一下好吧。

⚫ 比如你是写 Redis，那没问题了，反正每次都是 set，天然幂等性。

⚫ 比如你不是上面两个场景，那做的稍微复杂一点，你需要让生产者发送每条数据的时候，

里面加一个全局唯一的 id，类似订单 id 之类的东西，然后你这里消费到了之后，先根据

这个 id 去比如 Redis 里查一下，之前消费过吗？如果没有消费过，你就处理，然后这个

id 写 Redis。如果消费过了，那你就别处理了，保证别重复处理相同的消息即可。

⚫ 比如基于数据库的唯一键来保证重复数据不会重复插入多条。因为有唯一键约束了，重复

数据插入只会报错，不会导致数据库中出现脏数据。

**6****、如何保证消息的可靠性传输？（如何处理消息丢失的****问****题）**

**Ra****b****b****i****tMQ**  
生产者将数据发送到 RabbitMQ 的时候，可能数据就在半路给搞丢了，因为网络问题啥

的，都有可能。

此时可以选择用 RabbitMQ 提供的事务功能，就是生产者发送数据之前开启 RabbitMQ 事务channel.txSelect，然后发送消息，如果消息没有成功被 RabbitMQ 接收到，那么生产者会收到异常报错，此时就可以回滚事务channel.txRollback，然后重试发送消息；如果收到了消息，那么可以提交事务channel.txCommit。

**Ra****b****b****i****tM****Q****弄丢了****数****据**

就是 RabbitMQ 自己弄丢了数据，这个你必须开启 RabbitMQ 的持久化，就是消息写入之后会持久化到磁盘，哪怕是 RabbitMQ 自己挂了，恢复之后会自动读取之前存储的数据，一般数据不会丢。除非极其罕见的是，RabbitMQ 还没持久化，自己就挂了，可能导致少量数据丢失，但是这个概率较小。

**消费端弄丢了数据**  
RabbitMQ 如果丢失了数据，主要是因为你消费的时候，刚消费到，还没处理，结果进程

挂了，比如重启了，那么就尴尬了，RabbitMQ 认为你都消费了，这数据就丢了。

这个时候得用 RabbitMQ 提供的 ack 机制，简单来说，就是你必须关闭 RabbitMQ 的自动 ack，可以通过一个 api 来调用就行，然后每次你自己代码里确保处理完的时候，再在程序里 ack 一把。这样的话，如果你还没处理完，不就没有 ack 了？那 RabbitMQ 就认为你还没处理完，这个时候 RabbitMQ 会把这个消费分配给别的 consumer 去处理，消息是不会丢的。

Kafka

**消费端弄丢了数据**  
唯一可能导致消费者弄丢数据的情况，就是说，你消费到了这个消息，然后消费者那边自动提交了 offset，让 Kafka 以为你已经消费好了这个消息，但其实你才刚准备处理这个消息，你还没处理，你自己就挂了，此时这条消息就丢咯。

**K****afka****弄丢了数****据**  
Kafka 的 leader 机器宕机了，将 follower 切换为 leader 之后，就会发现说这个数据就丢了。

所以此时一般是要求起码设置如下 4 个参数：

⚫ 给 topic 设置 replication.factor 参数：这个值必须大于 1，要求每个 partition 必须有至

少 2 个副本。

⚫ 在 Kafka 服务端设置 min.insync.replicas 参数：这个值必须大于 1，这个是要求一个

leader 至少感知到有至少一个 follower 还跟自己保持联系，没掉队，这样才能确保

leader 挂了还有一个 follower 吧。

⚫ 在 producer 端设置 acks=all：这个是要求每条数据，必须是写入所有 replica 之后，才

能认为是写成功了。

<table style="margin-left:5.4pt;"> 
 <tbody> 
 <tr> 
 <td style="vertical-align:top;width:17.5pt;"> ⚫&nbsp; </td> 
 <td style="vertical-align:top;width:478.5pt;"> 在&nbsp;producer 端设置&nbsp;retries=MAX（很大很大很大的一个值，无限次重试的意思）：这个 </td> 
 </tr> 
 </tbody> 
</table>

是要求一旦写入失败，就无限重试，卡在这里了。

**7****、如果让你写一个消息队列，该如何进行架构设计啊？说一下你的思路****。**

⚫ 首先这个 mq 得支持可伸缩性吧，就是需要的时候快速扩容，就可以增加吞吐量和容量， 那怎么搞？设计个分布式的系统呗，参照一下 kafka 的设计理念，broker -> topic \->  partition，每个 partition 放一个机器，就存一部分数据。如果现在资源不够了，简单 啊，给 topic 增加 partition，然后做数据迁移，增加机器，不就可以存放更多数据，提 供更高的吞吐量了？

⚫ 其次你得考虑一下这个 mq 的数据要不要落地磁盘吧？那肯定要了，落磁盘才能保证别进 程挂了数据就丢了。那落磁盘的时候怎么落啊？顺序写，这样就没有磁盘随机读写的寻址 开销，磁盘顺序读写的性能是很高的，这就是 kafka 的思路。

⚫ 其次你考虑一下你的 mq 的可用性啊？这个事儿，具体参考之前可用性那个环节讲解的 kafka 的高可用保障机制。多副本 \-> leader & follower \-> broker 挂了重新选举 leader 即可对外服务。

**8****、如何解决消息队列的延时以及过期失效问题****？**

**大量消息****在****m****q****里积****压****了几个小时了还没解决**  
一个消费者一秒是1000条，一秒3个消费者是3000条，一分钟是18万条，1000多万条，所以如果你积压了几百万到上千万的数据，即使消费者恢复了，也需要大概1小时的时间才能恢复过来。

一般这个时候，只能操作临时紧急扩容了，具体操作步骤和思路如下：  
⚫ 先修复consumer的问题，确保其恢复消费速度，然后将现有cnosumer都停掉。 ⚫ 新建一个topic，partition是原来的10倍，临时建立好原先10倍或者20倍的queue 数量。

⚫ 然后写一个临时的分发数据的consumer程序，这个程序部署上去消费积压的数据，消费 之后不做耗时的处理，直接均匀轮询写入临时建立好的10倍数量的queue。

<table style="margin-left:5.4pt;"> 
 <tbody> 
 <tr> 
 <td style="vertical-align:top;width:17.5pt;"> ⚫&nbsp; </td> 
 <td style="vertical-align:top;width:478.5pt;"> 接着临时征用10倍的机器来部署consumer，每一批consumer消费一个临时queue </td> 
 </tr> 
 </tbody> 
</table>

的数据。

这种做法相当于是临时将queue资源和consumer资源扩大10倍，以正常的10倍速度来消费数据。等快速消费完积压数据之后，得恢复原先部署架构，重新用原先的consumer机器来消费消息。

**消息队列过期失效问题**  
假设你用的是rabbitmq，rabbitmq是可以设置过期时间的，就是TTL，如果消息在queue中积压超过一定的时间就会被rabbitmq给清理掉，这个数据就没了。那这就是第二个坑了。这就不是说数据会大量积压在mq里，而是大量的数据会直接搞丢。

这个情况下，就不是说要增加consumer消费积压的消息，因为实际上没啥积压，而是丢了大量的消息。我们可以采取一个方案，就是批量重导，  
这个时候我们就开始写程序，将丢失的那批数据，写个临时程序，一点一点的查出来，然后重新灌入mq里面去，把白天丢的数据给他补回来。也只能是这样了。

**9****、消息队列满了以后该怎么处理？**

如果走的方式是消息积压在mq里，那么如果你很长时间都没处理掉，此时导致mq都快写满了，咋办？这个还有别的办法吗？没有，谁让你第一个方案执行的太慢了，你临时写程序，接入数据来消费，消费一个丢弃一个，都不要了，快速消费掉所有的消息。然后走第二个方案，到了晚上再补数据吧。

**1****0****、消息如何分发****？**

若该队列至少有一个消费者订阅，消息将以循环（round-robin）的方式发送给消费者。每条消息只会分发给一个订阅的消费者（前提是消费者能够正常处理消息并进行确认）。通过路由可实现多消费的功能。

**1****1****、消息怎么路由****？**

消息提供方\->路由\->一至多个队列消息发布到交换器时，消息将拥有一个路由键（routing key），在消息创建时设定。通过队列路由键，可以把队列绑定到交换器上。消息到达交换器后，RabbitMQ 会将消息的路由键与队列的路由键进行匹配（针对不同的交换器有不同的路由规则）；

常用的交换器主要分为一下三种：

⚫ fanout：如果交换器收到消息，将会广播到所有绑定的队列上

⚫ direct：如果路由键完全匹配，消息就被投递到相应的队列

用通配符

**1****2****、消息基于什么传****输****？**

由于 TCP 连接的创建和销毁开销较大，且并发数受系统资源限制，会造成性能瓶颈。RabbitMQ 使用信道的方式来传输数据。信道是建立在真实的 TCP 连接内的虚拟连接，且每条 TCP 连接上的信道数量没有限制。

**1****3****、为什么不应该对****所****有****的****mes****s****a****g****e** **都使****用****持久化机制？**

首先，必然导致性能的下降，因为写磁盘比写 RAM 慢的多，message 的吞吐量可能有 10 倍的差距。

其次，message 的持久化机制用在 RabbitMQ 的内置 cluster 方案时会出现“坑爹”问题。矛盾点在于，若 message 设置了 persistent 属性，但 queue 未设置 durable 属性，那么当该 queue 的 owner node 出现异常后，在未重建该 queue 前，发往该 queue 的 message 将被 blackholed ；若 message 设置了 persistent 属性，同时 queue 也设置了 durable 属性，那么当 queue 的 owner node 异常且无法重启的情况下，则该 queue 无法在其他 node 上重建，只能等待其 owner node 重启后，才能恢复该 queue 的使用，而在这段时间内发送给该 queue 的 message 将被 blackholed 。

所以，是否要对 message 进行持久化，需要综合考虑性能需要，以及可能遇到的问题。若想达到 100,000 条/秒以上的消息吞吐量（单 RabbitMQ 服务器），则要么使用其他的方式来确保 message 的可靠 delivery ，要么使用非常快速的存储系统以支持全持久化（例如使用 SSD）。另外一种处理原则是：仅对关键消息作持久化处理（根据业务重要程度），且应该保证关键消息的量不会导致性能瓶颈。

**1****4****、如何保证高可用****的****？****Rabb****i****tM****Q** **的集****群**

RabbitMQ 是比较有代表性的，因为是基于主从（非分布式）做高可用性的，我们就以 RabbitMQ 为例子讲解第一种 MQ 的高可用性怎么实现。RabbitMQ 有三种模式：单机模式、普通集群模式、镜像集群模式。

单机模式，就是 Demo 级别的，一般就是你本地启动了玩玩儿的?，没人生产用单机模式 普通集群模式，意思就是在多台机器上启动多个 RabbitMQ 实例，每个机器启动一个。你创建的 queue，只会放在一个 RabbitMQ 实例上，但是每个实例都同步 queue 的元数据（元数据可以认为是 queue 的一些配置信息，通过元数据，可以找到 queue 所在实例）。你消费的时候，实际上如果连接到了另外一个实例，那么那个实例会从 queue 所在实例上拉取数据过来。这方案主要是提高吞吐量的，就是说让集群中多个节点来服务某个 queue 的读写操作。 镜像集群模式：这种模式，才是所谓的 RabbitMQ 的高可用模式。跟普通集群模式不一样的是，在镜像集群模式下，你创建的 queue，无论元数据还是 queue 里的消息都会存在于多个实例上，就是说，每个 RabbitMQ 节点都有这个 queue 的一个完整镜像，包含 queue 的全部数据的意思。然后每次你写消息到 queue 的时候，都会自动把消息同步到多个实例的 queue 上。RabbitMQ 有很好的管理控制台，就是在后台新增一个策略，这个策略是镜像集群模式的策略，指定的时候是可以要求数据同步到所有节点的，也可以要求同步到指定数量的节点，再次创建 queue 的时候，应用这个策略，就会自动将数据同步到其他的节点上去了。

这样的话，好处在于，你任何一个机器宕机了，没事儿，其它机器（节点）还包含了这个 queue 的完整数据，别的 consumer 都可以到其它节点上去消费数据。坏处在于，第一，这个性能开销也太大了吧，消息需要同步到所有机器上，导致网络带宽压力和消耗很重！RabbitMQ 一个 queue 的数据都是放在一个节点里的，镜像集群下，也是每个节点都放这个 queue 的完整数据。

**1****5****、****Ra****b****b****itM****Q** **的****工****作模式**

⚫ simple模式（即最简单的收发模式）

⚫ work工作模式(资源的竞争)

⚫ publish/subscribe发布订阅(共享资源)

⚫ routing路由模式

⚫ topic主题模式

**1****6****、为什么需要消息****系****统****，****MySQ****L****不能满足需求吗？**

⚫ 解耦：  
允许你独立的扩展或修改两边的处理过程，只要确保它们遵守同样的接口约束。

⚫ 冗余：  
消息队列把数据进行持久化直到它们已经被完全处理，通过这一方式规避了数据丢失风险。许多消息队列所采用的”插入\-获取\-删除”范式中，在把一个消息从队列中删除之前，需要你的处理系统明确的指出该消息已经被处理完毕，从而确保你的数据被安全的保存直到你使用完毕。

⚫ 扩展性：  
因为消息队列解耦了你的处理过程，所以增大消息入队和处理的频率是很容易的，只要另外增加处理过程即可。

⚫ 灵活性 & 峰值处理能力：

在访问量剧增的情况下，应用仍然需要继续发挥作用，但是这样的突发流量并不常见。如果为以能处理这类峰值访问为标准来投入资源随时待命无疑是巨大的浪费。使用消息队列能够使关键组件顶住突发的访问压力，而不会因为突发的超负荷的请求而完全崩溃。