Tag: amqp

RabbitConfiguration.java

@Bean
public SimpleMessageListenerContainer messageListenerContainer() {
    // configure listener to receive messages from queues
    SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
    container.setConnectionFactory(connectionFactory());
    container.setConcurrentConsumers(8);
    container.setMaxConcurrentConsumers(32);
    // listen to queues
    container.setQueueNames("MonitoringV5_Queue1", "MonitoringV5_Queue2");
    container.setMessageListener(new RabbitMessageListener());
    return container;
}

public class RabbitMessageListener implements MessageListener {
    @Override
    public void onMessage(Message message) {
        switch (message.getMessageProperties().getConsumerQueue()) {
            case "MonitoringV5_Queue1" -> Queue1(message);
            case "MonitoringV5_Queue2" -> Queue2(message);
        }
    }

    private void Queue1(Message message) {
        String str=new String(message.getBody());
        System.out.println(str);
        Gson gson = new Gson();
        ReadValueDto value= gson.fromJson(str,ReadValueDto.class);
    }

    private void Queue2(Message message) {
        String str=new String(message.getBody());
        System.out.println(str);
        Gson gson = new Gson();
        ReadValueDto value= gson.fromJson(str,ReadValueDto.class);
    }
}

If your callback logic depends on the AMQP Channel instance for any reason, you may instead use the ChannelAwareMessageListener. It looks similar but has an extra parameter.

References
https://docs.spring.io/spring-amqp/docs/current/reference/html/#message-listener

@Bean
public RabbitTemplate rabbitTemplate() {
    RabbitTemplate template = new RabbitTemplate(connectionFactory());
    // configure template to use retry policy
    RetryTemplate retryTemplate = new RetryTemplate();
    ExponentialBackOffPolicy backOffPolicy = new ExponentialBackOffPolicy();
    backOffPolicy.setInitialInterval(500);
    backOffPolicy.setMultiplier(10.0);
    backOffPolicy.setMaxInterval(10000);
    retryTemplate.setBackOffPolicy(backOffPolicy);
    template.setRetryTemplate(retryTemplate);
    return template;
}

References
https://docs.spring.io/spring-amqp/docs/current/reference/html/#template-retry
https://www.baeldung.com/spring-retry

@Configuration
public class RabbitConfiguration {

    @Bean
    public CachingConnectionFactory connectionFactory() {
        // configure to connect with credential
        ConfigFile configFile = new ConfigFile();
        CachingConnectionFactory connectionFactory =
                new CachingConnectionFactory(configFile.getRabbitmqHost());
        connectionFactory.setUsername(configFile.getRabbitmqUserName());
        connectionFactory.setPassword(configFile.getRabbitmqPassword());
        return connectionFactory;
    }

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

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

    @Bean
    public Queue MonitoringV5_Queue1() {
        // declare queue1
        return new Queue("MonitoringV5_Queue1");
    }

    @Bean
    public Queue MonitoringV5_Queue2() {
        // declare queue2
        return new Queue("MonitoringV5_Queue2");
    }
}

References
https://docs.spring.io/spring-amqp/docs/current/reference/html/#cachingconnectionfactory