【年后找工作】Java八股必备知识 -- MQ篇（RabbitMQ）

1、什么是MQ？你能介绍一下么

MQ 是消息队列（Message Queue）的简称，是一种应用程序间进行异步通信的技术。它允许不同的应用程序通过发送和接收消息来进行解耦和协作。

消息队列的基本结构是一个中心化的消息中间件（Message Broker），它负责接收、存储和转发消息。应用程序可以向消息队列发送消息，而其他应用程序可以从队列中接收这些消息并进行处理。

使用消息队列的好处包括：

1. 异步通信：发送方将消息放入队列后即可继续处理其他任务，不需要等待接收方的响应。接收方可以在合适的时候从队列中获取消息进行处理。
2. 解耦和松散耦合：应用程序之间通过消息进行通信，发送方和接收方不需要直接知道彼此的存在，从而实现了解耦和松散耦合。
3. 增强系统可靠性：即使某个应用程序出现故障或暂时不可用，消息仍然可以被存储在队列中，等待被处理。
4. 缓冲和削峰填谷：当发送方的请求量大于接收方的处理能力时，消息队列可以作为缓冲区，平衡请求和处理的速度差异。
5. 扩展性：通过增加多个消费者来处理消息队列中的消息，可以实现系统的横向扩展。

常见的消息队列系统包括 RabbitMQ、Apache Kafka、ActiveMQ 等。它们提供了丰富的特性，如持久化、消息分发、消息过滤、事务支持等，以满足不同场景下的需求。

总而言之，消息队列是一种重要的通信机制，可用于实现应用程序之间的异步通信、解耦和削峰填谷等功能，提高系统的可靠性和可扩展性。

2、为什么要使用消息队列呢？

消息队列主要有三大用途，我们拿一个电商系统的下单举例：
消息队列在电商系统下单场景中的三大主要用途是解耦、异步和削峰。

1. 解耦：引入消息队列后，订单完成之后可以将消息放入队列中，不需要订单服务直接调用库存服务、营销服务等下游服务来处理。每个下游服务可以独立消费消息，实现了订单服务与其他服务的解耦，提高了系统的灵活性和可维护性。
2. 异步：在订单支付完成后，需要进行一系列的操作，如减库存、增加积分、发送消息等。引入消息队列后，除了必须同步更新订单状态外，其他操作可以通过消息队列异步处理。这样可以降低响应时间，提高用户体验，同时减少了对下游服务的依赖。
3. 削峰：在特定场景下，如秒杀活动，可能会出现突发的大量请求。如果直接让服务器、数据库和缓存等承受全部流量，容易导致系统负荷过大，甚至崩溃。通过将请求放入消息队列中，只允许系统处理其能够处理的流量，其他流量则暂时保存在队列中，从而实现了削峰的效果，保护了系统的稳定性。

总的来说，消息队列在电商系统下单场景中的解耦、异步和削峰三大作用，可以提高系统的可靠性、可扩展性和响应能力。

3、RabbitMQ的整体架构是怎么样的？

RabbitMQ是一个开源的消息中间件，用于在应用程序之间传递消息。它实现了AMQP（高级消息队列协议）并支持其他消息传递协议，例如STOMP（简单文本定向消息协议）和MQTT（物联网协议）。
他的整体架构大致如下：

组成部分说明：

• Broker：消息队列服务进程，此进程包括两个部分：Exchange和Queue
• Exchange：消息队列交换机，按一定的规则将消息路由转发到某个队列，对消息进行过虑。
• Queue：消息队列，存储消息的队列，消息到达队列并转发给指定的
• Producer：消息生产者，即生产方客户端，生产方客户端将消息发送产方客户端将消息发送
• Consumer：消息消费者，即消费方客户端，接收MQ转发的消息。

4、RabbitMQ是怎么做消息分发的？

RabbitMQ实现消息分发的机制主要是通过交换机（Exchange）和队列（Queue）之间的绑定关系来实现的。当生产者将消息发送到RabbitMQ时，消息首先会被发送到一个交换机，然后根据交换机与队列之间的绑定规则，最终路由到一个或多个队列，再由消费者从队列中接收消息进行处理。

RabbitMQ一共有6种工作模式（消息分发方式）分别是简单模式、工作队列模式、发布订阅模式、路由模式、主题模式以及RPC模式。
逐一为您介绍这六种 RabbitMQ 的工作模式：

1. 简单模式（Simple Mode）：- 在简单模式下，消息会被发送到队列，然后消费者从队列中接收并处理消息。- 这是最基本的消息传递模式，适用于一个生产者和一个消费者的简单场景。
2. 工作队列模式（Work Queue Mode）：- 在工作队列模式下，多个消费者监听同一个队列，竞争性地接收和处理消息。- 每条消息只会被一个消费者处理，确保消息的顺序处理。- 这种模式适用于多个消费者并发处理消息的情况，提高了系统的处理能力和吞吐量。
3. 发布订阅模式（Publish/Subscribe Mode）：- 在发布订阅模式下，消息被发送到交换机，然后交换机将消息广播给所有绑定的队列。- 每个消费者都会接收到相同的消息副本，适用于消息广播的场景。- 一条消息被多个消费者同时接收和处理
4. 路由模式（Routing Mode）：- 在路由模式下，消息被发送到交换机，并根据指定的路由键（Routing Key）被路由到对应的队列。- 消费者可以选择性地接收特定路由键的消息，实现消息的有选择性路由。- 这种模式适用于根据不同的条件将消息发送到不同的队列中，以实现消息的筛选和分发。
5. 主题模式（Topic Mode）：- 在主题模式下，消息被发送到交换机，并使用主题进行匹配，然后被路由到符合条件的队列。- 主题可以使用通配符进行模糊匹配，适用于需要灵活匹配消息的场景。- 这种模式适用于消息的复杂路由需求，可以实现高度灵活的消息筛选和分发。
6. RPC 模式（Remote Procedure Call Mode）：- 在 RPC 模式下，客户端发送请求消息到队列，并等待服务器返回响应。- 服务器接收请求、处理任务，并将响应发送回客户端指定的回调队列，实现远程过程调用。

这些工作模式提供了丰富的选择，以满足不同场景下的消息处理需求。通过灵活地配置交换机、队列和绑定关系，可以实现高效可靠的消息分发和处理。

5、RabbitMQ如何实现延迟消息？

RabbitMQ中是可以实现延迟消息的，一般有两种方式，分别是通过死信队列以及通过延迟消息插件来实现。

6、那你说说死信队列如何实现延迟消息

当RabbitMQ中的一条正常的消息，因为过了存活时间（TTL过期）、队列长度超限、被消费者拒绝等原因无法被消费时，就会变成Dead Message，即死信。

死信队列（Dead Letter Queue）是一种在消息队列系统中常见的概念，用于处理无法被消费的消息。当消息满足一定条件时，可以被标记为“死信”，然后被路由到死信队列（其实是交换机-exchange）中进行进一步处理。

在 RabbitMQ 中，死信队列通常与 TTL（Time-To-Live，消息过期时间）和消息重试机制结合使用。当消息过期或者被拒绝消费时，就会被发送到死信队列中，而不是直接丢弃。这样可以方便地对死信消息进行监控、分析和处理，提高消息处理的可靠性和灵活性。

那么基于这样的机制，就可以实现延迟消息了。

给一个消息设定TTL，但是并不消费这个消息，等他过期，过期后就会进入到死信队列，然后我们再监听死信队列的消息消费就行了。

当消息变成"死信"之后，如果配置了死信队列，它将被发送到死信交换机，死信交换机将死信投递到一个队列上，这个队列就是死信队列。但是如果没有配置死信队列，那么这个消息将被丢弃。

RabbitMQ中的这个TTL是可以设置任意时长的，这相比于RocketMQ只支持一些固定的时长而显得更加灵活一些。

作用
RabbitMQ的死信队列其实有很多作用，比如我们可以借助他实现延迟消息，进而实现订单的到期关闭，超时关单等业务逻辑。

注意
但是，死信队列的实现方式存在一个问题，那就是可能造成队头阻塞，因为队列是先进先出的，而且每次只会判断队头的消息是否过期，那么，如果队头的消息时间很长，一直都不过期，那么就会阻塞整个队列，这时候即使排在他后面的消息过期了，那么也会被一直阻塞。

基于RabbitMQ的死信队列，可以实现延迟消息，非常灵活的实现定时关单，并且借助RabbitMQ的集群扩展性，可以实现高可用，以及处理大并发量。他的缺点第一是可能存在消息阻塞的问题，还有就是方案比较复杂，不仅要依赖RabbitMQ，而且还需要声明很多队列出来，增加系统的复杂度

代码示例

配置死信队列
在 RabbitMQ 中设置死信队列通常需要以下步骤：

1. 创建死信队列：定义一个用于存储死信消息的队列。
2. 创建死信交换机：为死信队列定义一个交换机，通常是一个 direct 类型的交换机。
3. 将队列与死信交换机绑定：将主要队列和死信交换机绑定，以便无法处理的消息能够被转发到死信队列。
4. 在主要队列上设置死信属性：通过设置队列的 x-dead-letter-exchange 和 x-dead-letter-routing-key 属性，指定死信消息应该被发送到哪个交换机和路由键。

当消息被标记为死信时，它将被发送到死信队列，并可以由应用程序进一步处理、审查或记录。这种机制有助于增加消息处理的可靠性和容错性，确保不丢失重要的消息，并提供了一种处理失败消息的方式。

importorg.springframework.amqp.core.*;importorg.springframework.context.annotation.Bean;importorg.springframework.context.annotation.Configuration;@ConfigurationpublicclassRabbitMQConfig{// 配置死信队列和交换机@BeanpublicDirectExchangedeadLetterExchange(){returnnewDirectExchange("dead-letter-exchange");}// 死信队列@BeanpublicQueuedeadLetterQueue(){returnnewQueue("dead-letter-queue");}// 绑定死信队列到死信交换机@BeanpublicBindingdeadLetterBinding(){returnBindingBuilder.bind(deadLetterQueue()).to(deadLetterExchange()).with("dead-letter-routing-key");}// 主队列的交换机@BeanpublicDirectExchangemainExchange(){returnnewDirectExchange("main-exchange");}// 主队列@BeanpublicQueuemainQueue(){Map<String,Object> args =newHashMap<>(2);// 声明当前队列绑定的死信交换机 
                args.put("x-dead-letter-exchange",DEAD_LETTER_EXCHANGE);// 这里声明当前队列的死信路由key 
                args.put("x-dead-letter-routing-key","dead-letter-routing-key");returnQueueBuilder.durable("main-queue").withArguments(args).build();}// 绑定主队列到主交换机@BeanpublicBindingbinding(){returnBindingBuilder.bind(mainQueue()).to(mainExchange()).with("main-routing-key");}}

这样，消费者在消费的时候，分别监听主队列和死信队列就可以了：

@ComponentpublicclassDeadLetterMessageReceiver{@RabbitListener(queues ="dead-letter-queue")publicvoidreceiveA(Message message,Channel channel)throwsIOException{System.out.println("收到死信消息："+newString(message.getBody())); 
      channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);}}@ComponentpublicclassMainMessageReceiver{@RabbitListener(queues ="main-queue")publicvoidreceiveA(Message message,Channel channel)throwsIOException{System.out.println("收到普通消息A："+newString(message.getBody())); 
      channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);}}

7、RabbitMQ如何实现消费端限流

什么是消费端限流，这是一种保护消费者的手段，假如说，现在是业务高峰期了，消息有大量堆积，导致MQ消费者需要不断地进行消息消费，很容易被打挂，甚至重启之后还是会被大量消息涌入，继续被打挂。

为了解决这个问题，RabbitMQ提供了basicQos的方式来实现消费端限流。我们可以在消费者端指定最大的未确认消息数，当达到这个限制时，RabbitMQ将不再推送新的消息给消费者，直到有一些消息得到确认。

想要实现这个功能，首先需要把自动提交关闭。

channel.basicConsume(queueName, false, consumer);

接着进行限流配置：

/**
  * 限流设置:  
  * prefetchSize：每条消息大小的设置，0是无限制
  * prefetchCount:标识每次推送多少条消息
  * global:false标识channel级别的  true:标识消费者级别的
  */
 channel.basicQos(0,10,false);

如以上配置，可以实现消费者在处理完一条消息后，才会获取下一条消息。

然后再在消费者处理完一条消息之后，手动发送确认消息给到RabbitMQ，这样就可以拉取下一条消息了：

channel.basicAck(deliveryTag, false); // 发送确认

完整代码如下：

importcom.rabbitmq.client.*;publicclassConsumerWithFlowControl{privatestaticfinalStringQUEUE_NAME="my_queue";privatestaticfinalStringHOST="localhost";publicstaticvoidmain(String[] args)throwsException{ConnectionFactory factory =newConnectionFactory();
        factory.setHost(HOST);try(Connection connection = factory.newConnection();Channel channel = connection.createChannel()){// 声明队列
            channel.queueDeclare(QUEUE_NAME,false,false,false,null);// 设置消费者限流，每次只获取一条消息int prefetchCount =1;
            channel.basicQos(prefetchCount);// 创建消费者DefaultConsumer consumer =newDefaultConsumer(channel){@OverridepublicvoidhandleDelivery(String consumerTag,Envelope envelope,AMQP.BasicProperties properties,byte[] body)throwsIOException{String message =newString(body,"UTF-8");System.out.println("Received: "+ message);// 模拟消息处理耗时try{Thread.sleep(1000);}catch(InterruptedException e){Thread.currentThread().interrupt();}// 发送消息确认
                    channel.basicAck(envelope.getDeliveryTag(),false);}};// 指定队列，并关闭自动确认
            channel.basicConsume(QUEUE_NAME,false, consumer);}}}

8、RabbitMQ如何防止重复消费？

RabbitMQ的消息消费是有确认机制的，正常情况下，消费者在消息消费成功后，会发送一个确认消息，消息队列接收到之后，就会将该消息从消息队列中删除，下次也就不会再投递了。

但是如果存在网络延迟的问题，导致确认消息没有发送到消息队列，导致消息重投了，是有可能，所以，当我们使用MQ的时候，消费者端自己也需要做好幂等控制来防止消息被重复消费。

一般来说，处理这种幂等问题就是我们提过无数次的’一锁、二判、三更新’

也就是说我们在发送消息是需要生成一个唯一的标识并且把它放到消息体中，根据这个标识就可以判断两次消息是不是同一条。这样我们在消费者端，接收到消息以后，只需要解析出消息体中的这个唯一标识，就可以通过’一锁、二判、三更新’的方式来判断是否消费成功过了。

"一锁、二判、三更新"是一种常用的实现幂等性的方法，它包括以下三个步骤：

1. 一锁（First Locking）：- 在消息处理之前，需要获取一个用于保证操作原子性的锁。- 这个锁可以是分布式锁，比如基于 Redis、ZooKeeper 等工具实现的锁，也可以是数据库的行级锁或乐观锁等。- 通过获取锁，确保同一时刻只有一个消费者能够进行消息处理，避免多个消费者并发处理同一条消息。
2. 二判（Second Checking）：- 获取到锁之后，在处理消息之前进行二次检查，判断消息是否已经被处理过。- 这个判断可以根据唯一标识或其他状态信息来进行，比如检查数据库中是否存在相同的记录或者检查某个标识字段的状态。- 如果判断消息已经被处理过，则可以直接放弃处理当前消息，避免重复操作。
3. 三更新（Third Updating）：- 如果经过二次检查后确认消息尚未被处理过，就执行消息的处理逻辑。- 在处理完成后，需要完成相应的更新操作，以便标记消息已被成功处理。- 这个更新可以是更新数据库记录、修改状态字段、更新缓存等，根据实际情况选择合适的方式。

通过一锁、二判、三更新的步骤，可以确保在分布式环境下对消息进行幂等性处理。获取锁保证了同一时刻只有一个消费者能够处理消息，二次检查避免了重复处理已被处理过的消息，三次更新标记消息已被成功处理。这样可以有效防止重复消费带来的副作用。

9、如何保障消息一定能发送到RabbitMQ

当我们作为一个消费发送方，如何保证我们给RabbitMQ发送的消息一定能发送成功，如何确保他一定能收到这个消息呢？

我们知道，RabbitMQ的消息最终时存储在Queue上的，而在Queue之前还要经过Exchange，那么这个过程中就有两个地方可能导致消息丢失。第一个是Producer到Exchange的过程，第二个是Exchange到Queue的过程。为了解决这个问题，有两种方案，一种是通过confirm机制，另外一种是事务机制，因为事务机制并不推荐，这里先介绍Confirm机制。

Publisher Confirm是一种机制，用于确保消息已经被Exchange成功接收和处理。一旦消息成功到达Exchange并被处理，RabbitMQ会向消息生产者发送确认信号（ACK）。如果由于某种原因（例如，Exchange不存在或路由键不匹配）消息无法被处理，RabbitMQ会向消息生产者发送否认信号（NACK）

// 启用Publisher Confirms
channel.confirmSelect();// 设置Publisher Confirms回调
channel.addConfirmListener(newConfirmListener(){@OverridepublicvoidhandleAck(long deliveryTag,boolean multiple)throwsIOException{System.out.println("Message confirmed with deliveryTag: "+ deliveryTag);// 在这里处理消息确认}@OverridepublicvoidhandleNack(long deliveryTag,boolean multiple)throwsIOException{System.out.println("Message not confirmed with deliveryTag: "+ deliveryTag);// 在这里处理消息未确认}});

Publisher Returns机制与Publisher Confirms类似，但用于处理在消息无法路由到任何队列时的情况。当RabbitMQ在无法路由消息时将消息返回给消息生产者，但是如果能正确路由，则不会返回消息。

// 启用Publisher Returns
channel.addReturnListener(newReturnListener(){@OverridepublicvoidhandleReturn(int replyCode,String replyText,String exchange,String routingKey,AMQP.BasicProperties properties,byte[] body)throwsIOException{System.out.println("Message returned with replyCode: "+ replyCode);// 在这里处理消息发送到Queue失败的返回}});

通过以上方式，我们注册了两个回调监听，用于在消息发送到Exchange或者Queue失败时进行异常处理。通常我们可以在失败时进行报警或者重试来保障一定能发送成功。

10、RabbitMQ如何保证消息不丢

介绍了如何确保RabbitMQ的发送者把消息能够投递给RabbitMQ的Exchange和Queue，那么，Queue又是如何保证消息能不丢的呢？

RabbitMQ在接收到消息后，默认并不会立即进行持久化，而是先把消息暂存在内存中，这时候如果MQ挂了，那么消息就会丢失。所以需要通过持久化机制来保证消息可以被持久化下来。

队列和交换机的持久化

在声明队列时，可以通过设置durable参数为true来创建一个持久化队列。持久化队列会在RabbitMQ服务器重启后保留，确保队列的元数据不会丢失。

在声明交换机时，也可以通过设置durable参数为true来创建一个持久化交换机。持久化交换机会在RabbitMQ服务器重启后保留，以确保交换机的元数据不会丢失。

绑定关系通常与队列和交换机相关联。当创建绑定关系时，还是可以设置durable参数为true，以创建一个持久化绑定。持久化绑定关系会在服务器重启后保留，以确保绑定关系不会丢失。

@BeanpublicQueueTestQueue(){// 第二个参数durable:是否持久化,默认是falsereturnnewQueue("queue-name",true,true,false);}@BeanpublicDirectExchangemainExchange(){//第二个参数durable:是否持久化,默认是falsereturnnewDirectExchange("main-exchange",true,false);}

持久化消息

生产者发送的消息可以通过设置消息的deliveryMode为2来创建持久化消息。持久化消息在发送到持久化队列后，将在服务器重启后保留，以确保消息不会丢失。

deliveryMode是一项用于设置消息传递模式的属性，用于指定消息的持久性级别。deliveryMode可以具有两个值：
α、1（非持久化）：这是默认的传递模式。如果消息被设置为非持久化，RabbitMQ将尽力将消息传递给消费者，但不会将其写入磁盘，这意味着如果RabbitMQ服务器在消息传递之前崩溃或重启，消息可能会丢失。
β、2（持久化）：如果消息被设置为持久化，RabbitMQ会将消息写入磁盘，以确保即使在RabbitMQ服务器重启时，消息也不会丢失。持久化消息对于重要的消息非常有用，以确保它们不会在传递过程中丢失

Message message =MessageBuilder.withBody("hello, spring".getBytes(StandardCharsets.UTF_8))//kp 消息体，字符集.setDeliveryMode(MessageDeliveryMode.PERSISTENT).build();

rabbitTemplate.convertAndSend("simple.queue", message);

通过设置deliveryMode类实现消息的持久化。但是需要注意，将消息设置为持久化会增加磁盘I/O开销。

消费者确认机制

有了持久化机制后，那么怎么保证消息在持久化下来之后一定能被消费者消费呢？这里就涉及到消息的消费确认机制。

在RabbitMQ中，消费者处理消息成功后可以向MQ发送ack回执，MQ收到ack回执后才会删除该消息，这样才能确保消息不会丢失。如果消费者在处理消息中出现了异常，那么就会返回nack回执，MQ收到回执之后就会重新投递一次消息，如果消费者一直都没有返回ACK/NACK的话，那么他也会在尝试重新投递。

无法做到100%不丢

虽然我们通过发送者端进行异步回调、MQ进行持久化、消费者做确认机制，但是也没办法保证100%不丢，因为MQ的持久化过程其实是异步的。即使我们开了持久化，也有可能在内存暂存成功后，异步持久化之前宕机了，那么这个消息就会丢失。

如果想要做到100%不丢失，就需要引入本地消息表，来通过轮询的方式来进行消息重投。

11、介绍下RabbitMQ的事务机制

想要保证发送者一定能把消息发送给RabbitMQ，一种是通过confirm机制，另外一种就是通过事务机制。

RabbitMQ的事务机制，允许生产者将一组操作打包成一个原子事务单元，要么全部执行成功，要么全部失败。事务提供了一种确保消息完整性的方法，但需要谨慎使用，因为它们对性能有一定的影响。

RabbitMQ是基于AMQP协议实现的，RabbitMQ中，事务是通过在通道（Channel）上启用的，与事务机制有关的方法有三个：
●txSelect()：将当前channel设置成transaction模式。
●txCommit()：提交事务。
●txRollback()：回滚事务。

我们需要先通过txSelect开启事务，然后就可以发布消息给MQ了，如果txCommit提交成功了，则消息一定到达了RabbitMQ，如果在txCommit执行之前RabbitMQ实例异常崩溃或者抛出异常，那我们就可以捕获这个异常然后执行txRollback进行回滚事务。

所以， 通过事务机制，我们也能保证消息一定可以发送给RabbitMQ。

以下，是一个通过事务发送消息的方法示例：

importcom.rabbitmq.client.*;publicclassRabbitMQTransactionExample{publicstaticvoidmain(String[] args)throwsException{ConnectionFactory factory =newConnectionFactory();
        factory.setHost("localhost");try(Connection connection = factory.newConnection();Channel channel = connection.createChannel()){// 启用事务
            channel.txSelect();String exchangeName ="my_exchange";String routingKey ="my_routing_key";try{// 发送第一条消息String message1 ="Transaction Message 1";
                channel.basicPublish(exchangeName, routingKey,null, message1.getBytes());// 发送第二条消息String message2 ="Transaction Message 2";
                channel.basicPublish(exchangeName, routingKey,null, message2.getBytes());// 模拟一个错误int x =1/0;// 提交事务（如果没有发生错误）
                channel.txCommit();System.out.println("Transaction committed.");}catch(Exception e){// 发生错误，回滚事务
                channel.txRollback();System.err.println("Transaction rolled back.");}}}}

本文参考自二哥的Java进阶之路、三分恶 作者老三、Hollis Java 8Gux、ChatGPT。