Spring RabbitMQ: one @RabbitListener multiple queues with the priority of the specified queue

Viewed 122

I have 2 queues - PRIORITY_QUEUE and SIMPLE_QUEUE. The SIMPLE_QUEUE is large, the number of messages in it reaches several thousand, the PRIORITY_QUEUE can be empty and only sometimes receive a few messages. I need to ensure that messages from the SIMPLE_QUEUE are not read while there are messages in the PRIORITY_QUEUE, and if there are messages in the priority queue, reading the SIMPLE_QUEUE is blocked/paused. When the PRIORITY_QUEUE has received a message, we must pause reading from the SIMPLE_QUEUE.

prefetch property: we read messages by one

spring.rabbitmq.listener.simple.prefetch=1

listener example

    @RabbitListener(queues = {priorityQueue , simpleQueue})
    public void processMyQueue(String message) {
        // if priorityQueue is not empty we shouldn't consume messages from simpleQueue
    }

rabbit configuration

@Slf4j
@Configuration
@PropertySource({"...,..."})
@RequiredArgsConstructor
public class RabbitConfiguration {

    @Value("${spring.rabbitmq.host}")
    private String queueHost;
    @Value("${spring.rabbitmq.username}")
    private String username;
    @Value("${spring.rabbitmq.password}")
    private String password;

    @Bean
    public ConnectionFactory connectionFactory() {
        CachingConnectionFactory cachingConnectionFactory = new 
                            CachingConnectionFactory(queueHost);
        cachingConnectionFactory.setUsername(username);
        cachingConnectionFactory.setPassword(password);
        return cachingConnectionFactory;
    }

    @Bean
    public AmqpAdmin amqpAdmin() {
        return new RabbitAdmin(connectionFactory());
    }

    @Bean
    public RabbitTemplate rabbitTemplate() {
        return new RabbitTemplate(connectionFactory());
    }

    @Bean(value = "simpleQueue")
    public Queue simpleQueue() {
        return new Queue("simpleQueue");
    }

    @Bean(name = "priorityQueue")
    public Queue priorityQueue() {
        return new Queue("priorityQueue");
    }
}
1 Answers

You can't do it with a single listener; use two listeners and stop the simple listener container when the priority listener receives any message; then start the simple queue listener when the priority listener goes idle.

You can use a ListenerContainerIdleEvent to detect an idle listener.

https://docs.spring.io/spring-amqp/docs/current/reference/html/#consumer-events

Use the RabbitListenerEndpointRegistry to stop/start containers.

Related