第六篇：消息中间件篇

“求其上，得其中；求其中，得其下，求其下，必败”

新版Java面试专题视频教程，java八股文面试全套真题+深度详解（含大厂高频面试真题）_哔哩哔哩_bilibili

开篇

RabbitMQ

如何保证消息不丢失

• 消息的正常流程，消息什么时候可能丢失

  

  由消息的发送者发送消息到交换机，由交换机路由到队列，由消费者消费消息。

  上述过程中所有环节都有可能丢失消息。

  

• 生产者确认机制

  RabbitMQ 提供了生产者确认机制，以确保在消息发送到消息队列过程中不会丢失。一旦消息被成功发送到消息队列，RabbitMQ 会向发送者返回一个结果，指示消息是否成功处理（若成功发送到队列，则返回 ack publish-confirm，其中 ack 表示确认acknowledge）。

  在不同阶段出现不同错误时，RabbitMQ 会向生产者发送不同的信息。

  • 如果消息未能到达交换机，则会返回 nack publish-confirm。

  • 如果消息未能成功路由到队列，则会返回 ack publish-return。

  

  消息失败之后如何处理呢？

  1. 回调方法即时重发

  2. 记录日志

  3. 保存到数据库然后定时重发，成功发送后即刻删除表中的数据

• 消息持久化

  如果消息已经到达MQ中，但是MQ宕机了，怎么办？

  MQ默认是内存存储消息，开启持久化功能可以确保缓存在MQ中的消息不丢失。

  1. 交换机持久化

     1 2 3 4 5

     @Bean public DirectExchange simpleExchange(){ //三个参数：交换机名称、是否持久化、当没有queue与其绑定时是否自动删除 return new DirectExchange("simple.direct", true, false); }

  2. 队列持久化

     1 2 3 4 5

     @Bean public Queue simpleQueue(){ //使用QueueBuilder构建队列，durable就是持久化的 return QueueBuilder.durable("simple.queue").build(); }

  3. 消息持久化。SpringAMQP（AMQP，高级消息队列协议标准）中的消息默认是持久的，可以通过MessageProperties中的DeliveryMode来指定

     1 2 3 4

     Message msg = MessageBuilder .withBody(message.getBytes(standardCharsets.UTF_8)) // 消息体 .setDeliveryMode(MessageDeliveryMode.PERSISTENT) // 特久化 .build();

• 消费者确认

  RabbitMQ支持消费者确认机制，即：消费者处理消息后可以向MQ发送ack回执，MQ收到ack回执后才会删除该消息。而SpringAMQP则充许配置三种确认模式：

  • 手动ACKmanual：需要在业务代码执行完毕后手动调用API发送ACK确认消息已被处理。

  • 自动ACKauto：由Spring监测Listener代码是否出现异常。如果没有异常，则自动返回ACK确认消息已被处理；若出现异常，则返回NACK。

  • 关闭ACKnone：选择关闭ACK时，消息队列假定消费者在接收消息后会成功处理。因此，消息一旦被投递，即被立即删除。

  如无特殊情况 ，我们一般选择第二种方式，即自动ACK。如果消费者接收消息失败了，将会返回NACK。而此时就不做别的了吗？我们可以利用Spring的retry机制，在消费者出现异常时利用本地重试，设置重试次数，当次数达到了以后，如果消息依然失败，将消息投递到异常交换机，交由人工处理。

  

RabbitMQ消息的重复消费问题

我们之前提到，消费者会有一个消息确认的机制来通知MQ消息已被消费，但如果消费者还没来得及给MQ发送确认，网络发生抖动或消费者宕机了，在网络恢复后，因为我们设置了重试机制，那么消费者就会重新消费这个消息，这里不就是重复消费了吗。这个该如何解决呢？

解决方案：

• 每条消息设置一个唯一的标识id

  生产者发送消息的时候，有一个业务的标识id，可以是支付id或者订单id等。当消费者接收到消息的时候，去校验这个id是否存在，比如根据订单去表中查询。如果不存在，就正常接收消息就行了；如果存在，就证明此消息已经消费过了，那就不需要再消费了。

• 幂等方案：【分布式锁、数据库锁（悲观、乐观）】

RabbitMQ中的死信交换机（RabbitMQ延迟队列）

• 延迟队列

  延迟队列是指消息在进入队列后不会立即被消费，而是会在一定延迟时间后才会被消费。（延迟队列 = 死信交换机+TTL）

  场景：超时订单、限时优惠、定时发布

  

• 死信交换机

  当一个队列中的消息满足下列情况之一的，可以成为死信（dead letter）：

  • 消费者可通过 basic.reject（拒绝单条消息）或 basic.nack（拒绝多条消息） 声明消费失败，并将消息的 requeue 参数设为 false。

  • 消息因为超时过期而无人消费。

  • 当要投递的队列消息堆积达到上限时，最早进入队列的消息可能会成为死信。

  如果该队列配置了dead-letter-exchange属性，指定了一个交换机之后，那么队列中的死信就会投递到这个交换机中，而这个交换机称为死信交换机（DeadLetterExchange，简称DLX）。

  

  我们可以如下配置：

  

  TTL（Time-To-Live，存活时间）是消息在队列中允许存活的时间。当消息的 TTL 结束后仍未被消费，它就会变成死信。TTL 超时通常有两种情况：

  • 队列本身设置了存活时间。

  • 消息本身设置了存活时间。

  在消息发送时，我们可以指定一个 TTL。当消息的 TTL 超时（上述两者只要有一个超时，消息就超时）后，它将被存入死信交换机中。而死信交换机中的消息将会由指定的消费者进行处理。

• 延迟队列插件

  对于延迟队列，RabbitMQ还有另一种实现方式，就是延迟队列插件 DelayExchange，它需要安装在RabbitMQ中。

  

  RabbitMQ有一个官方的插件社区，地址为：Community Plugins | RabbitMQ

  DelayExchange本质还是官方的三种交换机，只是添加了延迟功能。因此使用时只需要声明一个交换机，交换机的类型可以是任意类型，然后设置delayed为true即可。如下：

  

  然后，在生产者发送消息的时候，需要加上x-delay这个head，值为超时时间。

  

消息堆积如何解决

• 消息堆积产生的原因

  当生产者发送消息的速度超过了消费者处理消息的速度，就会导致队列中的消息堆积，直到队列存储消息达到上限。之后发送的消息就会成为死信，可能会被去弃，这就是消息堆积问题。

  解决思路有三种：

  • 增加更多消费者，提高消费速度

  • 在消费者内开启线程池加快消息处理速度

  • 使用惰性队列。扩大队列容积，提高堆积上限

  其中第三种解决方法需要使用到RabbitMQ中的一种惰性队列来解决。

• 惰性队列

  情性队列的特征如下：

  • 接收到消息后直接存入磁盘而非内存

  • 消费者要消费消息时才会从磁盘中读取并加载到内存

  • 支持数百万条的消息存储

  如何开启惰性队列？如果是配置的方式，在声明队列的时候，添加.lazy()属性即可。

  

  如果是注解的方式，我们需要做如下配置：

  1 2 3 4 5

  @RabbitListener(queuesToDeclare = @Queue( name = "lazy.queue", durable = "true", arguments = @Argument(name = "x-queue-mode", value = "lazy") ))

  

RabbitMQ的高可用机制

• 普通集群

  或者叫标准集群（classic cluster），具备下列特征：

  • 会在集群的各个节点间共享部分数据，包括：交换机、队列元信息。不包含队列中的消息

    如果有一个交换机 test.exchange，其信息将在各个节点上都有。而每个节点上的队列可能不同，但如果一个节点A上有队列 test.queue1，则节点B也可以有对 test.queue1 的引用信息，这些引用信息也称为元信息。但这些信息不包含队列中的实际消息内容。

  • 当访问集群某节点时，如果队列不在该节点，会从数据所在节点传递到当前节点并返回

    假设有一个消费者要消费 q1 队列中的消息，但他查找的是集群中的第三个节点。即使队列 q1 不在第三个节点上，但因为第三个节点中有对 q1 的引用信息，所以请求会传递到包含 q1 队列的节点上，以便消费消息。无论访问的是哪个节点，都会经过这样的传递过程，确保去到包含队列的节点上消费消息。

  • 队列所在节点宕机，队列中的消息就会消失。

  

• 镜像集群

  本质上是主从模式，具备以下特征：

  • 交换机、队列以及队列中的消息在各个MQ的镜像节点之间同步备份。

  • 创建队列的节点被称为该队列的主节点，备份到的其他节点称为该队列的镜像节点。

  • 一个队列的主节点也可能是另一个队列的镜像节点。

  • 所有操作都由主节点完成，然后同步给镜像节点。

  • 主节点宕机后，镜像节点会替代它成为新的主节点。

  例如，节点1中有一个 q1 队列，节点2也有一个 q1 队列，它们之间建立了主从关系，数据是同步的，都保留了队列和队列中的消息。在这种情况下，节点1创建了队列 q1，因此被称为主节点，而备份此队列的节点2被称为镜像节点。如果节点1、2、3都有自己的队列，相对而言，创建 q3 队列的节点3是主节点，备份 q3 的节点1就是镜像节点。如果某个主节点宕机了，那么镜像节点就会替代它成为新的主节点。

  然而，凡是都有意外。如果节点1中的 q1 发生了修改，还没有来得及同步给节点2，那么就可能丢失数据。尽管这种情况比较罕见，但若要解决此问题，可以使用RabbitMQ中的第三种集群模式：仲裁队列。

  

• 仲裁队列

  仲裁队列是3.8版本以后才有的新功能，用来替代镜像队列，具备下列特征：

  • 与镜像队列一样，都是主从模式，支持主从数据同步

  • 使用非常简单，没有复杂的配置

  • 主从同步基于Raft协议，强一致

  

Kafka

Kafka是如何保证消息不丢失

• Kafka的消息是如何丢失的

  和之前RabbitMQ一样，Kafka的消息也可能会在这些情况下丢失：生产者发送消息到broker；消息在broker中存储；消费者从broker中消费消息。对于这三种可能，Kafka也给出了不同的解决方案。

  

• 生产者发送消息到Broker丢失

  Broker是Kafka的单个服务器，负责接收来自生产者的消息、分配偏移量并将消息提交到磁盘存储。

  • 设置异步发送

    Kafka支持同步和异步发送消息，在异步发送消息时，可以设置其回调和发送失败后的处理方法。这给了我们很大操作空间，能够对失败的消息做记录等操作。

    

  • 消息重试

    如果是网络失败，我们可以设置重试机制

    

• 消息在broker中储存丢失

  • 发送确认机制acks

    • 消息存储的大体流程

      当生产者发送消息到了Broker之后，这些消息都会存储在分区中，而分区又分为两个不同的角色，leader和follower。其中，leader只能有一个，而follower可能有多个。分区接收到消息之后，首先会将数据保存到leader分区中，然后由leader同步到follower中。

    其中 ，确认机制acks有三个值，分别是0，1，2。解释如下：

    确认机制	说明
    acks=0	生产者在成功写入消息之前不会等待任何来自服务器的响应，消息有丢失的风险，但是速度最快。
    acks=1	只要集群首领节点收到消息，生产者就会收到一个来自服务器的成功响应。
    acks=all	只有当所有参与赋值的节点全部收到消息时，生产者才会收到一个来自服务器的成功响应。

    在生产环境中，acks至少也要为1。

• 消费者从Broker中接收消息丢失

  • Kafka中的分区机制指的是将每个主题划分成多个分区（Partition）。

    分区是Kafka中的基本概念，主题被分成多个分区，每个分区是一个独立的日志文件。

  • 每个分区中的消息只能由消费者组中的唯一一个消费者处理，而不同的分区会被分配给不同的消费者（同一个消费者组）。

  一个Kafka集群由多个Broker组成，每个Broker可以理解为一个独立的Kafka实例。在每个Broker中，存在不同的topic，而每个topic可能会有多个分区。这些分区可能存储在不同的Broker中。例如，一个名为T1的topic可能有4个分区P1、P2、P3、P4，其中分区P1和P3存储在Broker1中，而分区P2和P4存储在Broker2中。

  一个消费者组可能包含多个消费者，它们都是消费同一个topic的消息。由于某个topic可能有多个分区，因此消费者组的每个消费者可能会负责消费不同的分区中的消息。

  

  下图中有三个分区P1、P2、P3，它们都属于同一个Topic T1。每个分区都是由顺序的、不可变的消息队列组成，并且可以持续地添加消息。分区中的消息都分配了一个序列号，称为偏移量，它们从0开始自增。在每个分区中，添加数据也是在分区中进行的。

  

  当消费者完成对消息的消费时，会做一个标记，即消费的偏移量。默认情况下，每个消费者会每隔5秒提交一次自己已消费的偏移量。但如果出现重平衡的情况，可能会导致消息被重复消费或者丢失。

  重平衡指的是当消费者组中的某个消费者宕机后，其负责消费的分区需要重新分配给组内其他消费者来继续消费。例如，如果消费者2宕机了，负责消费分区3的任务将被重新分配给其他消费者，如下图中重新分配给了消费者1。这个重新分配的过程就是重平衡。如果宕机的消费者2消费到偏移量6，但提交的偏移量只有3，那么新接手的消费者1会从偏移量3开始继续消费。这可能导致3到6之间的数据被重复消费。或者偏移量只提交到3，消费者只消费到偏移量1，那么1到3之间的数据就会被浪费。

  为了解决重平衡导致的消费数据问题，可以禁用消费者自动提交偏移量的功能，改为手动配置提交偏移量。Kafka提供了两种提交偏移量的方式，一种是同步，一种是异步。通常采用同步+异步组合提交的方式，即在消费完成后设置一个异步提交，在finally块中设置一个同步提交。代码如下：

  

Kafka是如何保证消息的顺序性的

• 应用场景

  • 即时消息中的单对单聊天和群聊，保证发送方消息发送顺序与接收方的顺序一致
  • 充值转账两个渠道在同一个时间进行余额变更，短信通知必须要有顺序

我们知道，消息存储的时候，是按照一定的规则来找到不同的分区来存储的。消费也是一样的 ，所以说并不能保证消息的顺序性。要想保证消息的顺序性，我们可以只提供一个分区或者是相同的业务只在同一个分区下进行存储和消费。 因为同一个分区的偏移量是有顺序的。

在我们发送消息的时候， 调用了send()方法。如果是4个参数，就如代码第一行所示 。如果是3个，就如代码第二行所示。

1
2
3
4

// 指定分区 topic,partition,key,data
kafkaTemplate.send("springboot-kafka-topic", 0, "key-001", "value-0001");	// 如果我们的业务对消息的顺序性有要求，可以指定分区。
// 相同的业务key topic,key,data
kafkaTemplate.send("springboot-kafka-topic", "key-001", "value-0001");	// 如果没有指定分区，那么默认将会根据key来计算Hash，然后根据hash来找到合适的分区。如果我们还是想要指定分区，可以将key设置为一样的。

Kafka高可用机制

• 集群模式

  在集群模式中，如果一台Broker宕机，其他的Broker仍然能够对外提供服务。

• 分区备份机制

  考虑一个由三台Broker组成的集群，某个topic有三个分区：P0, P1, P2。这些分区可能存储在不同的Broker中。生产者发送消息时，可以按照一定策略找到不同的分区进行消息存储。为了提高可用性，Kafka引入了分区备份机制，即每个分区可以有多个副本。例如，P0有3个副本，这些副本存储在不同的Broker中。P1和P2也是一样的，各有3个副本分布在不同的Broker上。这些副本中有一个是Leader，负责写数据，其他的是Follower，它们与Leader保持主从关系。同一个分区下所有副本存储的内容是相同的。当Leader宕机时，系统会从剩余的Follower中选择一个作为新的Leader，以确保服务的持续提供，从而保证了系统的容错性。

  

  • 一个topic有多个分区，每个分区有多个副本，其中有一个leader，其余的是follower，副本存储在不同的broker中

  • 所有的分区副本的内容是都是相同的，如果leader发生故障时，会自动将其中一个follower提升为leader

  在备份副本时，如果一个Follower被标记为ISR（in-sync replica）副本，那么Leader副本在同步数据时是同步请求，而普通的副本则是异步请求。

  因此，如果Leader失效后，需要选举新的Leader，选举原则如下：

  1. 优先从ISR列表中选取，因为这个列表中的Follower的数据是与Leader同步的。

  2. 如果ISR列表中的Follower都不可用，则只能从其他Follower中选取新的Leader。

  

  在brocker配置中我们可以设置topic副本的数量，也可以设置ISR的最小个数。

  

Kafka数据清理机制

• Kafka文件储存机制

  在Kafka中，Topic只是一个逻辑概念，实际上数据存储在分区中。当向某个Topic发送消息时，每个分区都对应一个文件夹，命名方式为topic-name-partition-number，例如itheima-0, itheima-1, itheima-2。在这些文件夹内部，存储了分区文件。

  每个分区文件也是分段存储的，即由多个日志文件段（segment）组成。每个segment对应三个文件：index（索引）、log（数据）、timeindex（时间索引）。这种分段存储的方式有助于快速删除无用文件，提高磁盘效率，并且使查找数据更加高效便捷。

  

• 数据清理机制

  日志的清理机制有两个

  1. 根据消息的保留时间，当消息在Kafka中保存的时间超过了指定的时间（默认168小时），就会触发清理过程

     

  2. 根据topic存储的数据大小，当topic所占的日志文件大小大于一定的值，则开始删除最久的消息。需手动开启。

     

Kafka的高性能设计

• 摘要

  • 消息分区：消息分区不受单台服务器的限制，能够处理更多的数据。
  • 顺序读写：使用磁盘顺序读写，提升读写效率。
  • 页缓存：将磁盘中的数据缓存到内存中，将对磁盘的访问变为对内存的访问。
  • 零拷贝：减少上下文切换和数据拷贝的操作。
  • 消息压缩：通过消息压缩，减少磁盘IO和网络IO的开销。
  • 分批发送：将消息打包批量发送，降低网络开销。

• 零拷贝

  在Linux系统中，空间分为用户空间和内核空间，用户空间权限较低，而内核空间权限较高。当一个Kafka服务运行时，它属于用户空间。

  • 正常拷贝：

    • 生产者发送消息：当生产者发送消息时，数据需要存储到磁盘文件中。但由于用户空间没有权限直接进行磁盘读写操作，因此数据会先被拷贝到内核空间进行处理。内核空间中的页缓存负责缓存数据，当达到一定批次时，数据会被写入磁盘文件，完成消息的保存。

    • 消费者消费消息：当消费者需要消费消息时，首先会在用户空间的Kafka中查找消息。如果消息不在用户空间中，则会从磁盘文件中读取，并将消息拷贝到页缓存中，然后再从页缓存中拷贝到用户空间的Kafka中。

    • 消息发送给消费者：当需要将消息发送给消费者时，会使用Socket接口和网卡进行数据传输。此时，消息会从用户空间的Kafka拷贝到内核空间的Socket缓冲区，最后再被拷贝到网卡中，由网卡传送给消费者。

    

  • 零拷贝

    首先消费者需要消费数据，Kafka会检查页缓存中是否有数据。如果没有，Kafka会从磁盘文件中拷贝数据到页缓存中。但是，由于Kafka已经知道要将数据发送给哪个消费者，因此它直接委托操作给系统，让系统直接将数据从页缓存中拷贝给网卡，然后传输给对应的消费者。这样，数据拷贝次数减少，提升了性能。