RabbitMQ

文章目录

RabbitMQ 的作用

• 提供了系统之间的异步调用，比如一个支付功能，用户在支付完成之后，会去数据库中执行后续操作，然后更新支付状态，会生成订单信息，如果后续还需要添加功能，就需要去业务逻辑中修改代码，这样就会出现业务耦合。同时想要执行后续操作，需要等待支付功能完成，在此等待过程中会耗费时间，CPU空转，性能比较差。当业务中有操作失败，就会将全部操作回滚。如果下一个操作依赖上一个操作时就需要用到同步操作。但是后续的很多业务操作只需要知道支付成功之后就去执行，不需要等待其他业务执行完成之后再去执行。同步操作的时效性强，但是拓展性差，并且性能下降还会出现级联失败等问题。
• 异步调用的方式就是基于消息通知的方式，其中有三个角色：消息发送者、消费代理、消息接收者。微信消息发送、送外卖。支付服务就不在同步调用业务关联度低的服务，而是发送消息通知Broker，这样做具有以下优势 - 解除耦合，拓展性强。- 无需等待，性能好。- 故障隔离：当某一个业务接收服务宕机，其他的服务可以正常执行，这个服务重连之后只需要去MQ中去获取数据就行。- 缓存消息，削峰填谷作用：当突然有大量的支付请求过来后，不会第一时间去冲击数据库，而是存放在MQ中，根据业务处理的速度自己去取，业务服务压力就很小。
• 异步调用的问题： - 不能立刻得到调用结果，时效性差。- 不确定下游业务是否执行成功。- 业务安全依赖于Broker的可靠性。

为什么使用RabbitMQ

MQ就是MessageQueue，存放消息的队列，也就是异步调用中的Broker。

在日常开发过程中，常见的消息队列有四种，RabbitMQ、ActiveMQ、RocketMQ、Kafka。 这四中的对比性下图可以看到，其中RabbitMQ是Rabbit公司专门研究的，相较于其他消息中间件它支持SMTP协议，并且它的消息延迟更是达到了恐怖的微秒级。当然它的消息可靠性以及可用性也是非常高的，所以一般项目开发没有特殊要求都是使用的是RabbitMQ。

数据隔离

交换机和队列都有自己的VirtualHost，不同的VirtualHost都有自己不同的交换机和队列。一个MQ中可以有多个VirtualHost，在发消息的时候去连接对应的VirtualHost就行。每个user可以去操作自己创建的的VirtualHost，查看的话时根据管理员创建user时分配的权限决定。

SpringAMQP

<!--RabbitMQ--><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId></dependency>
        
spring:     
  rabbitmq:
    host:  # 主机名
    port:5672 # 端口
    virtual-host:/ # 虚拟主机
    username:  # 用户名
    password:  # 密码

加入RabbitMQ依赖，在yml文件中配置，然后通过RabbitTemple向队列中发送消息。

work模式

默认情况下，RabbitMQ会将消息依次轮询投递给绑定在队列上的每一个消费者。并没有考虑到消费者是否已经处理完消息，这种情况可能出现的问题就是当我们不知道消费者消费能力的时候容易出现消息堆积。比如此时有两个消费者，消费者a一秒钟可以处理50条数据，消费者b一秒钟只能处理5条数据，此时有1000条消息发送到队列中，一次轮询绑定消费者，每一个消费者绑定了500个数据，但是消费者a10秒钟就处理完成，此时消费者b还在处理消息，a此时就空闲着，可用性比较低。

因此我们需要在yml中设置prefetch值为1，确保同一时刻最多投递给消费者1条消息，处理完之后才能获取下一条消息。

Rabbitmq:
    listener:
      simple:
        prefetch:1

交换机

交换机主要分为三种类型：Fanout（广播）、Direct（定向）、Topic（话题）。

• Fanout：Fanout Exchange会将接收到的消息广播到每一个跟其绑定的queue中，所以也叫广播模式。
• Direct：Direct Exchange会将接收到的消息根据规则路由到指定的Queue，因此称为定向路由。 - 每一个Queue都与Exchange设置一个BindingKey。- 发布者发布消息时，指定消息的RoutingKey。- Exchange将消息路由到BindingKey与消息BindingKey一致的队列。
• Topic：TopicExchange与DirectExchange类似，特殊之处在于 - routingKey可以是多个单词的列表，并且以 . 分割。- Queue与Exchange指定BinddingKey时可以使用通配符： - #：代指0个或多个单词。- *：代指一个单词。

如何声明队列和交换机

1. Spring AMQP提供了几个类，用来声明队列、交换机以及其绑定关系。

• Queue：用于声明队列，可以用工厂类QueueBuilder构建。
• Exchange：用于声明交换机，可以用工厂类ExchangeBuilder构建。
• Binding：用于声明队列和交换机的绑定关系，可以用工厂类BindingBuilder构建。

@BeanpublicFanoutExchangefanoutExchange(){//        ExchangeBuilder.fanoutExchange("").build();returnnewFanoutExchange("shuqg.fanout2");}@BeanpublicQueuefanoutQueue3(){//        QueueBuilder.durable("").build();returnnewQueue("shuqg.queue3");}@BeanpublicBindingfanoutBinging3(Queue fanoutqueue3,FanoutExchange fanoutExchange){// 如果需要绑定bindingkey在后面.with("")returnBindingBuilder.bind(fanoutqueue3).to(fanoutExchange);}

2. 基于注解声明

@RabbitListener(bindings =@QueueBinding(
            value =@Queue(name ="direct.queue2", durable ="true"),
            exchange =@Exchange(name ="shuqg.direct", type =ExchangeTypes.DIRECT),
            key ={"red","yellow"}))publicvoidlistenDirectQueue2(String msg)throwsInterruptedException{System.out.println("消费者2 收到了 direct.queue2的消息：【"+msg+"】");Thread.sleep(200);}

消息转换器

• Spring对消息处理默认实现的是SimpleMessageConverter，基于JDK的ObjectOutputStream完成序列化。
• 但是存在以下问题，JDK的序列化有安全风险、JDK序列化的消息太大、JDK序列化的消息可读性差。- 在传输put类型消息的时候，RabbitMQ默认会将消息序列化转换为字节码，但是可读性非常差，原本非常短的一个消息变得非常大，并且有乱码风险。
• 建议采用JSON序列化代替默认序列化，在SpringAMQP中有JSON的接口，只不过没有生效，我们只需要引入JSON依赖。

        <!--JSON-->
        <dependency>
            <groupId>com.fasterxml.jackson.core</groupId>
            <artifactId>jackson-databind</artifactId>
        </dependency>

然后在publisher和consumer中都要配置MessageConverter

@BeanpublicMessageConverterjacksonMessageConvertor(){returnnewJackson2JsonMessageConverter();}

生产者重连

有时候由于网络波动，可能会出现客户端连接MQ失败的情况。我们可以通过配置开启失败后的重连机制。 当网络不稳定时，使用重试机制可以有效提高消息发送成功概率。不过SpringAMQP消息重试机制是阻塞式的重试，也就是多次重试等待过程中，线程是被阻塞的，会影响业务性能。如果对业务性能有要求，建议禁用重试机制，如果要使用就合理配置等待时长和重试次数，也可以考虑使用异步线程来执行发送消息的代码。

  rabbitmq:
    template:
      retry:
        enabled: true # 开启超时重试机制
        initial-interval: 1000ms # 失败后的初始等待时间
        multiplier: 1 # 失败后下次的等待时长倍数，下次等待时长 = initial-interval * multiplier
        max-attempts: 3 # 最大重试次数

生产者确认

RabbitMQ有Publisher Confirm和Publisher Return两种确认机制。开启确认机制后，在MQ成功发送消息后返回确认消息给生产者。

• 消息投递到了交换机，但是路由失败。此时会通过PublisherReturn返回路由异常的原因，然后返回ACK，告知投递成功，此时消息成功发送到了交换机中，但是路由失败的原因可能是交换机没有关联队列或者交换机没有BindingKey与队列相匹配。
• 如果临时消息（未开启持久化non durable）投递到了MQ，并且入队成功，返回ack，表示投递成功。
• 如果持久消息（开启了持久化durable）投递到了MQ，并且入队完成持久化，返回ack，表示投递成功。
• 其他情况都会返回nack，出现nack可能的情况有 - 如果消息投递到交换机失败，会通过Publisher Confirm返回nack，表示消息投递失败，这种情况一般很少发生，如果发生就要不是代码写的有问题，要不就是交换机的配置有问题。- 消息投递到队列时队列已满。

配置生产者确认机制

  rabbitmq:
    publisher-confirm-type: correlated # 开启publisher-confirm机制，并设置confirm类型
    # 这里有三种参数，默认none关闭，其次simple是同步阻塞等待MQ回执消息，然后是correlated是MQ异步回调方式返回回执消息。
    publisher-returns:true # 开启publisher return机制

每一个RabbitTemplate只能配置一个ReturnCallback，因此需要在项目启动过程中配置

@Slf4j@ConfigurationpublicclassCommonConfigimplementsApplicationContextAware{@OverridepublicvoidsetApplicationContext(ApplicationContext applicationContext)throwsBeansException{// 获取RabbitTemplateRabbitTemplate rabbitTemplate = applicationContext.getBean(RabbitTemplate.class);// 设置ReturnCallback
        rabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey)->{
            log.info("消息发送失败，应答码{}，原因{}，交换机{}，路由键{}，消息{}",
                    replyCode, replyText,exchange, routingKey, message.toString());});}}

生产者发送消息

    @Test
    void testConfirmCallback() throws InterruptedException {
        // 创建cd
        CorrelationData cd = new CorrelationData(UUID.randomUUID().toString());
        // 添加ConfirmCallback
        cd.getFuture().addCallback(new ListenableFutureCallback<CorrelationData.Confirm>() {
            @Override
            public void onFailure(Throwable ex) {
                log.error("消息回调失败", ex);
            }

            @Override
            public void onSuccess(CorrelationData.Confirm result) {
                log.debug("收到confirm callback回执");
                if (result.isAck()) {
                    // 消息发送成功
                    log.debug("消息发送成功，返回ack");
                }else{
                    // 消息发送失败
                    log.error("消息发送失败，返回nack，原因{}", result.getReason());
                }
            }
        });
        rabbitTemplate.convertAndSend("shuqg.direct", "red","hello", cd);
        Thread.sleep(2000);
    }

生产者确认需要额外的网络和系统资源开销，尽量不要使用。如果一定要使用，无需开启Publisher-Return机制，因为一般路由失败是自己业务的问题。对于nack消息可以设置有限的重试次数，依然失败则记录异常消息到日志中。

MQ持久化

RabbitMQ如何保证消息可靠性

• 首先通过配置可以让交换机、队列、以及发送的消息都持久化。这样队列中的消息会持久化到磁盘，MQ重启消息依然存在。
• RabbitMQ在3.6版本引入了LazyQueue，并且在3.12版本后会称为队列的默认模式。LazyQueue会将所有消息都持久化
• 开启持久化和生产者确认时，RabbitMQ只有在消息持久化完成后才会给生产者返回ACK回执

在默认情况下，RabbitMQ会将接收到的消息保存到内存中以降低消息收发的延迟。这样会导致两个问题：

• 一旦MQ宕机，内存中的消息会丢失。
• 内存空间有限，当消费者故障或处理过慢，会导致消息积压，引发MQ阻塞。
• 可以通过数据持久化（mq3.6以前）和Lazy Queue（mq3.6以后）去解决这两个问题。

数据持久化

• 交换机持久化：创建交换机的时候如果勾选Transient就是临时的，但是我们平时都是勾选Durable是要保证交换机持久化的，mq重启之后交换机不会消失。
• 队列持久化：创建队列的时候也是同理，默认的是Druable持久化的，不然mq重启之后队列就会丢失。在Spring中创建交换机和队列的时候默认就是持久化的。

LazyQueue

从mq3.6之后开始增加了LazyQueue的概念，也就是惰性队列。惰性队列有以下特点：

• 接收到消息之后直接存入磁盘而非内存（内存中只保留最近的消息，默认2048条）
• 消费者要消费消息时才会从磁盘中读取并加载到内存。
• 支持数百万条的数据存储。
• 在3.12版本之后，所有的队列都是LazyQueue模式，无法更改。

消费者的可靠性

1. 消费者确认机制

为了确认消费者是否成功处理消息，RabbitMQ提供了消费者确认机制（Consumer Acknowledgement），当消费者处理消息结束后，应该向RabbitMQ发送一个回执，告知RabbitMQ自己消息处理状态。

• ack：成功处理消息，RabbitMQ从队列中删除该消息。
• nack：消息处理失败，RabbitMQ重新发送消息。
• reject：消息处理失败并拒绝该消息，RabbitMQ从队列中删除该消息。

开启消费者确认机制为auto，由Spring确认消息处理成功后返回ack。开启消费者确认机制，RabbitMQ支持消费者确认机制，当消费者处理消息之后可以向MQ发送ack回执，MQ收到ack回执之后才会去删除该消息。 SpringAMQP中允许配置三种确认模式：

  rabbitmq:
    listener:
      simple:
        prefetch:1
        acknowledge-mode: none # none, manual手动， auto自动

• none：默认情况，不处理，即消息投递给消费者后立刻ack，消息会立刻从MQ删除。非常不安全，不建议使用。
• manual：手动模式，需要在业务代码结束后，调用api发送ack或reject，存在业务入侵，但是更灵活。
• auto（一般选择这种）：自动模式，由Spring监听listener代码是否出现异常，当业务正常执行时则自动返回ack.当业务出现异常时，根据异常判断返回不同结果: - 如果是业务异常，会自动返回nack。- 如果是消息处理或校验异常，自动返回reject。
• 当消费者异常返回时，我们可以开启消费者失败重试机制，利用Spring的retry机制，在消费者出现异常时利用本地重试，设置重试次数，多次重试失败后将消息投递到异常交换机，交由人工处理。

2. 消费失败问题

当消费者出现异常后，会不断requeue（重新入队到队列），再重新发送给消费者，然后再次异常，再次requeue，无限循环，导致mq消息处理飙升，带来不必要的压力。 我们可以利用Spring的retry机制，在消费者出现异常时利用本地重试，而不是无限制的requeue到mq队列。

• 消息失败后处理策略：在开始重试模式后，默认情况下报错三次，也就是重试三次就会放弃，此时需要使用MessageRecoverer接口来处理，包含三种实现 - RejectAndDontRequeueRecoverer：重试耗尽后，直接丢弃消息。默认的就是这种方式。- ImmediateRequeueMessageRecoverer：重试耗尽后，返回nack，消息重新入队。- RepublishMessageRecoverer：重试耗尽后，将失败消息投递到指定交换机。 - 首先将失败处理策略改为第三种RepublishMessageRecoverer。- 然后定义接收失败消息的交换机、队列及其绑定关系。- 然后定义RepublishMessageRecoverer。

3. 业务幂等性

幂等性是一个数学概念，就是f(x) = f(f(x))，在程序开发中，指的是同一个业务，执行一次或多次对业务状态的影响是一致的。重复消费问题。

• 查询删除这些业务天生就是幂等的，新增修改这些业务就不是幂等的。
• 给每一个消息都设置一个唯一id，利用id判断是否重复消费- 每一个消息都生成一个唯一id，与消息一起投递给消费者。- 消费者接收到消息后处理自己的业务，业务处理成功后将消息id保存到数据库中。- 如果下次又收到相同的消息，去数据库查询判断是否存在，存在则为重复消息，放弃处理。- 使用自带的Jackson2JsonMessageConverter，可以实现自动生成唯一id，当将CreateMessageIds设置为true，底层会自动创建唯一id，并返回。- @Bean public MessageConverter jacksonMessageConverter(){ // 定义消息转换器 Jackson2JsonMessageConverter jjmc = new Jackson2JsonMessageConverter(); // 配置自动创建消息id，用于识别不同消息，也可以在业务中基于id判断是否重复消息 jjmc.setCreateMessageIds(true); return jjmc; }
• 业务判断- 结合业务逻辑，基于业务本身做判断。以支付修改订单业务为例，我们要在支付后修改订单状态为已支付，应该在修改订单状态前先查询订单状态，判断状态是否未支付。只有未支付订单才需要修改，其他状态不做处理。

如何保证支付服务与交易服务之间的订单状态一致性

使用MQ完成订单状态同步->为了保证mq可靠，使用了生产者确认，消费者确认，生产者重试，同时开启mq持久化，最后做了幂等性判断。

如何保证消息不丢失

• 可能导致消息丢失的场景：生产者发送消息没有到达交换机或者没有到达队列，MQ宕机，消费者服务宕机。
• 开启生产者确认机制，确保生产者的消息能到达队列。RabbitMQ中提供了一个确认机制用来避免消息发送到MQ过程中丢失，消息发送到MQ之后，会返回一个结果给发送者，表示消息是否处理成功。- 如果消息发送到交换机失败，交换机会返回一个nack，如果是发送到MQ失败会返回一个ack。- 消息失败之后，回调方法重新发送消息，如果还是失败，可以记录到日志中通过查看日志进行补充，或者将失败的消息记录到数据库中，做一个定时发送任务，发送成功之后删除数据库中的数据。
• 开启消息持久化功能，确保消息未消费前在队列中不会丢失。MQ默认是在内存中存储消息，开启持久化功能可以将数据存储在磁盘上，即使MQ宕机或重启也不会丢失数据。- 持久化交换机- 持久化队列- 持久化消息
• 开启消费者确认机制为auto，由Spring确认消息处理成功后返回ack。
• 开启消费者确认机制，RabbitMQ支持消费者确认机制，当消费者处理消息之后可以向MQ发送ack回执，MQ收到ack回执之后才会去删除该消息。SpringAMQP中允许配置三种确认模式： rabbitmq: listener: simple: prefetch:1 acknowledge-mode: none # none, manual手动， auto自动- none：默认情况，不处理，即消息投递给消费者后立刻ack，消息会立刻从MQ删除。非常不安全，不建议使用。- manual：手动模式，需要在业务代码结束后，调用api发送ack或reject，存在业务入侵，但是更灵活。- auto（一般选择这种）：自动模式，由Spring监听listener代码是否出现异常，当业务正常执行时则自动返回ack.当业务出现异常时，根据异常判断返回不同结果: - 如果是业务异常，会自动返回nack。- 如果是消息处理或校验异常，自动返回reject。- 当消费者异常返回时，我们可以开启消费者失败重试机制，利用Spring的retry机制，在消费者出现异常时利用本地重试，设置重试次数，多次重试失败后将消息投递到异常交换机，交由人工处理。

消息重复消费问题

• 重复消费发生的地方：在消费者消费队列中的消息的时候会向队列中返回ack，此时如果因为网络问题或者队列宕机，没有收到消费者的ack，重连之后会重试机制导致重复消费问题。
• 解决方法：每条消息设置一个唯一的标识id，当消费者接收到消息时去校验这个业务id是否存在，根据这个id去表中查询，如果id不存在则正常去接收消息，如果id已经存在了就证明这个消息已经消费过了，就不需要去消费了，这样就解决了重复消费的问题。
• 幂等方案：分布式锁，数据库锁（悲观锁、乐观锁），但是加锁的化性能会大大降低，如果数据库中有唯一标识id，则优先采用第一种方案。

RabbitMQ中死信交换机？延迟队列了解哪些？

• 一般使用在下单的时候，当下单之后当下单之后会有一个过期的时间，当在指定时间内未支付，就会将这个订单销毁。如果使用定时任务，设置key value在redis中设置过期时间，我们需要定时去查询数据库中用户支付状态，如果到达过期时间还没有支付，就会删除订单表，这个时候，如果设置时间间隔较短，对数据库的压力会非常巨大，但是如果设置间隔时间较长，就会导致时效性较差。
• 延迟队列就是进入队列的消息会被延迟消费的队列，我们当时的某一个业务使用到了延迟队列（超时订单、限时优惠、定时发布。。）
• 其中延迟队列就用到了死信交换机和TTL实现的。
• 当队列中的消息满足下面情况之一，就可以成为死信- 消息消费失败，返回nack，并且请求参数为false。- 消息超时未消费（设置TTL）。设置TTL一般有两种方式（哪个存活时间短以哪个为准） - 消息所在队列设置了存活时间。- 消息本身设置了存活时间。- 要传递的队列消息堆积满了，最早的消息可能成为死信。
• 一般死信消息是会被直接丢弃的，但是我们可以给该队列配置一个dead-letter-exchange属性，指定一个交换机，队列中的死信就会投递到该交换机中，这个交换机就是死信交换机。这个交换机也可以绑定一个队列，死信消息可以直接从交换机投递到该队列中，其他消费者可以去消费该队列中的消息。
• RabbitMQ中有一个延迟队列插件实现延迟队列DelayExchange- 声明一个交换机，添加delayed属性为true，这个就是一个可以实现延迟队列的交换机。- 发送消息时，通过消息头x-delay，设置消息存活时间。

消息堆积问题怎么解决

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

• 增加更多消费者，提高消费速度。
• 在消费者内开启线程池加快消息处理速度。消费者只负责去接收消息，所有的处理消息，处理业务逻辑都交给线程池去处理，但是线程池的作用是最大程度的利用CPU的资源，需要根据硬件配置去设置线程池。
• 扩大队列容积，提高堆积上限，采用惰性队列，在声明队列的时候可以设置属性x-queue-mode为lazy，即为惰性队列。使用惰性队列的好处是： - 接收到消息后直接存入磁盘而非内存，消息的上限比较高。- 消费者要消费消息时才会从磁盘中读取并加载到内存。- 支持数百万条消息存储。- 性能比较稳定，但基于磁盘存储，受限于磁盘IO，时效性会降低。

RabbitMQ高可用机制

• 普通集群，又叫标准集群，这个集群中每一个节点都有同一个交换机的信息，每个节点都有不同的队列，但是其他节点会有队列的引用信息。 - 会在集群的各个节点间共享部分数据，包含交换机、队列元信息。但是不包含队列中的消息。- 当访问集群中某节点时，如果队列不在该节点，会从数据所在节点传递到当前节点并返回。- 队列所在节点宕机，队列中的消息就会丢失。
• 镜像集群，本质是主从模式 - 交换机、队列、队列中的消息会在各个mq镜像节点之间同步备份。- 创建队列的节点是该队列的主节点，备份到其他节点的该队列是该队列的镜像节点。- 镜像队列结构是一主多从（从就是镜像），所有操作都是主节点完成，然后同步给镜像节点。- 主节点宕机后，镜像节点会替代成为新的主节点（如果在主从同步前主节点就已经宕机，可能会出现数据丢失）
• 如果担心出现数据丢失我们可以采用仲裁队列替代镜像队列，与镜像队列一样，都是主从模式，支持主从数据同步，主从协议基于Raft协议，是强一致性的，并且使用起来也非常简单，不需要额外的配置，在声明队列的时候只需要指定这个是仲裁队列即可。