消息中间件

消息中间件面试题

该笔记基于b站黑马程序员Java面试题视频制作

一. RabbitMQ

1. RabbitMQ如何保证消息不丢失

异步发送（验证码、短信、邮件…）
MYSQL和Redis , ES之间的数据同步
分布式事务
削峰填谷
…

生产者确认机制

RabbitMQ提供了publisher confirm机制来避免消息发送到MQ过程中丢失。消息发送到MQ以后，会返回一个结果给发送者，表示消息是否处理成功

消息失败之后的处理方法

回调日志及时重新发送
记录日志
保存到数据库然后定时重新发送，成功发送后即刻删除表中的数据

消息持久化

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

交换机持久化

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

队列持久化

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

消息持久化，SpringAMQP中的消息默认是成就的，可以通过MessageProperties中的DeliveryMode来指定

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

消费者确认

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

manual：手动ack，需要在业务代码结束后，调用api发送ack。
auto：自动ack，由spring监测listener代码是否出现异常，没有异常则返回ack；抛出异常则返回nack
none：关闭ack，MQ假定消费者获取消息后会成功处理，因此消息投递后立即被删除

我们可以利用Spring的retry机制，在消费者出现异常时利用本地重试，设置重试次数，当次数达到了以后，如果消息依然失败，将消息投递到异常交换机，交由人工处理

RabbitMQ如何保证消息不丢失？

开启生产者确认机制，确保生产者的消息能到达队列
开启持久化功能，确保消息未消费前在队列中不会丢失开启消费者确认机制为auto，由spring确认消息处理成功后完成ack
开启消费者失败重试机制，多次重试失败后将消息投递到异常交换机，交由人工处理

2. RabbitMQ消息的重复消费问题

网络抖动
消费者挂了

都可能出现出现消费问题

解决方案：

每条消息设置一个唯一的标识id
幂等方案：分布式锁、数据库锁（乐观锁、悲观锁）

3. RabbitMQ中的死信交换机(RabbitMQ 延迟队列)

延迟队列：进入队列的消息会被延迟消费的队列
场景：超时订单、限时优惠、定时发布

延迟队列 = 死信交换机 + TTL

死信交换机

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

消费者使用basic.reject或 basic.nack声明消费失败，并且消息的requeue参数设置为false
消息是一个过期消息，超时无人消费
要投递的队列消息堆积满了，最早的消息可能成为死信

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

@Bean
public Queue ttlQueue() {
    return QueueBuilder.durable("simple.queue") // 指定队列名称，并持久化
        .ttl(10000) // 设置队列的超时时间，10秒
        .deadLetterExchange("dl.direct") // 指定死信交换机
        .build();
}

TTL

TTL，也就是Time-To-Live。如果一个队列中的消息TTL结束仍未消费，则会变为死信，ttl超时分为两种情况:

消息所在的队列设置了存活时间
消息本身设置了存活时间

// 创建消息
Message message = MessageBuilder
    .withBody("hello, ttl message").getBytes(StandardCharsets.UTF_8)
    .setExpiration("500")
    .build();
// 消息Id，需要封装到CorrelationData中
CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
// 发送消息
rabbitTemplate.convertAndSend("ttl.direct", "ttl", message, correlationData);

延迟队列插件

DelayExchange插件，需要安装在RabbitMQ中

RabbitMQ有一个官方的插件社区，地址为：https://www.rabbitmq.com/community-plugins.html

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

@RabbitListener(bindinds = @QueueBinding(
	value = @Queue(name = "delay.queue", durable = "true"),
    exchange = @Exchange(name = "delay.direct", delayed = "true"),
    key = "delay"
))
public void listenDelayedQueue(String msg) {
	log.info("接收到 delay.queue的延迟消息：{}", msg);
}

// 创建消息
Message message = MessageBuilder
    .withBody("hello,delayed message".getBytes(StandardCharsets.UTF_8))
    .setHeader("x-delay", 10000)
    .build();
// 消息ID，需要封装到CorrelationData中
CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
// 发送消息
rabbitTemplate.convertAndSend("delay.direct", "delay", message, correlationData);

RabbitMQ中死信交换机？（RabbitMQ延迟队列有了解过吗？）

我们当时一个什么业务使用到了延迟队列（超时订单、限时优惠、定时发布…）
其中延迟队列就用到了死信交换机和TTL（消息存活时间）实现的
消息超时未消费就会变成死信（死信的其他情况：拒绝被消费，队列满了）

延迟队列插件实现延迟队列DelayExchange
声明一个交换机，添加delayed属性为true
发送消息时，添加x-delay头，值为超时时间

4. 消息堆积怎么解决

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

解决消费堆积问题有三种思路：

增加更多的消费者，提高消费速度
在消费者内开启线程，也加快消费者的处理速度
扩大队列容积，提高堆积上限

惰性队列

惰性队列的特征：

接收到消息之后直接存入磁盘而非内存中
消费者要消费消息时才会从磁盘中读取并加载到内存
支持数百万条的消息存储

@Bean
public Queue lazyQueue() {
    return QueueBuilder
        .durable(lazy.queue)
        .lazy()
        .build();
}

@RabbitListener(queuesToDeclare = @Queue(
    name = "lzay.queue",
    durable = "true",
    arguments = @Argument(name = "x-queue-mode", value = "lazy")
))
public void listenLazyQueue(String msg) {
	log.info("接收到 lazy.queue的消息:{}", msg);
}

RabbitMQ如果有100万消息堆积在MQ，如何解决（消息堆积怎么解决）

解决消息堆积有三种种思路：

增加更多消费者，提高消费速度
在消费者内开启线程池加快消息处理速度
扩大队列容积，提高堆积上限，采用惰性队列
- 在声明队列的时候可以设置属性x-queue-mode为lazy，即为惰性队列
- 基于磁盘存储，消息上限高
- 性能比较稳定，但基于磁盘存储，受限于磁盘IO，时效性会降低

5. RabbitMQ的高可用机制

在生产环境下，使用集群来保证高可用性
普通集群、镜像集群、仲裁集群

普通集群

普通集群，或者叫标准集群，具备下列特征：

会在集群的各个节点间共享部分数据，包括：交换机、队列元信息。不包含队列中的消息。
当访问集群某节点时，如果队列不在该节点，会从数据所在节点传递到当前节点并返回
队列所在节点宕机，队列中的消息就会丢失

镜像集群

镜像集群：本质是主从模式，具备下面的特征：

交换机、队列、队列中的消息会在各个mq的镜像节点之间同步备份。
创建队列的节点被称为该队列的主节点，备份到的其它节点叫做该队列的镜像节点。
一个队列的主节点可能是另一个队列的镜像节点
所有操作都是主节点完成，然后同步给镜像节点
主宕机后，镜像节点会替代成新的主

仲裁集群

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

与镜像队列一样，都是主从模式，支持主从数据同步
使用非常简单，没有复杂的配置
主从同步基于Raft协议，强一致

@Bean
public Queue quorumQueue() {
    return QueueBuilder
        .durable("quorumqueue") // 持久化
        .quorum() // 仲裁队列
        .build();
}

RabbitMQ的高可用机制有了解过吗？

在生产环境下，我们当时采用的镜像模式搭建的集群，共有3个节点
镜像队列结构是一主多从（从就是镜像），所有操作都是主节点完成，然后同步给镜像节点
主宕机后，镜像节点会替代成新的主（如果在主从同步完成前，主就已经宕机，可能出现数据丢失）

那么出现丢数据怎么解决？

我们可以采用仲裁队列，与镜像队列一样，都是主从模式，支持主从数据同步，主从同步基于Raft协议，强一致。
并且使用起来也非常简单，不需要额外的配置，在声明队列的时候只要指定这个是仲裁队列即可

二. Kafka

1. Kafka是如何保证消息不丢失

使用Kafka在消息的收发过程都会出现消息丢失 , Kafka分别给出了解决方案

生产者发送消息到Brocker丢失
消息在Brocker中存储丢失
消费者从Brocker接收消息丢失

//同步发送
RecordMetadata recordMetadata = kafkaProducer.send(record).get();
//异步发送
kafkaProducer.send(record, new Callback() {
    @Override
    public void onCompletion(RecordMetadata recordMetadata, Exception e) {
        if (e != null) {
            System.out.println("消息发送失败 | 记录日志");
        }
        long offset = recordMetadata.offset();
        int partition = recordMetadata.partition();
        String topic = recordMetadata.topic();
    }
});

1 2	`//设置重试次数 prop.put(ProducerConfig.RETRIES_CONFIG,10);`

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

消费者从Brocker接受消息丢失

禁用自动提交偏移量，改为手动

同步提交
异步提交
同步+异步组合提交

try {
    while (true) {
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
        for (ConsumerRecord<String, String> record : records) {
            System.outprintln(record.value());
            System.outprintln(record.key());
        }
        consumer.commitAsync();
    }
} catch (Exception e) {
    e.printStackTrace();
    System.out.println("记录错误消息: " + e);
} finally {
    try {
        consumer.commitSync();
    } finally {
        consumer.close();
    }
}

Kafka是如何保证消息不丢失?

需要从三个层面去解决这个问题：

生产者发送消息到Brocker丢失
- 设置异步发送，发送失败使用回调进行记录或重发
- 失败重试，参数配置，可以设置重试次数
消息在Brocker中存储丢失

发送确认acks，选择all，让所有的副本都参与保存数据后确认
消费者从Brocker接收消息丢失
- 关闭自动提交偏移量，开启手动提交偏移量
- 提交方式，最好是同步+异步提交

Kafka中消息的重复消费问题如何解决的

关闭自动提交偏移量，开启手动提交偏移量
提交方式，最好是同步+异步提交
幂等方案

2. Kafka是如何保证消费的顺序性

应用场景

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

消费者从Brocker接收消息丢失

// 指定分区
kafkaTemplate.send("springboot-kafka-topic", 0, "key-001", "value-0001");
// 相同的业务key
kafkaTemplate.send("springboot-kafka-topic", "key-001", "value-0001");

topic分区中消息只能由消费者组中的唯一一个消费者处理，所以消息肯定是按照先后顺序进行处理的。但是它也仅仅是保证Topic的一个分区顺序处理，不能保证跨分区的消息先后处理顺序。所以，如果你想要顺序的处理Topic的所有消息，那就只提供一个分区。

Kafka是如何保证消费的顺序性？

问题原因：
一个topic的数据可能存储在不同的分区中，每个分区都有一个按照顺序的存储的偏移量，如果消费者关联了多个分区不能保证顺序性

解决方案：