Is there any way to limit the channels created by the producer?

user_1675502 0
Viewed 154

We have and application that produces messages really fast and requires ACK.

This is a silly simulation:

    @Bean
    public CachingConnectionFactory ccf() {
        var ccf = new CachingConnectionFactory("localhost");
        ccf.setPublisherConfirmType(CORRELATED);
        ccf.setPublisherReturns(true);
        return ccf;
    }


    @Bean
    public ApplicationRunner run(RabbitTemplate template) {
        return args -> {
            for (int i = 1; i < 10001; i++) {
                template.convertAndSend("poc", "hey");
                if (i % 1000 == 0) {
                    LOG.info("{}", i);
            }   }
        };
    }

This code end up creating and destroying several channels, is there any way to limit the channels that are going to be created concurrently? (blocking or enqueueing)

This is one of the channels that remain cached in the rabbit console: 127.0.0.1:54190 (509).

spring-rabbit spring-amqp
1 Answers

See setChannelCheckoutTimeout.

    /**
     * Sets the channel checkout timeout. When greater than 0, enables channel limiting
     * in that the {@link #channelCacheSize} becomes the total number of available channels per
     * connection rather than a simple cache size. Note that changing the {@link #channelCacheSize}
     * does not affect the limit on existing connection(s), invoke {@link #destroy()} to cause a
     * new connection to be created with the new limit.
     * <p>
     * Since 1.5.5, also applies to getting a connection when the cache mode is CONNECTION.
     * @param channelCheckoutTimeout the timeout in milliseconds; default 0 (channel limiting not enabled).
     * @since 1.4.2
     * @see #setConnectionLimit(int)
     */
    public void setChannelCheckoutTimeout(long channelCheckoutTimeout) {

EDIT

This works as expected...

spring.rabbitmq.cache.channel.size=2
spring.rabbitmq.cache.channel.checkout-timeout=1s
spring.rabbitmq.publisher-confirm-type=correlated
spring.rabbitmq.publisher-returns=true

@SpringBootApplication
public class So69699961Application {

    public static void main(String[] args) {
        SpringApplication.run(So69699961Application.class, args).close();
    }

    @Bean
    ApplicationRunner runner(CachingConnectionFactory cf) {
        return args -> {
            Connection conn = cf.createConnection();
            Channel chann1 = conn.createChannel(false);
            Channel chann2 = conn.createChannel(false);
            try {
                Channel chann3 = conn.createChannel(false);
            }
            catch (Exception e) {
                e.printStackTrace();
            }
            chann2.close();
            chann1.close();
            conn.close();
        };
    }

}

org.springframework.amqp.AmqpTimeoutException: No available channels
user_12407630
Related