Rabbitmq 面试题

1. 结构

在这里插入图片描述
消息队列的作用：解耦、异步和削峰。RabbitMQ 的设计使得它能够高效地处理大量消息，并支持多种消息传递模式，非常适合于分布式系统中的异步通信。

Producer: 发送消息的应用程序或服务。生产者将消息发送到交换机。
Exchange（交换机）: 负责接收来自生产者的消息，并根据路由规则将其转发到一个或多个队列。RabbitMQ 支持几种类型的交换机： - Direct Exchange: 精确匹配路由键的队列。- Fanout Exchange: 将消息广播到所有绑定的队列。- Topic Exchange: 基于主题模式（通配符）进行路由。- Headers Exchange: 使用消息头进行路由。
Queue: 存储消息的地方。队列中的消息按照先进先出（FIFO）顺序处理。
Consumer: 从队列中接收和处理消息的应用程序或服务。消费者可以从一个或多个队列中获取消息。
Binding（绑定）: 定义交换机和队列之间的关系。通过绑定，交换机知道如何将消息路由到队列。
Virtual Hosts（虚拟主机）: 提供了逻辑上的隔离，可以在同一个 RabbitMQ 实例中创建多个独立的环境。
Management Plugin（管理插件）: 提供Web界面和API，用于管理和监控 RabbitMQ 的状态和性能。

关于 RabbitMQ 在 springboot 项目中的配置文件问题，使用 yml 格式，现在 springboot 支持 properties 和 yml 都支持，不过 yml 文件更适合需要管理大量复杂配置的项目和适用于需要清晰展示层级关系的配置场景，因此这里统一使用它作为配置文件。

工作原理

生产者将消息发送到交换机，指定一个路由键。
交换机根据路由键和绑定关系决定将消息转发到哪个队列。不同类型的交换机（如直连、主题、扇出）有不同的路由策略。
被转发的消息存储在一个或多个队列中，等待消费者处理。
消费者从队列中获取消息，进行处理。可以选择手动或自动确认消息。
消费者处理完消息后，发送确认给 RabbitMQ，以便删除该消息。如果处理失败，可以选择不确认，RabbitMQ 将重新投递消息。
通过这一流程，RabbitMQ 实现了异步通信和负载均衡，提高了系统的可靠性和扩展性。

RabbitMQ 中的 Direct Exchange 是一种消息路由机制，它根据路由键将消息精确地路由到一个或多个绑定的队列。它具有以下特点：

路由键: 每个发送到 Direct Exchange 的消息都带有一个路由键。交换机会根据这个路由键找到与之匹配的队列。
精确匹配: 只有那些绑定到交换机并且路由键与消息中的路由键完全匹配的队列才会接收到消息。
多个队列: 一个 Direct Exchange 可以绑定多个队列，这样相同的消息可以发送到多个队列，只要它们的路由键匹配。

使用场景

点对点通信: 适合需要将消息发送到特定队列的场景，比如任务处理系统。
简单的消息分发: 可以将消息发送到多个消费者，通过不同的路由键进行区分。

假设有一个 Direct Exchange，名称为 directExchange，你可以将多个队列（如 queueA 和 queueB）绑定到这个交换机，并使用不同的路由键（如 keyA 和 keyB）。

发送消息到 directExchange，如果消息的路由键是 keyA，那么只有绑定了 keyA 的队列（例如 queueA）会接收到这条消息。
如果发送的消息路由键是 keyB，那么只有绑定了 keyB 的队列（例如 queueB）会接收到。

创建一个配置类，用于设置 Direct Exchange 和队列：

@ConfigurationpublicclassRabbitConfig{publicstaticfinalString QUEUE_NAME ="testQueue";publicstaticfinalString EXCHANGE_NAME ="testExchange";publicstaticfinalString ROUTING_KEY ="testKey";@BeanpublicQueuequeue(){returnnewQueue(QUEUE_NAME,true);}@BeanpublicDirectExchangeexchange(){returnnewDirectExchange(EXCHANGE_NAME);}@BeanpublicBindingbinding(Queue queue,DirectExchange exchange){returnBindingBuilder.bind(queue).to(exchange).with(ROUTING_KEY);}}

创建一个简单的生产者发送消息到 RabbitMQ：

@ServicepublicclassMessageProducer{@AutowiredprivateAmqpTemplate amqpTemplate;publicvoidsendMessage(String message){
        amqpTemplate.convertAndSend(RabbitConfig.EXCHANGE_NAME,RabbitConfig.ROUTING_KEY, message);}}

创建消费者接收消息：

@ServicepublicclassMessageConsumer{@RabbitListener(queues =RabbitConfig.QUEUE_NAME)publicvoidreceiveMessage(String message){System.out.println("Received message: "+ message);}}

RabbitMQ 中的 Fanout Exchange 是一种将消息广播到所有绑定队列的交换机类型。具有以下特点：

广播功能: 发送到 Fanout Exchange 的消息会被传递到所有绑定的队列，无需考虑路由键。
无路由键过滤: 不管消息的路由键是什么，所有绑定到该交换机的队列都会接收到消息。
适合广播场景: 特别适用于需要将同一条消息发送给多个消费者的场景。

使用场景

实时通知: 比如聊天应用中的消息推送。
日志记录: 将日志消息发送到多个日志处理系统。
事件发布: 将事件发布给所有感兴趣的消费者。

创建配置类

@ConfigurationpublicclassRabbitConfig{publicstaticfinalString EXCHANGE_NAME ="broadcastExchange";publicstaticfinalString QUEUE_A ="queueA";publicstaticfinalString QUEUE_B ="queueB";@BeanpublicQueuequeueA(){returnnewQueue(QUEUE_A,true);}@BeanpublicQueuequeueB(){returnnewQueue(QUEUE_B,true);}@BeanpublicFanoutExchangeexchange(){returnnewFanoutExchange(EXCHANGE_NAME);}@BeanpublicBindingbindingA(Queue queueA,FanoutExchange exchange){returnBindingBuilder.bind(queueA).to(exchange);}@BeanpublicBindingbindingB(Queue queueB,FanoutExchange exchange){returnBindingBuilder.bind(queueB).to(exchange);}}

创建一个简单的生产者，发送消息到 Fanout Exchange：

@ServicepublicclassMessageProducer{@AutowiredprivateAmqpTemplate amqpTemplate;publicvoidsendMessage(String message){
        amqpTemplate.convertAndSend(RabbitConfig.EXCHANGE_NAME,"", message);}}

创建两个消费者，分别接收来自 Fanout Exchange 的消息：

@ServicepublicclassMessageConsumer{@RabbitListener(queues =RabbitConfig.QUEUE_A)publicvoidreceiveMessageFromQueueA(String message){System.out.println("Queue A received message: "+ message);}@RabbitListener(queues =RabbitConfig.QUEUE_B)publicvoidreceiveMessageFromQueueB(String message){System.out.println("Queue B received message: "+ message);}}

RabbitMQ 中的 Topic Exchange 是一种强大的消息路由机制，允许根据主题模式将消息路由到多个队列。它具有以下特点：

基于主题的路由: Topic Exchange 使用路由键的模式匹配来决定消息的去向。路由键可以包含多个词，使用点号（.）分隔。
通配符支持: Topic Exchange 支持两种通配符： - *（星号）: 匹配一个词。- #（井号）: 匹配零个或多个词。
灵活的路由: 允许更复杂的消息路由逻辑，适合多种场景。

使用场景

多层次分类: 可以根据主题对消息进行分类，例如日志消息可以按级别（如 error, info）和模块（如 auth, payment）路由。
事件驱动架构: 适合处理复杂的事件通知，支持不同的消费者对不同主题感兴趣。

假设有一个 Topic Exchange，名称为 topicExchange，并绑定了多个队列。可以使用以下路由键来路由消息：

order.created
order.cancelled
payment.completed

创建一个配置类，用于设置 Topic Exchange 和队列：

@ConfigurationpublicclassRabbitConfig{publicstaticfinalString EXCHANGE_NAME ="topicExchange";publicstaticfinalString QUEUE_NAME_A ="queueA";publicstaticfinalString QUEUE_NAME_B ="queueB";@BeanpublicQueuequeueA(){returnnewQueue(QUEUE_NAME_A,true);}@BeanpublicQueuequeueB(){returnnewQueue(QUEUE_NAME_B,true);}@BeanpublicTopicExchangeexchange(){returnnewTopicExchange(EXCHANGE_NAME);}@BeanpublicBindingbindingA(Queue queueA,TopicExchange exchange){returnBindingBuilder.bind(queueA).to(exchange).with("order.*");// 只接收 order.* 的消息}@BeanpublicBindingbindingB(Queue queueB,TopicExchange exchange){returnBindingBuilder.bind(queueB).to(exchange).with("payment.#");// 接收 payment 开头的所有消息}}

创建一个简单的生产者，发送消息到 Topic Exchange：

@ServicepublicclassMessageProducer{@AutowiredprivateAmqpTemplate amqpTemplate;publicvoidsendOrderMessage(String message){
        amqpTemplate.convertAndSend(RabbitConfig.EXCHANGE_NAME,"order.created", message);}publicvoidsendPaymentMessage(String message){
        amqpTemplate.convertAndSend(RabbitConfig.EXCHANGE_NAME,"payment.completed", message);}}

创建两个消费者，分别接收来自 Topic Exchange 的消息：

@ServicepublicclassMessageConsumer{@RabbitListener(queues =RabbitConfig.QUEUE_NAME_A)publicvoidreceiveOrderMessage(String message){System.out.println("Queue A received order message: "+ message);}@RabbitListener(queues =RabbitConfig.QUEUE_NAME_B)publicvoidreceivePaymentMessage(String message){System.out.println("Queue B received payment message: "+ message);}}

RabbitMQ 中的 Headers Exchange 是一种基于消息头部属性进行路由的交换机类型。它具有以下特点：

基于头部的路由: 消息被发送到 Headers Exchange 时，路由决策基于消息的头部属性，而不是路由键。
匹配条件: 可以定义多个头部属性的匹配条件，支持使用 x-match 参数： - all: 所有头部属性都必须匹配。- any: 只需任意一个头部属性匹配。
灵活性: 允许更复杂的路由逻辑，适用于多种场景。

使用场景

复杂路由需求: 当需要基于多个属性来决定消息路由时，使用 Headers Exchange 可以提供更高的灵活性。
多样化消息分类: 适合处理多维度的消息分类，例如根据多个消息属性进行路由。

创建一个配置类，用于设置 Headers Exchange 和队列：

@ConfigurationpublicclassRabbitConfig{publicstaticfinalString EXCHANGE_NAME ="headersExchange";publicstaticfinalString QUEUE_NAME_A ="queueA";publicstaticfinalString QUEUE_NAME_B ="queueB";@BeanpublicQueuequeueA(){returnnewQueue(QUEUE_NAME_A,true);}@BeanpublicQueuequeueB(){returnnewQueue(QUEUE_NAME_B,true);}@BeanpublicHeadersExchangeexchange(){returnnewHeadersExchange(EXCHANGE_NAME);}@BeanpublicBindingbindingA(Queue queueA,HeadersExchange exchange){Map<String,Object> headers =newHashMap<>();
        headers.put("type","order");returnBindingBuilder.bind(queueA).to(exchange).whereAll(headers).match();}@BeanpublicBindingbindingB(Queue queueB,HeadersExchange exchange){Map<String,Object> headers =newHashMap<>();
        headers.put("type","payment");returnBindingBuilder.bind(queueB).to(exchange).whereAll(headers).match();}}

创建一个简单的生产者，发送消息到 Headers Exchange：

@ServicepublicclassMessageProducer{@AutowiredprivateAmqpTemplate amqpTemplate;publicvoidsendOrderMessage(String message){Map<String,Object> headers =newHashMap<>();
        headers.put("type","order");
        amqpTemplate.convertAndSend(RabbitConfig.EXCHANGE_NAME,"", message, msg ->{
            msg.getMessageProperties().getHeaders().putAll(headers);return msg;});}publicvoidsendPaymentMessage(String message){Map<String,Object> headers =newHashMap<>();
        headers.put("type","payment");
        amqpTemplate.convertAndSend(RabbitConfig.EXCHANGE_NAME,"", message, msg ->{
            msg.getMessageProperties().getHeaders().putAll(headers);return msg;});}}

创建两个消费者，分别接收来自 Headers Exchange 的消息：

@ServicepublicclassMessageConsumer{@RabbitListener(queues =RabbitConfig.QUEUE_NAME_A)publicvoidreceiveOrderMessage(String message){System.out.println("Queue A received order message: "+ message);}@RabbitListener(queues =RabbitConfig.QUEUE_NAME_B)publicvoidreceivePaymentMessage(String message){System.out.println("Queue B received payment message: "+ message);}}

在 Spring Boot 项目中使用 RabbitMQ 的 Virtual Hosts (虚拟主机) 可以实现多个隔离的消息队列和配置。以下是使用 Virtual Hosts 的基本步骤：

首先，需要在 RabbitMQ 管理界面或通过命令行创建一个虚拟主机。例如，创建一个名为 my_vhost 的虚拟主机：

rabbitmqctl add_vhost my_vhost

为用户设置访问新创建的虚拟主机的权限。例如：

rabbitmqctl set_permissions -p my_vhost user_name ".*"".*"".*"

配置rabbitmq

spring:rabbitmq:host: localhost
    port:5672username: guest
    password: guest
    virtual-host: my_vhost  # 配置虚拟主机

在配置类中，正常配置 RabbitMQ 的交换机、队列和绑定，不需要做额外更改，因为配置类会自动使用指定的虚拟主机。

@ConfigurationpublicclassRabbitConfig{publicstaticfinalString EXCHANGE_NAME ="topicExchange";publicstaticfinalString QUEUE_NAME ="myQueue";@BeanpublicQueuequeue(){returnnewQueue(QUEUE_NAME,true);}@BeanpublicTopicExchangeexchange(){returnnewTopicExchange(EXCHANGE_NAME);}@BeanpublicBindingbinding(Queue queue,TopicExchange exchange){returnBindingBuilder.bind(queue).to(exchange).with("my.routing.key");}}

通过 RabbitTemplate 发送和接收消息，默认使用配置中的虚拟主机：

@ServicepublicclassMessageProducer{@AutowiredprivateAmqpTemplate amqpTemplate;publicvoidsendMessage(String message){
        amqpTemplate.convertAndSend(RabbitConfig.EXCHANGE_NAME,"my.routing.key", message);}}

通过以上步骤，可以在 Spring Boot 项目中有效使用 RabbitMQ 的 Virtual Hosts。虚拟主机可以帮助你在同一 RabbitMQ 服务器上隔离不同的应用程序或环境，提升管理灵活性和安全性。

2. 死信队列

RabbitMQ 中的死信队列（Dead Letter Queue, DLQ）用于处理无法被正常消费的消息。当消息因为某些原因无法被处理时（如超时、消费失败、队列达到最大长度等），这些消息可以被发送到一个指定的死信队列，以便后续的分析和处理。

死信条件：

消息被拒绝（basic.reject 或 basic.nack）并且没有重新排入队列。
超出最大重试次数：当消息在被消费的过程中多次重试仍然失败时，如果设置了最大重试次数，超过后将被送入死信队列。
消息 TTL（存活时间）到期。
队列达到最大长度。

配置死信队列：

创建一个专门的队列来接收死信。
在原始队列中设置 x-dead-letter-exchange 和 x-dead-letter-routing-key 属性，指定死信交换机和路由键。

以下是如何在 Spring Boot 中配置死信队列的示例：

@ConfigurationpublicclassRabbitConfig{publicstaticfinalString MAIN_QUEUE ="mainQueue";publicstaticfinalString DLQ_QUEUE ="deadLetterQueue";publicstaticfinalString DLX_EXCHANGE ="deadLetterExchange";publicstaticfinalString DLQ_ROUTING_KEY ="deadLetterRoutingKey";@BeanpublicQueuemainQueue(){// 设置死信交换机和路由键Map<String,Object> args =newHashMap<>();
        args.put("x-dead-letter-exchange", DLX_EXCHANGE);// 指定死信交换机
        args.put("x-dead-letter-routing-key", DLQ_ROUTING_KEY);// 指定死信路由键returnnewQueue(MAIN_QUEUE,true,false,false, args);}@BeanpublicQueuedeadLetterQueue(){returnnewQueue(DLQ_QUEUE,true);// 死信队列}@BeanpublicExchangedeadLetterExchange(){returnnewDirectExchange(DLX_EXCHANGE);// 死信交换机}@BeanpublicBindingbindingDLQ(Queue deadLetterQueue,Exchange deadLetterExchange){// 绑定死信队列到死信交换机，指定路由键returnBindingBuilder.bind(deadLetterQueue).to((DirectExchange) deadLetterExchange).with(DLQ_ROUTING_KEY);}}

在消费者中处理消息时，如果处理失败，可以将消息拒绝并发送到死信队列：

@ServicepublicclassMessageConsumer{@RabbitListener(queues =RabbitConfig.MAIN_QUEUE)publicvoidreceiveMessage(String message){try{// 处理消息System.out.println("Processing message: "+ message);// 如果处理失败，抛出异常thrownewRuntimeException("Processing failed!");}catch(Exception e){// 这里可以选择拒绝消息，消息会进入死信队列// 这里抛出异常后，消息自动进入死信队列thrownewAmqpRejectAndDontRequeueException("Rejecting message: "+ message);}}}

3. 消息堆积

在生产环境中，消息堆积（消息积压）是 RabbitMQ 运行过程中常见的问题，尤其当消费者的处理能力跟不上消息生产者时。消息堆积如果不及时处理，可能会导致 RabbitMQ 性能下降甚至宕机。以下是一些常用的解决方案来应对消息堆积问题：

增加消费者实例数量，使用更多的消费者同时处理消息。

水平扩展：增加消费者服务的实例数量（在微服务架构中通过 Kubernetes、Docker 实现服务集群的自动扩展）。

优化消费者的代码，提高消费速度。

优化业务逻辑：减少每个消息处理的时间。
异步处理：对于非实时要求较低的操作，考虑异步处理，比如消息入库、日志处理等操作可异步化。
批量处理：如果业务允许，可以启用批量处理，一次处理多条消息，减少每次调用时的开销。

使用 RabbitMQ 的集群模式，可以在多个节点之间分发消息，这样可以平衡负载并提高吞吐量。

在生产者侧，可以通过一些机制来控制消息的发送速度，避免消息过快堆积。

根据消费者处理能力，对生产者的发送速率进行限制。可以使用限流算法（如令牌桶、漏桶算法）来动态调整生产者的发送速率。
配置 RabbitMQ 的流控（流量控制）参数，通过 high-watermark（高水位线）控制生产者的发送速度。

rabbitmq:watermark:high:0.7# 当内存使用超过 70% 时触发流控low:0.5# 当内存使用低于 50% 时解除流控

实现生产者和消费者之间的反馈机制，消费者通知生产者自己处理的速度，通过后压机制降低消息的生产速度，避免过多的消息积压。

设置消息的过期时间（TTL），当消息超过指定时间未被消费时将被丢弃，减少堆积的消息量。

@BeanpublicQueuemyQueue(){Map<String,Object> args =newHashMap<>();
    args.put("x-message-ttl",60000);// 设置队列中消息的 TTL 为 60 秒returnnewQueue("myQueue",true,false,false, args);}

为不同的消息设置优先级，让关键消息被优先处理，避免高优先级的消息被阻塞。

@BeanpublicQueuepriorityQueue(){Map<String,Object> args =newHashMap<>();
    args.put("x-max-priority",10);// 设置优先级范围 0-10returnnewQueue("priorityQueue",true,false,false, args);}

当消息达到最大重试次数或超出存活时间（TTL）后，消息可以被路由到死信队列。死信队列中的消息可以被进一步分析或处理，避免队列中的消息无限积压。配置死信交换机，将无法处理的消息路由到死信队列中。

通过创建多个队列，将生产者的消息分发到多个队列中，消费者再从这些不同的队列消费，从而提高并发性，减少单一队列的堆积。可以使用插件 rabbitmq-sharding 来实现队列的分区。

通过监控 RabbitMQ 的消息队列深度（消息积压数量）和消费者性能，可以及时发现消息堆积问题。RabbitMQ 提供了很多指标，比如 message_ready 和 message_unacked，可以使用 Prometheus、Grafana 等工具进行监控。

设置队列的最大长度，超出长度后，最早的消息会被丢弃或被送往死信队列。

@BeanpublicQueuelimitedQueue(){Map<String,Object> args =newHashMap<>();
    args.put("x-max-length",10000);// 设置队列最大长度为 10000 条returnnewQueue("limitedQueue",true,false,false, args);}

如果业务允许，启用批量消费。批量获取消息一次性处理多条，可以减少频繁的网络传输和 I/O 开销，从而提升整体的消息处理吞吐量。

在 Spring Boot 项目中实现 RabbitMQ 的批量消费，可以通过配置 SimpleRabbitListenerContainerFactory 来设置批量消费的特性。以下是一个完整的示例，包括配置、消费者、以及消息发送部分。

在配置类中，设置 SimpleRabbitListenerContainerFactory 来启用批量消费。

@Configuration@EnableRabbitpublicclassRabbitConfigimplementsRabbitListenerConfigurer{@BeanpublicSimpleRabbitListenerContainerFactorybatchListenerContainerFactory(ConnectionFactory connectionFactory){SimpleRabbitListenerContainerFactory factory =newSimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        factory.setBatchListener(true);// 启用批量消费
        factory.setBatchSize(10);// 每次获取10条消息return factory;}}

创建一个消费者类来处理批量接收的消息。使用 @RabbitListener 注解来指定队列。

@ServicepublicclassMessageConsumer{@RabbitListener(queues ="myQueue", containerFactory ="batchListenerContainerFactory")publicvoidreceiveMessages(List<String> messages){for(String message : messages){System.out.println("Processing message: "+ message);// 处理每条消息的逻辑}}}

@ServicepublicclassMessageProducer{@AutowiredprivateAmqpTemplate amqpTemplate;publicvoidsendMessage(String message){
        amqpTemplate.convertAndSend("myQueue", message);System.out.println("Sent message: "+ message);}}

批量消费的大小（setBatchSize）可以根据实际需求进行调整。
在处理消息时，要确保你的处理逻辑不会阻塞，以便及时消费后续消息。
可以添加异常处理机制，确保在处理失败时适当地处理这些消息，比如重试或发送到死信队列。

在 Spring Boot 项目中实现 RabbitMQ 的并发消费，可以通过配置 SimpleRabbitListenerContainerFactory 来设置并发消费者的数量。下面是一个完整的示例，包括配置、消费者、以及消息发送部分。

在配置类中，设置 SimpleRabbitListenerContainerFactory 来启用并发消费。

@Configuration@EnableRabbitpublicclassRabbitConfigimplementsRabbitListenerConfigurer{@BeanpublicSimpleRabbitListenerContainerFactoryrabbitListenerContainerFactory(ConnectionFactory connectionFactory){SimpleRabbitListenerContainerFactory factory =newSimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        factory.setConcurrentConsumers(5);// 设置并发消费者数量
        factory.setMaxConcurrentConsumers(10);// 设置最大并发消费者数量
        factory.setPrefetchCount(1);// 每个消费者一次只获取1条消息return factory;}}

创建一个消费者类，使用 @RabbitListener 注解来指定队列。多个实例会并发处理来自同一队列的消息。

@ServicepublicclassMessageConsumer{@RabbitListener(queues ="myQueue")publicvoidreceiveMessage(String message){System.out.println("Processing message: "+ message);// 这里可以添加处理消息的逻辑try{// 模拟处理耗时操作Thread.sleep(1000);}catch(InterruptedException e){Thread.currentThread().interrupt();}}}

@ServicepublicclassMessageProducer{@AutowiredprivateAmqpTemplate amqpTemplate;publicvoidsendMessage(String message){
        amqpTemplate.convertAndSend("myQueue", message);System.out.println("Sent message: "+ message);}}

确保 setPrefetchCount 设为 1，以确保每个消费者在处理完当前消息之前不会接收新的消息，这样可以防止消息在消费者之间的竞争。
根据业务需求调整 setConcurrentConsumers 和 setMaxConcurrentConsumers 的值。
在处理消息时，要确保处理逻辑不会阻塞太长时间，以免影响其他消息的消费。
可以添加异常处理机制，以处理消费过程中可能出现的错误。
异步处理：可以有效防止消费者在处理消息时被阻塞。通过将消息的处理过程异步化，可以让消费者快速响应并继续接收其他消息。
重试机制：如果消费失败，可以使用 Spring 的 RetryTemplate 或者通过死信队列进行重试处理，避免消息一直堆积在队列中。

4. 消息有序性

在 RabbitMQ 中，保证消息的有序性是一个重要但相对复杂的问题，尤其是在分布式系统中。以下是一些常见的方法和策略，用于确保消息的有序性：

最简单的方法是将所有相关消息发送到同一个队列中。RabbitMQ 会按照消息的发送顺序处理这些消息，从队列中消费时能够保持有序性。使用单个队列的缺点是可能会导致性能瓶颈，消费者的并发能力受限。如果消息量很大，这可能会影响系统的吞吐量。

如果希望在队列中保证消息的严格顺序，确保只有一个消费者从该队列中消费消息。即使有多个消费者，仍然保持每个消费者仅处理来自一个队列的消息。这样可以确保处理顺序。

对于需要处理的消息，确保它们具有某种标识符（如用户 ID、订单 ID 等），可以将同一组的消息发送到同一队列中。这样就能保证相同标识符的消息能够保持有序。可以通过 routing key 或 header 中的属性来决定将消息发送到哪个队列。

在某些情况下，可以使用分布式锁来确保消息处理的有序性。这种方式可以在处理消息时锁定资源，确保同一时间只处理一条消息。使用分布式锁会增加复杂性，并可能导致性能下降。

在某些情况下，应用程序本身可能需要负责保持消息的顺序。这可以通过控制消息的接收和处理顺序来实现。在处理完消息 A 之前，不处理消息 B。可以在代码逻辑中实现这种顺序。

在应用程序中，可以实现一些监控机制，以便在发现消息顺序不正确时进行调整或重试。

监控示例：

记录处理状态: 在处理每条消息时，记录处理结果、时间戳和相关信息，以便后续分析。
回调机制: 可以定义一个回调接口，当消息处理成功或失败时，回调相应的方法。

publicinterfaceMessageProcessingCallback{voidonSuccess(String message);voidonFailure(String message,Exception e);}// 使用回调机制privatevoidprocessMessage(String message,MessageProcessingCallback callback){try{// 处理逻辑
        callback.onSuccess(message);}catch(Exception e){
        callback.onFailure(message, e);throw e;// 继续抛出异常以进行 nack}}

5. 消息可靠性

在RabbitMQ中保证消息的可靠性是确保消息不会丢失并能够被成功处理的重要部分。RabbitMQ提供了多种机制来增强消息的可靠性，包括消息持久化、队列持久化、消息确认、死信队列等。

在 RabbitMQ 中，当消息被成功消费后，消息会被删除。这一过程是通过消息确认机制来实现的。

持久化是指将消息存储到磁盘，以防止在 RabbitMQ 服务崩溃或重启时丢失消息。要启用持久化，可以采取以下措施：

创建一个配置类，用于声明队列、交换机和绑定它们。确保队列和交换机都被标记为持久化：

@Configuration@EnableRabbitpublicclassRabbitConfig{publicstaticfinalString QUEUE_NAME ="myQueue";publicstaticfinalString EXCHANGE_NAME ="myExchange";publicstaticfinalString ROUTING_KEY ="myRoutingKey";// 声明持久化队列@BeanpublicQueuemyQueue(){returnnewQueue(QUEUE_NAME,true);// 第二个参数为true表示持久化}// 声明持久化交换机@BeanpublicTopicExchangemyExchange(){returnnewTopicExchange(EXCHANGE_NAME,true,false);// 第一个参数是持久化，第二个参数是自动删除}// 绑定队列和交换机@BeanpublicBindingbinding(){returnBindingBuilder.bind(myQueue()).to(myExchange()).with(ROUTING_KEY);}}

创建一个服务类来发送持久化消息。确保在发送消息时，设置消息的持久化属性：

@ServicepublicclassMessageSender{@AutowiredprivateRabbitTemplate rabbitTemplate;publicvoidsendMessage(String message){// 创建持久化消息
        rabbitTemplate.convertAndSend(RabbitConfig.EXCHANGE_NAME,RabbitConfig.ROUTING_KEY, message, messagePostProcessor ->{MessageProperties properties = messagePostProcessor.getMessageProperties();
            properties.setDeliveryMode(MessageDeliveryMode.PERSISTENT);// 设置为持久化消息return messagePostProcessor;});System.out.println("Sent message: "+ message);}}

创建一个消息接收器类，使用 @RabbitListener 注解来处理消息：

@ComponentpublicclassMessageReceiver{@RabbitListener(queues =RabbitConfig.QUEUE_NAME)publicvoidreceiveMessage(String message){System.out.println("Received message: "+ message);// 处理消息逻辑}}

通过设置持久化队列和持久化交换机，确保消息在 RabbitMQ 服务重启后不会丢失。在发送消息时，设置了消息的持久化属性，使得消息即使在 RabbitMQ 崩溃后仍然能够被保留。

消息持久化数据存储在RabbitMQ的数据目录中的文件中。这个目录通常位于RabbitMQ配置文件中指定的/var/lib/rabbitmq/mnesia路径下。具体路径可以根据RabbitMQ的配置进行调整。

队列的持久化数据也存储在RabbitMQ的数据目录中的文件中。

RabbitMQ 提供了消息确认机制来确保消息被消费者成功处理后才从队列中删除。使用手动确认模式可以避免消息丢失。

spring:rabbitmq:host: localhost
    port:5672username: guest
    password: guest
    listener:simple:acknowledge-mode: manual  # 设置手动确认模式

@ServicepublicclassMessageSender{@AutowiredprivateRabbitTemplate rabbitTemplate;privatefinalString exchange ="myExchange";privatefinalString routingKey ="myRoutingKey";publicvoidsendMessage(String message){
        rabbitTemplate.convertAndSend(exchange, routingKey, message);System.out.println("Sent message: "+ message);}}

@Component@EnableRabbitpublicclassMessageReceiverimplementsChannelAwareMessageListener{@RabbitListener(queues ="myQueue")// 监听队列@OverridepublicvoidonMessage(Message message,Channel channel)throwsException{try{String msgBody =newString(message.getBody(),"UTF-8");System.out.println("Received message: "+ msgBody);// 处理消息逻辑// ...// 手动确认消息
            channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);System.out.println("Message acknowledged.");}catch(Exception e){// 处理失败，拒绝消息并重新入队
            channel.basicNack(message.getMessageProperties().getDeliveryTag(),false,true);System.err.println("Failed to process message: "+ e.getMessage());}}}

消费者在成功处理消息后发送确认（ACK），如果处理失败，可以选择否定确认（NACK）并重新入队。

当消息处理失败或被拒绝时，可以将其发送到死信队列。通过设置队列属性，可以将处理失败的消息转发到指定的死信队列。在队列属性中设置 x-dead-letter-exchange 和 x-dead-letter-routing-key 来指定死信交换机和路由键。

RabbitMQ 支持高可用性队列（HA Queues），可以在多个节点之间复制队列，确保即使某个节点失败，消息仍然可用。将 RabbitMQ 部署为集群，确保高可用性和负载均衡。

RabbitMQ 的镜像队列集群（Mirrored Queues Cluster）是实现高可用性的一种模式，它通过在集群中的多个节点之间复制队列，确保消息在集群的某些节点宕机时不会丢失，并且队列仍然可以正常工作。镜像队列集群广泛用于需要保证消息可靠性的业务场景，如金融、支付、订单处理等关键任务。

在镜像队列集群中，队列的主副本（主队列）会存在于一个节点上，称为主队列节点（Master）。同时，这个队列会被复制到集群中的其他节点，这些副本称为镜像（Mirrors）。

核心工作机制

主队列（Master Queue）： - 队列的原始副本，所有的消息发布、消费、ACK 都是在主队列上处理的。- 每当有消息发送到主队列时，主队列会将消息同步到其镜像节点。
镜像队列（Mirrored Queue）： - 镜像队列是主队列的完全副本，存在于集群的其他节点上，会实时同步主队列中的所有消息和状态。这些镜像队列与主队列保持同步，以确保所有消息和状态在每个镜像中都相同。- 如果主队列所在的节点发生故障，集群中的其中一个镜像队列会自动提升为新的主队列，并继续提供服务。
故障转移（Failover）： - 当主队列的节点发生故障时，RabbitMQ 会自动从剩下的镜像队列中选择一个提升为主队列。此过程通常是无缝的，在故障转移期间，消费和生产可能会有短暂的中断，但当新的主队列节点被选定后，消息处理会恢复正常。消费者和生产者可以继续与新的主队列通信。- 故障转移后，新的主队列会自动在其他节点上创建新的镜像，以保持高可用性。
镜像的自动管理： - RabbitMQ 可以根据策略自动管理镜像队列的数量和分布。可以配置镜像策略，控制镜像队列的创建、复制的节点数量等。

镜像队列集群的配置
镜像队列集群的配置通常通过策略（Policy）来实现。可以通过 RabbitMQ 管理界面或命令行工具 rabbitmqctl 来定义镜像队列的策略。
使用 rabbitmqctl 命令创建一个策略，将队列的镜像复制到集群的所有节点上。

rabbitmqctl set_policy ha-all "^ha\."'{"ha-mode":"all"}'

ha-all: 策略名称。
^ha.: 队列名称的正则表达式，所有以 ha. 开头的队列都会应用此策略。
{“ha-mode”:“all”}: 表示所有节点都会有该队列的镜像。
ha-mode： - all: 将队列镜像到集群中的所有节点上。- exactly: 将队列镜像到集群中的指定数量的节点上。- nodes: 选择特定的节点进行队列镜像。
ha-sync-mode: - automatic: 当新的镜像节点加入时，自动同步主队列中的消息到新镜像节点。- manual: 需要手动执行同步操作。

如果不希望将队列镜像到集群的所有节点，而是只希望将其复制到指定数量的节点，可以使用 ha-mode 的 exactly 选项。
假设有 3 个节点的 RabbitMQ 集群（节点 A、B、C），可以配置队列只在两个节点上进行镜像，如下：

rabbitmqctl set_policy ha-two "^ha\."'{"ha-mode":"exactly","ha-params":2,"ha-sync-mode":"automatic"}'

这会将所有以 ha. 开头的队列镜像到集群的任意两个节点上。
ha-mode: 使用 exactly 来指定镜像的数量。
ha-params: 表示该队列会被镜像到集群中两个节点上。
ha-sync-mode: 使用 automatic 模式，表示镜像队列会自动与主队列同步。

也可以选择将镜像队列只复制到特定节点上，通过 nodes 参数指定节点。

rabbitmqctl set_policy ha-nodes "^ha\."'{"ha-mode":"nodes","ha-params":["rabbit@node1","rabbit@node2"]}'

ha-mode: 使用 nodes 模式，将队列镜像到 rabbit@node1 和 rabbit@node2 上。

镜像队列会增加集群的网络带宽和存储开销，因为每一条消息都要在多个节点之间同步。应根据业务场景合理设置镜像的节点数量，平衡高可用性和性能。

优点：

高可用性：当某个节点故障时，镜像队列可以迅速接管，确保队列和消息的可用性。
无缝故障转移：RabbitMQ 会自动进行故障转移，无需额外的操作或维护。
消息可靠性：通过在多个节点上同步队列和消息，确保了消息不会因为节点故障而丢失。

缺点：

性能损耗：由于镜像队列需要同步消息和状态，增加了网络、磁盘和内存的负载，可能会影响消息处理的性能。
配置复杂性：在集群规模增大时，需要考虑如何合理配置镜像策略，避免不必要的资源消耗；维护镜像队列集群需要更复杂的集群管理和监控，尤其是在节点数量增加时。
分区问题：如果集群中存在网络分区，可能会导致队列数据不一致，镜像队列可能在不同分区中各自成为主队列，带来数据冲突问题。
扩展性受限：随着集群中节点数量的增加，同步的负载会成倍增加，因此不适合非常大规模的集群。

Spring Boot 项目中配置镜像队列
在 Spring Boot 项目中配置 RabbitMQ 镜像队列（Mirrored Queues）进行消息的收发，需要在 RabbitMQ 中设置队列为高可用队列，并在 Spring Boot 应用中进行相应的配置。

镜像队列的核心特性是，消息会被复制到集群中的多个节点（主节点 + 副节点），从而实现高可用性。其工作原理如下：

主节点（Master Node）负责处理所有入队和出队的操作。
副节点（Mirror Node）仅同步主节点的数据，并在主节点不可用时，自动切换成为新的主节点。

当主节点挂掉后，RabbitMQ 会通过仲裁机制，自动选择一个副节点作为新的主节点。此时，消息消费者和生产者会自动与新的主节点进行通信，而无需手动更改配置。

在 Spring Boot 中，可以通过配置 RabbitMQ 的镜像队列集群，使应用程序在主节点宕机后自动切换到副节点。为此，最常见的做法是配置 RabbitMQ 的多个节点地址或使用一个负载均衡的 DNS 入口来实现故障切换。

使用多个节点地址 (addresses)
Spring Boot 提供了 addresses 配置项，可以指定多个 RabbitMQ 节点地址。当其中一个节点不可用时，应用程序会自动尝试连接其他节点，而无需手动修改配置文件。这是最简单和直接的方式来实现高可用性。

spring:rabbitmq:addresses: host1:5672,host2:5672,host3:5672# 配置多个 RabbitMQ 集群节点port:5672username: guest
    password: guest
    virtual-host: /
    listener:simple:concurrency:3max-concurrency:10connection-timeout:15000queues:-name: mirroredQueue
        durable:trueexclusive:falseauto-delete:falsearguments:x-ha-policy: all

在这个配置中，addresses 通过逗号分隔的多个 host:port 配置，告诉 Spring Boot 的 RabbitMQ 客户端尝试连接多个节点。如果 host1 宕机，客户端会自动尝试连接 host2，然后是 host3。当其中某个节点不可用时，Spring Boot 会自动切换到下一个可用节点，避免手动修改配置文件。

如果 RabbitMQ 集群通过 DNS 提供了一个负载均衡的入口地址，可以使用该 DNS 入口来实现节点的自动切换。负载均衡 DNS 可以自动将请求路由到可用的 RabbitMQ 节点，并在节点故障时自动进行切换。

spring:rabbitmq:host: rabbitmq-cluster.example.com  # 指定 RabbitMQ 集群的负载均衡 DNS 名称port:5672# RabbitMQ 默认端口username: guest
    password: guest
    virtual-host: /
    listener:simple:concurrency:5max-concurrency:10connection-timeout:15000

在这个配置中，host 被设置为 RabbitMQ 集群的 DNS 名称（例如 rabbitmq-cluster.example.com）。这个 DNS 名称应该指向 RabbitMQ 集群中的所有节点，并能够处理主节点宕机时的自动切换。

创建一个 RabbitMQ 配置类，配置高可用队列（镜像队列）：

当镜像队列被正确配置（如 x-ha-policy: all），消息会自动复制到副节点。
当主节点故障时，副节点会被提升为新的主节点，继续处理消息的发送和消费。
Spring Boot 中的 RabbitMQ 连接使用 spring-rabbit 和 AMQP 协议库，它内置了自动重连机制，当连接的主节点宕机时，Spring Boot 应用程序会尝试重新连接到集群中的其他节点，重新建立连接。

@Configuration@EnableRabbitpublicclassRabbitConfig{@BeanpublicQueuemirroredQueue(){// 创建一个高可用队列（镜像队列）returnnewQueue("mirroredQueue",true,false,false,Map.of("x-ha-policy","all"));}}

创建一个消息发送者，用于发送消息到镜像队列：

@ServicepublicclassMessageProducer{@AutowiredprivateRabbitTemplate rabbitTemplate;publicvoidsendMessage(String message){
        rabbitTemplate.convertAndSend("mirroredQueue", message);System.out.println("Sent message: "+ message);}}

创建一个消息接收者，用于接收来自镜像队列的消息：

@ComponentpublicclassMessageListener{@RabbitListener(queues ="mirroredQueue")publicvoidreceiveMessage(String message){System.out.println("Received message: "+ message);}}

监控：使用 RabbitMQ 管理插件或其他监控工具，监控队列的状态、消费者的处理情况和系统的性能指标。
告警：设置告警机制，当消息堆积或消费者处理速度缓慢时，及时通知相关人员。
特性RabbitMQRocketMQKafka协议AMQP自定义协议（类似于 JMS）自定义协议消息模型Broker 模式，多种交换机模型Topic-Tag 模型Pub/Sub，Topic-Partition 模型吞吐量中等，适合中小型应用高，适合高并发事务极高，适合大数据场景延迟低，适合低延迟场景低，适合实时场景可能较高，适合大吞吐场景可靠性高，持久化和事务支持高，支持事务消息高，依赖分区复制应用场景企业级应用、复杂路由、事务消息分布式事务、高并发、延时消息大数据处理、日志收集、流式处理

标签： rabbitmq 分布式

本文转载自: https://blog.csdn.net/qq_41893505/article/details/142623776
版权归原作者 信徒_ 所有，如有侵权，请联系我们删除。

1. 结构

2. 死信队列

3. 消息堆积

4. 消息有序性

5. 消息可靠性

发表评论

“Rabbitmq 面试题”的评论:

关于作者

overfit同步小助手

相关阅读

文章导航