spring integration message aggregator timeout and asynchronous aggregation clarification

Question

We are using aggregator in our spring integration application. We are processing more than 1 million product against 3 different services asynchronously using service activator where poller has been provided ThreadPoolExector.

Now what is happening is that, when say 3 thread are working on same product(biz logic to check prod against three services for it best price), all these three threads kind of wait until aggregation is completed. Now we have timeout on each service, so we are good here. But I think this is affecting the throughput of the system, since threads from the thread pool has to wait for aggregation to complete as they are handing over message to aggregator via direct channel.

Now I would like to make this channel as Queue based channel so I can make asynchronous call to aggregate responses.

If I do that, then I have few questions

1. When does the timer on specific message group starts?
2. Is it a good design to provide multiple threads to aggregator poller to do aggregation via multiple threads? If yes then how grouping works for each message?

Answer 1

The timer starts when the last message is inserted into a group.

Yes, the aggregator is thread safe - but only one thread can manipulate a particular group (identified by the correlation strategy) at a time. It uses locks to prevent multiple threads updating the same group concurrently.