spring boot rabbitmq 如何保持顺序消费

RabbitMQ 是一个消息代理和队列功能的开源实现，可以帮助构建分布式应用程序。Spring Boot 集成 RabbitMQ 可以方便地在应用程序中使用消息队列，保持顺序消费可以通过以下方式来实现：

1. 单线程消费：使用一个线程消费消息，因为 RabbitMQ 的队列是有序的，所以保证单线程的消费能够保证消息的顺序。需要注意的是，单线程消费可能影响整体的性能。

2. 有序分片消费：将消息队列按照一定的规则进行分割，每个分片使用一个线程消费，这样可以减少单线程消费的性能影响。保证消息有序性的关键是要确保分片规则是有序的。

3. 使用 RabbitMQ 提供的优先级队列：优先级队列会按照消息的优先级进行排序，可以通过设置优先级来保证消息的顺序。缺点是需要将队列中的所有消息都进行排序，因此可能会影响整体性能。

4. 使用 RabbitMQ 提供的插件：RabbitMQ 提供了插件来实现有序消费，比如 rabbitmq_delayed_message_exchange 插件可以延迟消息投递，保证消息的有序性。此外，还有 RabbitMQ Stream 插件等。

如果实现有序分片消费？

要实现有序分片消费，可以先将消息队列按照一定的规则（如消息 ID、时间戳等）分成多个分片，然后每个分片使用一个单独的消费者线程消费消息。要保证消息的顺序，需要在分片规则上做额外的处理，确保分片规则是有序的，然后让每个消费者只消费自己所负责分片的消息。

以下是实现有序分片消费的代码示例：

首先定义一个分片规则，例如按照消息 ID 的 hash 值分片：

int numShards = 10; // 分成 10 个分片
public int getShardIndex(String messageId) {
    int hash = Math.abs(messageId.hashCode());
    return hash % numShards;
}

然后创建多个消费者线程，每个线程只负责消费自己所负责的分片：

@RabbitListener(queues = "myQueue")
public void processMessage(Message message) {
    String messageId = extractMessageId(message);
    int shardIndex = getShardIndex(messageId);
    if (shardIndex == myShardIndex) {
        // 处理消息逻辑
    }
}

可以使用 Spring Boot 提供的 @RabbitListener 注解来监听消息队列。在消费消息时，先从消息中提取出消息 ID，然后根据分片规则计算出当前消费者线程负责的分片编号，如果当前线程负责的分片与消息所在分片相同，则处理该消息。这样每个消费者线程只会消费自己负责的分片，就能保证消息的有序性。

下面是一个完整的示例，包括消费者类、消息发送者类和一个测试用例：

消息消费者类：

import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
import java.util.concurrent.atomic.AtomicInteger;
@Component
public class MyConsumer {
    private int myShardIndex;
    private int numShards = 10;
    private AtomicInteger counter = new AtomicInteger(0);
    public MyConsumer() {
        // 假设从配置文件中读取 myShardIndex
        myShardIndex = 3;
    }
    @RabbitListener(queues = "myQueue")
    public void processMessage(Message message) {
        String messageId = extractMessageId(message);
        int shardIndex = getShardIndex(messageId);
        if (shardIndex == myShardIndex) {
            int count = counter.getAndIncrement();
            System.out.println("Consumer " + myShardIndex + " received message " + message.getBody() + " (" + count + ")");
        }
    }
    private int getShardIndex(String messageId) {
        int hash = Math.abs(messageId.hashCode());
        return hash % numShards;
    }
    private String extractMessageId(Message message) {
        // 假设 message 的 messageId 在 messageProperties 的 headers 中
        return message.getMessageProperties().getHeaders().get("messageId").toString();
    }
}

消息发送者类：

import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import java.util.UUID;
@Component
public class MySender {
    @Autowired
    private AmqpTemplate rabbitTemplate;
    public void sendMessage() {
        String messageId = UUID.randomUUID().toString();
        String message = "Hello, RabbitMQ";
        rabbitTemplate.convertAndSend("myExchange", "myRoutingKey", message, msg -> {
            msg.getMessageProperties().getHeaders().put("messageId", messageId);
            return msg;
        });
    }
}

测试用例：

import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit.jupiter.SpringExtension;
import static org.junit.jupiter.api.Assertions.*;
@ExtendWith(SpringExtension.class)
@SpringBootTest
public class MyConsumerTest {
    @Autowired
    private MySender sender;
    @Autowired
    private RabbitTemplate rabbitTemplate;
    @Test
    public void testSharding() throws InterruptedException {
        // 发送消息
        for (int i = 0; i < 100; i++) {
            sender.sendMessage();
        }
        // 等待消息被消费完毕
        Thread.sleep(5000);
        // 检查是否有所有 shard 都有消息被消费到
        for (int i = 0; i < 10; i++) {
            int count = (int) rabbitTemplate.receiveAndConvert("myQueue", 10000);
            assertTrue(count > 0, "Shard " + i + " has not received any message");
        }
        // 清空队列中的消息
        while (rabbitTemplate.receiveAndConvert("myQueue") != null) {}
    }
}

这个示例中，MyConsumer 类处理来自 "myQueue" 队列的消息，并根据消息的 messageId 对消息进行分片。如果消息对应的 shard 索引和当前实例的 shard 索引相同，则处理消息。否则忽略该消息。

MySender 类负责发送消息到 "myExchange" 交换器，交换器将消息路由到 "myRoutingKey" 绑定的队列中。这里通过设置消息的 messageId，来模拟产生不同的 shard 索引。

MyConsumerTest 测试用例会发送 100 条消息到队列中，并等待 5 秒钟，然后检查所有的 shard 是否都收到了消息。如果有 shard 没有收到消息，则测试失败。