Amazon SQS FIFO队列多线程消费者：保证MessageGroupId内消息有序

1. 多线程消费者是否会打破同一MessageGroupId内的FIFO处理顺序？ #

会。SQS FIFO队列仅保证投递顺序，即同一MessageGroupId的消息会按顺序被投递到消费者，但默认多线程消费模式下，这些消息会被分配到不同线程并行处理。由于线程执行速度受异步逻辑、资源占用等因素影响，先投递的消息（如message1）可能比后投递的消息（如message2）晚完成处理，直接导致业务层面的顺序错乱。

尤其是你提到的Web Flux异步+自动确认场景：线程获取message1后进入异步块，SQS自动确认消息后会立即投递message2到其他线程，此时message1的异步逻辑可能还在运行，message2已经开始处理，最终处理完成顺序完全不可控。

2. 多线程架构下保证同一分组消息顺序处理的最佳方案 #

核心原则是：同一MessageGroupId的消息必须串行处理，不同MessageGroupId的消息可以并行处理，既保留多线程的性能优势，又保障分组内的顺序性。推荐两种方案：

方案1：利用Spring Cloud AWS SQS的分组级并发控制 #

Spring Cloud AWS SQS支持针对MessageGroupId的并发限制，通过配置让同一分组的消息只能被一个线程处理，不同分组可以并行。

• 配置方式：在@SqsListener中指定maxConcurrentMessagesPerMessageGroupId=1，或者全局配置：

spring.cloud.aws.sqs.listener.max-concurrent-messages-per-message-group-id=1

该配置会自动将同一分组的消息路由到同一个消费线程，确保串行处理，不同分组则分配到不同线程并行。

方案2：自定义分组级串行处理队列 #

如果需要更灵活的控制，可以自己维护每个MessageGroupId的串行处理队列，用ConcurrentHashMap存储每个分组的任务队列，同一分组的任务排队执行，不同分组并行：

@Component
public class SqsMessageHandler {
    private final ConcurrentHashMap<String, Queue<Runnable>> groupTaskQueues = new ConcurrentHashMap<>();
    private final ExecutorService executor = Executors.newFixedThreadPool(5);

    @SqsListener("your-fifo-queue")
    public void handleMessage(String message, @Header("MessageGroupId") String groupId) {
        // 获取或创建当前分组的任务队列
        Queue<Runnable> taskQueue = groupTaskQueues.computeIfAbsent(groupId, k -> new LinkedList<>());
        // 将当前消息处理逻辑加入队列
        taskQueue.add(() -> processMessage(message, groupId));
        // 如果队列只有当前任务，立即提交执行；否则等待前序任务完成
        if (taskQueue.size() == 1) {
            executor.submit(() -> processQueue(taskQueue, groupId));
        }
    }

    private void processQueue(Queue<Runnable> taskQueue, String groupId) {
        while (!taskQueue.isEmpty()) {
            Runnable task = taskQueue.poll();
            try {
                task.run();
            } catch (Exception e) {
                // 处理异常，比如重试或死信队列
                e.printStackTrace();
            }
        }
        // 任务处理完后移除分组队列，避免内存泄漏
        groupTaskQueues.remove(groupId);
    }

    private void processMessage(String message, String groupId) {
        // 等待Web Flux异步逻辑完成
        Mono.just(message)
                .flatMap(this::asyncBusinessLogic)
                .block();
    }
}

3. 维持消息严格顺序的最佳实践与技术 #

（1）禁用自动确认，改用手动确认 #

自动确认会导致SQS在消息被消费者接收后立即标记为已处理，无法控制后续投递。手动确认可以确保只有当前消息处理完成（包括异步逻辑）后，才通知SQS投递下一条：

@SqsListener("your-fifo-queue")
public void handleMessage(String message, Acknowledgment acknowledgment, @Header("MessageGroupId") String groupId) {
    try {
        // 执行异步处理逻辑并等待完成
        asyncBusinessLogic(message).block();
        // 处理完成后手动确认
        acknowledgment.acknowledge();
    } catch (Exception e) {
        // 处理失败，不确认，让SQS重新投递
        e.printStackTrace();
    }
}

同时要配置SQS的可见性超时，确保在消息处理完成前，不会被重新投递到其他消费者。

（2）分组级别的同步锁 #

使用Striped锁（Guava提供）或ConcurrentHashMap维护每个分组的锁，确保同一分组的消息只能被一个线程处理：

private final Striped<Lock> groupLocks = Striped.lock(10); // 预创建10个锁，可按需调整

@SqsListener("your-fifo-queue")
public void handleMessage(String message, @Header("MessageGroupId") String groupId) {
    Lock lock = groupLocks.get(groupId);
    lock.lock();
    try {
        // 处理消息（包括异步逻辑，需同步等待完成）
        asyncBusinessLogic(message).block();
    } finally {
        lock.unlock();
    }
}

（3）配置合理的可见性超时 #

SQS FIFO队列的可见性超时要设置为大于单条消息的最大处理时间，避免消息在处理过程中被重新投递，导致重复处理或顺序错乱。

（4）异步逻辑同步化处理 #

如果使用Web Flux等异步框架，必须确保异步逻辑完全完成后再结束消费任务。可以用block()（适合非高并发场景），或者用CountDownLatch等待异步结果：

private void processAsyncMessage(String message) throws InterruptedException {
    CountDownLatch latch = new CountDownLatch(1);
    asyncBusinessLogic(message)
            .doOnTerminate(latch::countDown)
            .subscribe();
    latch.await(); // 等待异步逻辑完成
}

（5）避免同一分组的消息被批量拉取 #

如果开启了批量拉取，确保同一批次内的消息来自不同MessageGroupId，或者限制同一分组的批量拉取数量为1，避免同一分组的多条消息被同时拉取到消费者，导致并行处理。

内容的提问来源于stack exchange，提问作者Ohyu