如何让多个RabbitMQ队列按顺序执行？Spring Boot AMQP场景

RabbitMQ（AMQP的主流实现）本身没有内置的“等待某队列完全处理完再启动另一队列消费”的机制，你之前尝试的优先级机制仅作用于单队列内部的消息优先级排序，无法控制不同队列之间的执行顺序，所以没生效是正常的。

以下是几种可落地的实现方案：

方案1：通过消息监听容器的启停控制 #

这是最直接的实现方式，核心思路是让队列B的监听容器在队列A处理完成前保持停止状态，待A处理完所有消息后再启动B的消费。

• 配置队列B的监听容器时，设置autoStartup = false，让它初始不启动。
• 在队列A的消息处理器中维护计数器，当处理消息数达到预设值时，触发启动队列B的容器。

代码示例：

// 队列B的监听容器配置
@Bean
public DirectMessageListenerContainer queueBListenerContainer(ConnectionFactory connectionFactory) {
    DirectMessageListenerContainer container = new DirectMessageListenerContainer(connectionFactory);
    container.setQueueNames("queueB");
    container.setMessageListener(new MessageListenerAdapter(new QueueBConsumer()));
    container.setAutoStartup(false); // 初始不启动
    return container;
}

// 队列A的消费者
@Component
public class QueueAConsumer {
    private int processedCount = 0;
    private final DirectMessageListenerContainer queueBListenerContainer;

    public QueueAConsumer(DirectMessageListenerContainer queueBListenerContainer) {
        this.queueBListenerContainer = queueBListenerContainer;
    }

    @RabbitListener(queues = "queueA")
    public void processMessage(String message) {
        // 处理队列A的消息逻辑
        System.out.println("处理队列A消息：" + message);
        
        processedCount++;
        // 处理完20条后启动队列B
        if (processedCount == 20) {
            queueBListenerContainer.start();
        }
    }
}

方案2：延迟发送队列B的消息 #

如果业务允许，可以不提前将消息发送到队列B，而是等队列A处理完成后，再批量发送消息到B，自然保证执行顺序。

• 先将队列B的消息暂存到数据库或内存缓存中。
• 队列A处理完所有消息后，批量将缓存消息发送到队列B。

代码示例：

@Component
public class QueueAConsumer {
    private int processedCount = 0;
    private final RabbitTemplate rabbitTemplate;
    private final List<String> queueBMessages = new ArrayList<>();

    public QueueAConsumer(RabbitTemplate rabbitTemplate) {
        this.rabbitTemplate = rabbitTemplate;
        // 模拟缓存队列B的20条消息
        for (int i = 1; i <= 20; i++) {
            queueBMessages.add("队列B消息" + i);
        }
    }

    @RabbitListener(queues = "queueA")
    public void processMessage(String message) {
        // 处理队列A的消息逻辑
        System.out.println("处理队列A消息：" + message);
        
        processedCount++;
        if (processedCount == 20) {
            // 批量发送消息到队列B
            queueBMessages.forEach(msg -> rabbitTemplate.convertAndSend("queueB", msg));
        }
    }
}

方案3：基于事件驱动的触发机制 #

利用Spring的事件发布/订阅机制，在队列A处理完成后发布事件，由监听器触发队列B的消费启动。

• 定义自定义事件类，标记队列A处理完成的状态。
• 队列A处理完所有消息时发布该事件。
• 事件监听器收到通知后，启动队列B的监听容器。

代码示例：

// 自定义事件类
public class QueueAProcessCompletedEvent extends ApplicationEvent {
    public QueueAProcessCompletedEvent(Object source) {
        super(source);
    }
}

// 队列A的消费者
@Component
public class QueueAConsumer {
    private int processedCount = 0;
    private final ApplicationEventPublisher eventPublisher;

    public QueueAConsumer(ApplicationEventPublisher eventPublisher) {
        this.eventPublisher = eventPublisher;
    }

    @RabbitListener(queues = "queueA")
    public void processMessage(String message) {
        // 处理队列A的消息逻辑
        System.out.println("处理队列A消息：" + message);
        
        processedCount++;
        if (processedCount == 20) {
            eventPublisher.publishEvent(new QueueAProcessCompletedEvent(this));
        }
    }
}

// 事件监听器
@Component
public class QueueBStartupListener {
    private final DirectMessageListenerContainer queueBListenerContainer;

    public QueueBStartupListener(DirectMessageListenerContainer queueBListenerContainer) {
        this.queueBListenerContainer = queueBListenerContainer;
    }

    @EventListener
    public void handleQueueACompleted(QueueAProcessCompletedEvent event) {
        queueBListenerContainer.start();
    }
}

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