Improve documentation regarding executors [AMQP-411]

[spring-operator]

Igor Berman opened AMQP-411 and commented

There are different options to configure threads/thread factories/executors.
In addition there are differences when using spring-amqp as a producer and as a consumer. In my opinion the documentation is not clear enough. It's not clear how I can a) influence creation of threads b) which component uses what executor/factory(there are at least 4 such components) - consumers/connections threads(?)/connection executors(?)/heartbeat

1)I can provide thread factory to the rabbitmq connection factory

com.rabbitmq.client.ConnectionFactory lowLevelFactory = new com.rabbitmq.client.ConnectionFactory();
ThreadFactory threadFactory = new ThreadFactoryBuilder().setDaemon(true).setNameFormat("Message producer thread %d").build();
lowLevelFactory.setThreadFactory(threadFactory);
return new CachingConnectionFactory(lowLevelFactory);

2. I can set executor to the ampq connection factory

connectionFactory.setExecutor(daemonThreadsExecutor());
private Executor daemonThreadsExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setMaxPoolSize(1);
        executor.setCorePoolSize(1);
        executor.setDaemon(true);
        executor.initialize();
        return executor;
    }

3. I haven't found the way to set heartbeat thread to be what I want. Might be You want to take it as a different feature request/improvement

4. When defining SimpleMessageListenerContainer I can set executor

container.setTaskExecutor(taskExecuter());

It's not clear what is the difference between 1 & 2 and when each one of them is used
For producer only 1 is relevant
For consumer(as message listener) probably all of them are relevant is some way.

Thanks in advance.
Anyway enjoying your product.

---

Affects: 1.3.5

Referenced from: pull request #228

[spring-operator]

Gary Russell commented

1. Is for the rabbitMQ client itself for async calls (deliveries into the container). It uses a thread pool (5) by default - or it did the last time I looked.

2. You can also set an Executor on the Spring-AMQP as an alternative (it is passed into the client library when creating a connection). It will use that instead of the default.

3. I haven't looked at the internals of the client library to see what they use for heartbeats.

4. Is used by the container (one thread per consumer), which blocks on a queue, waiting for async deliveries from the client library.

Hope that helps a little, and we'll look at improving the docs.

[spring-operator]

Igor Berman commented

Hi, Gary, thanks for feedback
I've wrote that I'm confused regarding 1 & 2 since simple test case proves that I see threads from both mechanisms were created. Sorry to bother you, however in some cases one should control what threads are created under jvm(e.g. non-daemon can prevent from jvm to exiting etc.)

@Bean
public ConnectionFactory connectionFactory() {
        com.rabbitmq.client.ConnectionFactory lowLevelFactory = new com.rabbitmq.client.ConnectionFactory();
        ThreadFactory threadFactory = new ThreadFactoryBuilder().setDaemon(true).setNameFormat("Rabbitmq Connection thread %d").build();
        lowLevelFactory.setThreadFactory(threadFactory);
        
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory(lowLevelFactory);
        connectionFactory.setChannelCacheSize(3);
        connectionFactory.setExecutor(connectionsExecuter());
        return connectionFactory;
    }
private ThreadPoolTaskExecutor connectionsExecuter() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setDaemon(true);
        executor.setThreadNamePrefix("Executor Connections");
        executor.initialize();
        return executor;
    }

so in threads dump I see:
Daemon Thread [Rabbitmq Connection thread 0] (Running)	
Daemon Thread [Rabbitmq Connection thread 1] (Running)	

Daemon Thread [Executor Connections1] (Running)	
Daemon Thread [Executor Connections2] (Running)	
Daemon Thread [Executor Connections3] (Running)	

[spring-operator]

Gary Russell commented

1. The rabbit client ThreadFactory is used for low-level (e.g. IO) operations.
2. The Executor passed into the Spring CachingConnectionFactory is passed into the client (in newConnection()). It is used by the client to deliver new messages to the consumer. In 1.4, if supplied, this executor is also used to defer a channel close() if publisher confirms/returns are in use and the channel cache is full. If not supplied, the client uses an internal pooled executor with 5 threads; an internal cached thread pool is used for the deferred closes.
3. The Executor passed into the SimpleListenerContainer is used by the container to invoke the MessageListener. For historical reasons, the thread (from Remote host test #2) hands off to the container thread via a BlockingQueue. This was necessary with older RabbitMQ clients. We have an open JIRA issue (Add DirectMessageListenerContainer [AMQP-418] #1975) to provide an option in the listener container to invoke the MessageListener directly on the client consumer delivery thread.

I hope that helps.