RabbitMQ应用问题 - 消息顺序性保证、消息积压问题

文章目录

MQ 消息顺序性保证
- 概述
- 原因分析
- 解决方案
- 基于 spring-cloud-stream 实现`分区消费`
消息挤压问题
- 概述
- 原因分析
- 解决方案

MQ 消息顺序性保证

概述

a）消息顺序性：消费者消费的消息的顺序和生产者发送消息的顺序是一致的.

例如生产者发送消息顺序是 msg1、msg2、msg3，那么消费者也需要按照 msg1、msg2、msg3 的顺序进行消费.

b）顺序不一致可能会导致哪些问题？

例如用户系统中，用户需要对昵称进行了两次修改，此时生产者发送两条消息：

消息1：修改用户318 的昵称为 “白天”.
消息2：修改用户318 的昵称为 “黑夜”.
那么，按正常的逻辑来讲，用户318 的名称最后因该为 “黑夜”，但如果消息1 是最后一个被消费者消费的消息，那么用户318 的名称就变成了 “白天”.

原因分析

Note：以下场景成立的前提是，只能有一个生产者！因为生产者发送消息给 mq，中间都需要经过网络传输，而网络的不确定性是非常大的，因此无法保证多个生产者的消息谁先到 mq.

a）多个消费者：
多个消息会被不同的消费者并行处理，也就意味着有的消费者消费的快，有的消费者消费的慢，从而导致消息处理的顺序性无法保证.

b）网络波动：
网络波动可能会导致消费者消费完消息返回的 ack 丢失，从而使得 mq 以为消息发给消费者的中途丢失了，进而使得消息重新入队，这就意味着 如果队列中此时还有其他消息，那么这个重新入队的消息就会排在队列尾部，而头部的消息会被优先消费，导致顺序性问题.

实际上，也就意味着，只要触发了消息重新入队的操作，就会导致顺序性问题.

c）消息路由问题：
在复杂的路由场景中（例如大量应用 Topic 交换机），消息可能会根据 routingKey 被分发到不同的队列，使得无法保证全局的顺序性.

d）死信队列：
消息因为一些原因（例如被消费者返回 nack + requeue=false），然后放入死信队列，那么死信队列无论是网络传输，还是处理死信队列的消费者和普通队列的消费者并行处理，都会导致顺序不一致的情况.

解决方案

顺序性的保证分为 局部顺序性保证 和 全局顺序性保证.
例如如下，假设消息入队的顺序为 msg1、msg2、msg3、msg4、msg5…
在这里插入图片描述

消息顺序性保证的常见策略：

Note：以下顺序性保证策略往往不是单独使用进行保证的，而是多种组合使用.

a）单队列，单消费者（全局顺序性）
最简单的方式就是使用单个队列，并由单个消费者进行消息. 对于消息在队列先进先出，这是 RabbitMQ 给我们保证的.

b）业务逻辑控制（全局顺序性）
例如给每个消息引入一个序号（类似 TCP 确认应答），序号 3 消费之前，要保证序号 2 被成功消费…

c）手动消息确认机制（局部顺序性）
消费者在处理完消息后，显式的发送确认，这样 RabbitMQ 才会移除并继续处理下一个消息.

Ps：在 RabbitMQ 中，当消费者接收到一条消息时，这条消息并不会立即从队列中删除。相反，消息会保持在队列中，直到 RabbitMQ 收到消费者发回的确认.

d）分区消费（局部顺序性）
单个消费者的吞吐量太低了，当需要多个消费者来提高处理速度时，可以使用分区消费. 也就是把一个队列分割成多个分区（例如根据订单系统，将订单id 进行 hash 或者其他算法 -> 保证同样的订单 id，经过这个算法后，得到的队列名称是一致的（如果同样的订单 id 一会跑到队列1，一会跑到队列2，就会导致多个消费并行消费，最终消费顺序不一致）），最后每个分区由一个消费者处理，保证每个分区内消息的顺序性.

Ps：RabbitMQ 本身没有实现分区消费

基于 spring-cloud-stream 实现`分区消费`

Note: https://docs.spring.io/spring-cloud-stream/reference/rabbit/rabbit_partitions.html

RabbitMQ 并没有实现分区消费，因此这里可以引入一些其他的机制来实现.

a）引入依赖

    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>3.3.2</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>

    <dependencyManagement>
        <dependencies>
            <dependency>
                <groupId>org.springframework.cloud</groupId>
                <artifactId>spring-cloud-dependencies</artifactId>
                <version>2023.0.2</version>
                <type>pom</type>
                <scope>import</scope>
            </dependency>
        </dependencies>
    </dependencyManagement>

    <dependencies>
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-stream-binder-rabbit</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
        </dependency>
    </dependencies>

b）配置文件如下：

spring:
  rabbitmq:
    host: env-base
    port: 5672
    username: root
    password: 1111
  cloud:
    stream:
      bindings: # bindings 表示消息通道绑定配置
        generate-out-0: # generate-out-0 是一个输出通信的名称，表示这是生成消息的第一个通道(还可能由类似 generate-out-1 的其他通道)
          destination: partitioned.destination # 消息发送的名称为 "partitioned.destination" 的目的地(目的地在这里就是 mq 消息队列).
          producer: # 生产者配置
            # partitioned: true
            partition-key-expression: headers['partitionKey'] # 表示消息应该发送到哪个分区(这个跟代码里配置的 header 有关)
            partition-count: 2 # 表示有两个分区(两个队列). 生产者会根据 "partition-key-expression" 计算的结果，将消息分配到这两个分区之一
            required-groups: # 配置消费组
              - myGroup

c）代码如下：

import org.springframework.boot.WebApplicationType;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.builder.SpringApplicationBuilder;
import org.springframework.context.annotation.Bean;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;

import java.util.Random;
import java.util.function.Supplier;

@SpringBootApplication
public class SpringCloudStreamMqApplication {

    private static final Random RANDOM = new Random(System.currentTimeMillis());

    private static final String[] data = new String[] {
            "abc1", "abc2", "abc3",
            "abc4",
    };

    public static void main(String[] args) {
        new SpringApplicationBuilder(SpringCloudStreamMqApplication.class)
                .web(WebApplicationType.NONE) //不运行其他 web 组件
                .run(args);
    }

    /**
     * 分区消息:
     * 方法返回一个函数，这个函数每次调用都会从 data 中随机选择一个字符串，
     * 生成一个带有分区键(partitionKey)的消息，并将这个消息返回.
     */
    @Bean
    public Supplier<Message<?>> generate() {
        return () -> {
            String value = data[RANDOM.nextInt(data.length)];
            System.out.println("Sending: " + value);
            return MessageBuilder.withPayload(value)
                    .setHeader("partitionKey", value)
                    .build();
        };
    }

}

d）效果演示：
在 mq 管理平台可以看到多出来了一个交换机和两个队列（分区）
在这里插入图片描述

在 partitioned.destination.myGroup-0 中获取消息，可以看到都是 “abc2” 和 “abc4”

在 partitioned.destination.myGroup-1 中获取消息，可以看到都是 “abc1” 和 “abc3”