无需POJO实现Spring Batch通用表归档转CSV框架可行性探讨

问题描述 #

我需要搭建一个通用框架，实现数据库表归档并将归档数据导出为CSV。核心需求是通过数据库中的配置表选择多个目标表，无需为每个表编写特定的Reader、Writer和POJO，仅通过在配置表新增条目即可完成任务，不用修改代码。

我的疑问是：在Spring Batch中不使用POJO是否可行？

我尝试过用HashMap实现，但性能很差，希望能优化性能或者得到新的解决方案。

补充说明：查询语句从配置表获取。

现有实现代码 #

Reader与Writer代码 #

public ItemReader<Map<String, Object>> jdbcItemReader() {
    JdbcCursorItemReader<Map<String, Object>> reader = new JdbcCursorItemReader<>();
    reader.setDataSource(dataSource);
    reader.setSql("select a1,a2,a3 from Table");
    reader.setRowMapper(new ColumnMapRowMapper());
    return reader;
}

@Bean
public ItemWriter<Map<String, Object>> jdbcItemWriter() {
    JdbcBatchItemWriter<Map<String, Object>> writer = new JdbcBatchItemWriter<>();
    writer.setDataSource(dataSource);
    writer.setSql("INSERT INTO table1_hist (a1,a2,a3) VALUES (:a1,:a2, :a3)");
    writer.setItemSqlParameterSourceProvider(new ItemSqlParameterSourceProvider<Map<String, Object>>() {
        @Override
        public SqlParameterSource createSqlParameterSource(Map<String, Object> item) {
            MapSqlParameterSource mapSqlParameterSource = new MapSqlParameterSource();
            mapSqlParameterSource.addValues(item);
            return mapSqlParameterSource;
        }
    });
    return writer;
}

Step与Job配置代码 #

@Bean
public Step myStep(ItemReader<Map<String, Object>> reader, ItemWriter<Map<String, Object>> writer) {
    return stepBuilderFactory.get("myStep")
            .<Map<String, Object>, Map<String, Object>>chunk(10000)
            .reader(reader)
            .writer(writer)
            .build();
}

@Bean
public Job myJob(Step myStep) {
    return jobBuilderFactory.get("myJob")
            .start(myStep)
            .build();
}

解决方案 #

可行性结论 #

完全可行，Spring Batch本身支持无POJO的通用数据处理，你的HashMap方案思路正确，性能问题可通过以下方式优化，或采用更高效的替代方案：

优化HashMap实现的性能

1. 替换ColumnMapRowMapper为自定义轻量级RowMapper
   ColumnMapRowMapper包含冗余的类型转换与元数据处理逻辑，自定义RowMapper可减少开销：

public class LightweightColumnMapRowMapper implements RowMapper<Map<String, Object>> {
    @Override
    public Map<String, Object> mapRow(ResultSet rs, int rowNum) throws SQLException {
        ResultSetMetaData metaData = rs.getMetaData();
        int columnCount = metaData.getColumnCount();
        Map<String, Object> map = new HashMap<>(columnCount); // 指定初始容量避免扩容
        for (int i = 1; i <= columnCount; i++) {
            String columnName = metaData.getColumnLabel(i); // 优先使用列别名
            map.put(columnName, rs.getObject(i));
        }
        return map;
    }
}

2. 简化Writer的参数源提供器
   直接使用Spring内置的MapSqlParameterSourceProvider替代自定义实现：

writer.setItemSqlParameterSourceProvider(new MapSqlParameterSourceProvider<>());

3. 调整Chunk大小
   Chunk=10000并非最优值，需根据数据库性能、内存情况测试调整，过大的Chunk会导致内存占用过高反而降低性能。

4. 改用JdbcPagingItemReader处理大数据量
   JdbcCursorItemReader会保持长连接并占用较多内存，JdbcPagingItemReader分页加载数据，适合大数据场景：

public ItemReader<Map<String, Object>> jdbcPagingItemReader(DataSource dataSource, String sql) throws Exception {
    JdbcPagingItemReader<Map<String, Object>> reader = new JdbcPagingItemReader<>();
    reader.setDataSource(dataSource);
    reader.setRowMapper(new LightweightColumnMapRowMapper());
    
    SqlPagingQueryProvider queryProvider = new SqlPagingQueryProviderFactoryBean() {{
        setDataSource(dataSource);
        setSelectClause("SELECT a1,a2,a3");
        setFromClause("FROM Table");
        setSortKey("a1"); // 必须指定排序键用于分页
    }}.getObject();
    
    reader.setQueryProvider(queryProvider);
    reader.setPageSize(1000);
    return reader;
}

极致性能方案：直接使用JdbcTemplate批量操作

绕过Spring Batch框架开销，直接用JdbcTemplate实现动态批量归档：

@Autowired
private JdbcTemplate jdbcTemplate;

public void archiveTable(String sourceTable, String targetTable, List<String> columns) {
    String selectSql = "SELECT " + String.join(",", columns) + " FROM " + sourceTable;
    String insertSql = "INSERT INTO " + targetTable + " (" + String.join(",", columns) + ") VALUES (" 
            + columns.stream().map(col -> ":" + col).collect(Collectors.joining(",")) + ")";
    
    jdbcTemplate.query(selectSql, rs -> {
        List<Map<String, Object>> batch = new ArrayList<>(1000);
        int count = 0;
        while (rs.next()) {
            Map<String, Object> row = new HashMap<>(columns.size());
            for (String col : columns) {
                row.put(col, rs.getObject(col));
            }
            batch.add(row);
            count++;
            if (count % 1000 == 0) {
                jdbcTemplate.batchUpdate(insertSql, batch, 1000, (ps, item) -> {
                    int idx = 1;
                    for (String col : columns) {
                        ps.setObject(idx++, item.get(col));
                    }
                });
                batch.clear();
            }
        }
        if (!batch.isEmpty()) {
            jdbcTemplate.batchUpdate(insertSql, batch, batch.size(), (ps, item) -> {
                int idx = 1;
                for (String col : columns) {
                    ps.setObject(idx++, item.get(col));
                }
            });
        }
    });
}

通用CSV导出实现

基于Map实现通用CSV Writer（依赖OpenCSV）：

public class GenericCsvItemWriter implements ItemWriter<Map<String, Object>> {
    private CSVWriter csvWriter;
    private List<String> columns;

    public GenericCsvItemWriter(String filePath, List<String> columns) throws IOException {
        this.columns = columns;
        csvWriter = new CSVWriter(new FileWriter(filePath));
        csvWriter.writeNext(columns.toArray(new String[0])); // 写入表头
    }

    @Override
    public void write(List<? extends Map<String, Object>> items) throws Exception {
        String[][] data = new String[items.size()][columns.size()];
        for (int i = 0; i < items.size(); i++) {
            Map<String, Object> item = items.get(i);
            for (int j = 0; j < columns.size(); j++) {
                data[i][j] = String.valueOf(item.get(columns.get(j)));
            }
        }
        csvWriter.writeAll(Arrays.asList(data));
        csvWriter.flush();
    }

    @PreDestroy
    public void close() throws IOException {
        csvWriter.close();
    }
}

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