Spring Boot项目中如何优化Apache Camel路由提升批量数据处理速度

问题描述 #

我正在开发一个Spring Boot服务，该服务接收ID列表后会逐个从数据库查询对应对象，将这些对象聚合成批次后保存到其他存储位置。目前聚合后的批次大小约为50个对象，流入聚合路由的速度约为每秒10个对象，我需要找到性能瓶颈并提升整体路由处理速度。

原始代码 #

并发请求发送Bean代码

@Produce(uri = "direct:findObject")
private ProducerTemplate producerTemplate;

public void send() {
    List<Long> listOfIDs = someService.getIDList(); // 列表包含100万个ID
    listOfIDs.parallelStream().forEach(producerTemplate::sendBody);
}

原始Camel路由定义

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);

优化方案与效果 #

采纳优化方案后整体性能至少提升10倍，具体修改内容如下：

配置调整（application.properties） #

camel.threadpool.pool-size=10
camel.threadpool.max-pool-size=400
camel.threadpool.max-queue-size=-1

更新后的生产者代码 #

@Produce("direct:splitter")
private ProducerTemplate producerTemplate;

public void sendList() {
    List<Long> listOfIDs = getIDList(); // 列表包含100万个ID
    producerTemplate.sendBody(listOfIDs);
}

修改后的Camel路由 #

from("direct:splitter")
        .split(body())
        .parallelProcessing()
        .bean(new DaoBean()) // 按ID从数据库查询对象
        .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());

优化前后效果对比 #

优化前

整体执行时间至少需要数小时（从未等待到执行完成），日志输出如下：

Saving... Batch size 30
Saving... Batch size 45
Saving... Batch size 41
Saving... Batch size 53
Saving... Batch size 47
Saving... Batch size 50

优化后

性能大幅提升，日志输出如下：

Saving... Batch size 499
Saving... Batch size 1000
Saving... Batch size 401
Saving... Batch size 1000
Saving... Batch size 257
Saving... Batch size 1000

核心优化点 #

• 砍掉冗余序列化开销：原始路由存在4次无意义的JSON序列化/反序列化操作，优化后全程在路由中传递Java对象，大幅降低CPU开销。
• 优化消息发送逻辑：原始逻辑循环调用ProducerTemplate发送百万次单ID请求，单次调用的上下文开销累计极高，改成单次发送全量ID列表，通过Camel内置Split组件并行拆分处理，减少了大量框架层面的额外开销。
• 适配IO密集型场景调整线程池：数据库查询属于IO密集型操作，调整Camel全局线程池参数，放大最大线程数到400，关闭队列限制，充分利用CPU等待IO的空窗期处理更多请求。
• 简化聚合逻辑：去掉聚合前后不必要的序列化步骤，直接聚合Java对象后批量存储，降低聚合器的处理压力。

内容的提问来源于stack exchange，提问作者Pavel