Apache Camel 3.14.5：SqlEndpoint如何配置自定义BeanPropertyRowMapper

问题背景 #

使用Apache Camel 3.14.5版本，当前路由中SqlEndpoint的使用方式如下：

.to("mySqlComponent:classpath:my_sql.sql?outputType=StreamList&outputClass=com.my.project.MyCustomMappedPojo")

查看DefaultSqlEndpoint源码发现，queryForStreamList方法硬编码使用BeanPropertyRowMapper，无法直接替换为自定义RowMapper：

@SuppressWarnings("unchecked")
public ResultSetIterator queryForStreamList(Connection connection, Statement statement, ResultSet rs) throws SQLException {
    if (outputClass == null) {
        RowMapper rowMapper = new ColumnMapRowMapper();
        return new ResultSetIterator(connection, statement, rs, rowMapper);
    } else {
        Class<?> outputClzz = getCamelContext().getClassResolver().resolveClass(outputClass);
        RowMapper rowMapper = new BeanPropertyRowMapper(outputClzz);
        return new ResultSetIterator(connection, statement, rs, rowMapper);
    }
}

尝试直接给SqlEndpoint传入自定义RowMapper，但无对应配置项；尝试自定义SqlEndpoint并注入SqlComponent，却发现SqlComponent硬编码使用默认端点，无法替换。

解决方案 #

针对3.14.5版本的限制，可通过扩展SqlComponent和SqlEndpoint的方式实现自定义RowMapper配置，具体步骤如下：

1. 自定义SqlEndpoint，重写RowMapper逻辑 #

继承DefaultSqlEndpoint，添加自定义RowMapper配置项，并在查询方法中优先使用自定义实现：

public class CustomSqlEndpoint extends DefaultSqlEndpoint {
    private String customRowMapperRef;
    private RowMapper customRowMapper;

    public CustomSqlEndpoint(String uri, SqlComponent component) {
        super(uri, component);
    }

    // 通过Spring Bean引用传入自定义RowMapper
    public void setCustomRowMapperRef(String customRowMapperRef) {
        this.customRowMapperRef = customRowMapperRef;
    }

    // 直接注入RowMapper实例
    public void setCustomRowMapper(RowMapper customRowMapper) {
        this.customRowMapper = customRowMapper;
    }

    @Override
    public ResultSetIterator queryForStreamList(Connection connection, Statement statement, ResultSet rs) throws SQLException {
        if (outputClass == null) {
            return super.queryForStreamList(connection, statement, rs);
        } else {
            RowMapper rowMapper;
            // 优先使用直接注入的实例
            if (customRowMapper != null) {
                rowMapper = customRowMapper;
            } else if (customRowMapperRef != null) {
                // 从Camel注册表中获取指定引用的RowMapper
                rowMapper = getCamelContext().getRegistry().lookupByNameAndType(customRowMapperRef, RowMapper.class);
            } else {
                // 未配置时 fallback到默认实现
                Class<?> outputClzz = getCamelContext().getClassResolver().resolveClass(outputClass);
                rowMapper = new BeanPropertyRowMapper(outputClzz);
            }
            return new ResultSetIterator(connection, statement, rs, rowMapper);
        }
    }
}

2. 自定义SqlComponent，返回自定义Endpoint #

继承SqlComponent，重写创建端点的方法，返回自定义的CustomSqlEndpoint：

public class CustomSqlComponent extends SqlComponent {
    @Override
    protected DefaultSqlEndpoint createEndpoint(String uri, String remaining, Map<String, Object> parameters) throws Exception {
        CustomSqlEndpoint endpoint = new CustomSqlEndpoint(uri, this);
        setProperties(endpoint, parameters);
        return endpoint;
    }
}

3. Spring中注册自定义组件和RowMapper #

在Spring配置里注册自定义RowMapper和SqlComponent，替换原有的SqlComponent：

<!-- 注册自定义RowMapper实例 -->
<bean id="myCustomRowMapper" class="com.my.project.MyCustomRowMapper"/>

<!-- 注册自定义SqlComponent，替换原mySqlComponent -->
<bean id="mySqlComponent" class="com.my.project.CustomSqlComponent">
    <property name="dataSource" ref="yourDataSource"/>
</bean>

4. 路由中使用自定义RowMapper #

在路由的SqlEndpoint参数中添加customRowMapperRef，指定自定义RowMapper的Spring Bean ID：

.to("mySqlComponent:classpath:my_sql.sql?outputType=StreamList&outputClass=com.my.project.MyCustomMappedPojo&customRowMapperRef=myCustomRowMapper")

补充说明 #

• 该方案仅适配Apache Camel 3.14.5版本，更高版本（如3.18+、4.x）已原生支持rowMapper参数配置，无需扩展组件。
• 若需全局默认RowMapper，可直接在CustomSqlComponent中注入，无需在每个Endpoint单独配置。

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