I'm working on a service (spring-boot) that takes a list of ids, one-by-one fetches objects from DB, aggregates those objects into batches, and then saves them somewhere else. Currently, the batch size after aggregation is ~50 objects which is roughly 10 objects per second being sent to the aggregation route. Now I'm trying to understand how I can speed up my routes or at least get some idea where is the bottleneck.
This is the bean that sends requests concurrently to direct:findObject
@Produce(uri = "direct:findObject")
private ProducerTemplate producerTemplate;
public void send() {
List<Long> listOfIDs = someService.getIDList(); //1000000 ids in the list
listOfIDs.parallelStream().forEach(producerTemplate::sendBody);
}
My camel routes:
from(direct:findObject)
.bean(objectDaoBean)
.marshal().json(JsonLibrary.Gson, MyObject.class)
.to("direct:aggregator");
from("direct:aggregator")
.unmarshal().json(JsonLibrary.Gson, MyObject.class)
.aggregate(constant(true), new GroupedBodyAggregationStrategy())
.parallelProcessing()
.completionInterval(TimeUnit.SECONDS.toMillis(5))
.completionSize(1000)
.marshal(new GsonDataFormat(new TypeToken<List<MyObject>>() {}.getType()))
.to("direct:bulkSave");
from("direct:bulkSave")
.unmarshal(new GsonDataFormat(new TypeToken<List<MyObject>>() {}.getType()))
.bean(bulkSaveBean);
Any help or insights on how I can make the whole process faster is greatly appreciated.
Update: I've managed to increase the overral performance (at least x10) with changes propoused by @Fábio Carmo Santos
application.properties:
camel.threadpool.pool-size=10
camel.threadpool.max-pool-size=400
camel.threadpool.max-queue-size=-1
Updated producer:
@Produce("direct:splitter")
private ProducerTemplate producerTemplate;
public void sendList() {
List<Long> listOfIDs = getIDList(); //1000000 ids in the list
producerTemplate.sendBody(listOfIDs);
}
Modified camel routes:
from("direct:splitter")
.split(body())
.parallelProcessing()
.bean(new DaoBean()) //fetch an object by id from db here
.to("direct:aggregator");
from("direct:aggregator")
.aggregate(body(), new GroupedBodyAggregationStrategy())
.parallelProcessing()
.completionInterval(TimeUnit.SECONDS.toMillis(5))
.completionSize(1000)
.to("direct:bulkSave");
from("direct:bulkSave")
.bean(new SaveBean());
Before I had in logs:
Saving... Batch size 30
Saving... Batch size 45
Saving... Batch size 41
Saving... Batch size 53
Saving... Batch size 47
Saving... Batch size 50
And overral execution time was at least hours (never waited until it finishes)
Now it's way better and faster
Saving... Batch size 499
Saving... Batch size 1000
Saving... Batch size 401
Saving... Batch size 1000
Saving... Batch size 257
Saving... Batch size 1000