Spring Boot RabbitMQ消费者仅首次消费问题及多监听器配置求助

Let’s break down how to fix your "only first message is consumed" issue, then walk through setting up multiple listeners for your automation queue system.

Fixing the One-Time Consumption Problem #

This issue almost always ties back to acknowledgment handling, unhandled exceptions, or misconfigured listener containers. Here are the most common fixes:

1. Ensure Message Acknowledgment is Working Correctly #

RabbitMQ won’t send new messages to a consumer if the previous message wasn’t acknowledged.

• By default, Spring AMQP uses AUTO acknowledgment mode (it auto-acks after your listener method completes successfully). If you’ve switched to MANUAL mode, you must explicitly acknowledge each message:

@RabbitListener(queues = "your-task-queue")
public void handleMessage(Message message, Channel channel) throws IOException {
    try {
        // Process your task here
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
    } catch (Exception e) {
        // Reject and requeue if processing fails
        channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
    }
}

• Verify your container factory isn’t overriding the default behavior:

@Bean
public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(ConnectionFactory connectionFactory) {
    SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
    factory.setConnectionFactory(connectionFactory);
    factory.setAcknowledgeMode(AcknowledgeMode.AUTO); // Keep this unless you need manual acks
    return factory;
}

2. Handle Unhandled Exceptions #

If your listener throws an uncaught exception, the container might stop consuming messages entirely. Add retry logic or an error handler to prevent this:

@Bean
public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(ConnectionFactory connectionFactory) {
    SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
    factory.setConnectionFactory(connectionFactory);
    
    // Add retry policy for transient errors
    factory.setRetryTemplate(retryTemplate());
    
    // Set error handler to avoid stopping the container
    factory.setErrorHandler(new ConditionalRejectingErrorHandler(new FatalExceptionStrategy() {
        @Override
        public boolean isFatal(Throwable t) {
            return t instanceof AmqpRejectAndDontRequeueException;
        }
    }));
    return factory;
}

private RetryTemplate retryTemplate() {
    RetryTemplate retryTemplate = new RetryTemplate();
    retryTemplate.setBackOffPolicy(new ExponentialBackOffPolicy());
    retryTemplate.setRetryPolicy(new SimpleRetryPolicy(3)); // Retry 3 times
    return retryTemplate;
}

3. Confirm Your Listener is a Singleton #

Spring components are singletons by default, but double-check your listener class is annotated with @Component or @Service—prototype-scoped listeners can cause unexpected behavior.

Configuring Multiple Listeners #

Setting up multiple listeners is flexible—you can target different queues, use concurrent consumers for the same queue, or even use separate container factories for different workloads.

1. Multiple Listeners for Different Queues #

Just add multiple @RabbitListener methods (in the same component or separate ones):

@Component
public class TaskQueueListeners {

    @RabbitListener(queues = "mapping-task-queue")
    public void handleMappingTask(String taskDetails) {
        System.out.println("Processing mapping task: " + taskDetails);
    }

    @RabbitListener(queues = "validation-task-queue")
    public void handleValidationTask(String taskDetails) {
        System.out.println("Processing validation task: " + taskDetails);
    }
}

2. Concurrent Consumers for a Single Queue #

To process messages in parallel for a high-volume queue, configure concurrent consumers in your container factory:

@Bean
public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(ConnectionFactory connectionFactory) {
    SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
    factory.setConnectionFactory(connectionFactory);
    factory.setConcurrentConsumers(4); // Start with 4 consumers
    factory.setMaxConcurrentConsumers(6); // Scale up to 6 if needed
    return factory;
}

Your single listener will now have 4 concurrent instances processing messages from the queue.

3. Separate Container Factories for Different Listener Types #

If you need different configurations (like manual acks for critical tasks, auto-acks for non-critical), define multiple factories:

@Bean(name = "criticalTaskFactory")
public SimpleRabbitListenerContainerFactory criticalTaskFactory(ConnectionFactory connectionFactory) {
    SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
    factory.setConnectionFactory(connectionFactory);
    factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
    factory.setConcurrentConsumers(3);
    return factory;
}

@Bean(name = "nonCriticalTaskFactory")
public SimpleRabbitListenerContainerFactory nonCriticalTaskFactory(ConnectionFactory connectionFactory) {
    SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
    factory.setConnectionFactory(connectionFactory);
    factory.setAcknowledgeMode(AcknowledgeMode.AUTO);
    factory.setConcurrentConsumers(1);
    return factory;
}

Then reference the factory in your listeners:

@RabbitListener(queues = "critical-task-queue", containerFactory = "criticalTaskFactory")
public void handleCriticalTask(Message message, Channel channel) throws IOException {
    // Process critical task
    channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
}

@RabbitListener(queues = "non-critical-task-queue", containerFactory = "nonCriticalTaskFactory")
public void handleNonCriticalTask(String taskDetails) {
    // Process non-critical task
}

Quick Check for Your Application Class #

Ensure your Application.java includes @EnableRabbit (it’s often auto-included with Spring Boot’s RabbitMQ starter, but explicit is safer):

@SpringBootApplication
@ComponentScan(basePackages = { "com.fractal.sago", "com.fractal.grpc" })
@EnableRabbit // Explicitly enable RabbitMQ listener support
public class Application extends SpringBootServletInitializer {
    public static void main(String[] args) {
        SpringApplication.run(Application.class, args);
    }
}

If the problem persists, sharing your full listener code and application.properties (RabbitMQ configs like spring.rabbitmq.host) would help narrow it down further.

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