使用SQL命名查询执行批量插入时出现参数找不到错误

路由代码 #

from("{{my.app.source}}")
                .unmarshal()
                .bindy(BindyType.Csv, EmployeeCsvRecord.class)
                .split(body())
                .streaming()
                .bean("employeeService", "getMap")
                .aggregate(constant(true), new EmployeeAggregationStrategy())
                .completionSize(500)
                .log("data ready to insert into database")
                .to("{{sql.insertEmployee}}")
                .log("data inserted into database");

SQL配置 #

sql.insertEmployee=sql:insert into employee (employeeName, employeeAge, employeeGender, employeeDepartment, employeeSalary) values (:#employeeName, :#employeeAge, :#employeeGender, :#employeeDepartment, :#employeeSalary);batch=true

错误信息 #

设置查询中的命名参数时，无法在消息体或头信息中找到键[employeeName]

说明：将batch=true改为false时可正常执行，但必须使用SQL组件实现批量插入。

问题原因 #

开启批量模式时，Camel SQL组件要求消息体是包含多个参数Map的List集合（List<Map<String, Object>>），但当前聚合后的消息体格式不符合要求——你的EmployeeAggregationStrategy没有正确将单个Map对象聚合成List结构，导致SQL组件无法识别批量参数。

解决方案 #

1. 修正聚合策略 #

修改EmployeeAggregationStrategy，确保每次聚合都将单个Map添加到List中：

public class EmployeeAggregationStrategy implements AggregationStrategy {
    @Override
    public Exchange aggregate(Exchange oldExchange, Exchange newExchange) {
        List<Map<String, Object>> employeeList;

        // 首次聚合初始化List
        if (oldExchange == null) {
            employeeList = new ArrayList<>();
            employeeList.add(newExchange.getIn().getBody(Map.class));
            newExchange.getIn().setBody(employeeList);
            return newExchange;
        }

        // 后续聚合将新Map加入已有List
        employeeList = oldExchange.getIn().getBody(List.class);
        employeeList.add(newExchange.getIn().getBody(Map.class));
        return oldExchange;
    }
}

2. 调整SQL配置（修正语法错误） #

原配置中的insert =属于冗余语法，需删除，确保配置正确：

sql.insertEmployee=sql:insert into employee (employeeName, employeeAge, employeeGender, employeeDepartment, employeeSalary) values (:#employeeName, :#employeeAge, :#employeeGender, :#employeeDepartment, :#employeeSalary);batch=true

3. 验证聚合结果（可选） #

在聚合后添加日志，确认消息体是List

.aggregate(constant(true), new EmployeeAggregationStrategy())
.completionSize(500)
.log("Aggregated data structure: ${body.class}") // 输出应为java.util.ArrayList
.log("data ready to insert into database")
.to("{{sql.insertEmployee}}")

修改完成后，SQL组件会自动迭代List中的每个Map，填充对应参数完成批量插入。

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