RabbitMQ无法稳定获取队列全部消息的问题咨询

Hey there! I’ve worked through that exact Spring RabbitMQ tutorial before, so I know exactly where this inconsistent message count issue comes from. Let’s break down the fixes to make sure you always get all 4 messages:

1. Use a CountDownLatch to wait for all messages to process #

The most common culprit here is that your app exits before the receiver can finish processing all messages. The fix is to add synchronization with a CountDownLatch to keep the app running until every message is handled.

First, update your Receiver.java to include the latch (initialize it to 4, since you’re sending 4 messages):

import java.util.concurrent.CountDownLatch;
import org.springframework.stereotype.Component;

@Component
public class Receiver {

    private final CountDownLatch latch = new CountDownLatch(4);

    public void receiveMessage(String message) {
        System.out.println("Received <" + message + ">");
        latch.countDown(); // Decrement the latch each time a message is processed
    }

    public CountDownLatch getLatch() {
        return latch;
    }
}

Then modify your Runner.java (the class that sends messages) to wait for the latch to reach zero after sending all messages:

import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Component;

@Component
public class Runner implements CommandLineRunner {

    private final RabbitTemplate rabbitTemplate;
    private final Receiver receiver;

    public Runner(RabbitTemplate rabbitTemplate, Receiver receiver) {
        this.rabbitTemplate = rabbitTemplate;
        this.receiver = receiver;
    }

    @Override
    public void run(String... args) throws Exception {
        System.out.println("Sending messages...");
        for (int i = 0; i < 4; i++) {
            rabbitTemplate.convertAndSend("hello", "Hello from RabbitMQ! " + i);
        }
        // Wait here until all 4 messages are processed
        receiver.getLatch().await();
    }
}

This will keep your application running until the receiver has processed every single message, so you’ll never miss any again.

2. Adjust listener container concurrency (optional) #

If you’re still seeing inconsistencies (though the latch fix should solve it), you can force single-threaded consumption to ensure messages are processed in order and no race conditions occur. Add this config class to your project:

import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class RabbitConfig {

    @Bean
    public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(ConnectionFactory connectionFactory) {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        // Set to single consumer to avoid concurrent processing quirks
        factory.setConcurrentConsumers(1);
        factory.setMaxConcurrentConsumers(1);
        return factory;
    }
}

This is useful if you need strict message ordering, but the latch solution alone should handle your original problem.

3. Double-check message acknowledgment (quick sanity check) #

By default, Spring AMQP uses auto-acknowledgment, which means messages are marked as processed as soon as they’re delivered to the receiver. If you ever switch to manual acknowledgment, make sure you call channel.basicAck() after processing each message to avoid messages being requeued or lost. This isn’t the issue here, but it’s a good thing to keep in mind for future work.

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