动态查询场景下Spring Batch FlatFileHeaderCallback优化方案问询

问题背景 #

开发Spring Batch应用时，需要从数据库读取通过JobParameters注入的动态SQL结果并写入CSV文件，CSV表头需与查询列一致，但列不固定无法硬编码。尝试用FlatFileHeaderCallback时，因表头写入时机（FlatFileItemWriter.doOpen阶段）早于RowMapper缓存列名的时机，导致无法获取动态列名。目前已通过自定义FlatFileItemWriter结合JobScope的ColumnNamesHolder实现需求，但希望更优雅的解法。

优化方案 #

方案一：预读SQL元数据获取列名（推荐） #

在JdbcCursorItemReader初始化时，直接执行SQL获取ResultSetMetaData，提前缓存列名，这样FlatFileHeaderCallback可以直接使用缓存的列名，无需修改Writer逻辑，完全贴合Spring Batch原生组件设计。

@Slf4j
@Configuration
@EnableBatchProcessing
public class OracleToSFTPJobConfig {

    @Qualifier("transactionManager")
    @Autowired
    PlatformTransactionManager transactionManager;

    @Bean("csvJob")
    public Job oracleToSFTPJob(Step oracleToCsvStep, JobRepository jobRepository) {
        return new JobBuilder("oracleToSFTPJob", jobRepository)
                .start(oracleToCsvStep)
                .build();
    }

    @Bean
    public Step oracleToCsvStep(JdbcCursorItemReader<Row> jdbcCursorItemReader,
                                 FlatFileItemWriter<Row> csvItemWriter, JobRepository jobRepository) {
        return new StepBuilder("oracleToCsvStep", jobRepository)
                .<Row, Row>chunk(2, transactionManager)
                .reader(jdbcCursorItemReader)
                .writer(csvItemWriter)
                .build();
    }

    @StepScope
    @Bean
    public JdbcCursorItemReader<Row> jdbcCursorItemReader(DataSource dataSource, 
                                                          @Value("#{jobParameters['sql']}") String sql,
                                                          ColumnNamesHolder columnNamesHolder) {
        JdbcCursorItemReader<Row> reader = new JdbcCursorItemReader<>();
        reader.setDataSource(dataSource);
        reader.setSql(sql);
        // 预读SQL元数据，缓存列名
        try (Connection conn = dataSource.getConnection();
             PreparedStatement stmt = conn.prepareStatement(sql);
             ResultSet rs = stmt.executeQuery()) {
            ResultSetMetaData metaData = rs.getMetaData();
            int columnCount = metaData.getColumnCount();
            for (int i = 1; i <= columnCount; i++) {
                columnNamesHolder.getColumnNames().add(metaData.getColumnName(i));
            }
        } catch (SQLException e) {
            throw new RuntimeException("获取列名失败", e);
        }
        reader.setRowMapper(rowMapper());
        return reader;
    }

    @Bean
    public RowMapper<Row> rowMapper() {
        return (rs, rowNum) -> {
            log.info("读取第{}行数据", rowNum);
            ResultSetMetaData metaData = rs.getMetaData();
            int columnCount = metaData.getColumnCount();
            Map<String, Object> resultMap = new LinkedHashMap<>();
            for (int i = 1; i <= columnCount; i++) {
                String columnName = metaData.getColumnName(i);
                resultMap.put(columnName, rs.getObject(i));
            }
            return new Row(resultMap);
        };
    }

    @StepScope
    @Bean
    public FlatFileItemWriter<Row> csvItemWriter(ColumnNamesHolder columnNamesHolder) {
        FlatFileItemWriter<Row> writer = new FlatFileItemWriter<>();
        writer.setResource(new FileSystemResource("output-" + UUID.randomUUID().toString().substring(0, 3) + ".csv"));
        writer.setLineAggregator(item -> StringUtils.collectionToCommaDelimitedString(item.getDbRow().values()));
        // 原生HeaderCallback直接使用预缓存的列名
        writer.setHeaderCallback(writer -> {
            writer.write(StringUtils.collectionToCommaDelimitedString(columnNamesHolder.getColumnNames()));
        });
        return writer;
    }

    // 其余JobLauncher、TaskExecutor等配置保持不变

    @Data
    public static class ColumnNamesHolder {
        private final List<String> columnNames = new ArrayList<>();
    }

    @JobScope
    @Bean
    public ColumnNamesHolder columnNamesHolder() {
        return new ColumnNamesHolder();
    }
}

方案二：利用JobExecutionContext共享列名 #

在RowMapper第一次映射数据时，将列名存入JobExecutionContext，FlatFileHeaderCallback从上下文获取列名。需注意确保HeaderCallback执行时，上下文已存在列名（需验证Step执行顺序）。

// RowMapper修改部分
@Bean
public RowMapper<Row> rowMapper() {
    return (rs, rowNum) -> {
        ResultSetMetaData metaData = rs.getMetaData();
        int columnCount = metaData.getColumnCount();
        Map<String, Object> resultMap = new LinkedHashMap<>();
        List<String> columnNames = new ArrayList<>();
        for (int i = 1; i <= columnCount; i++) {
            String columnName = metaData.getColumnName(i);
            columnNames.add(columnName);
            resultMap.put(columnName, rs.getObject(i));
        }
        // 第一次读取时将列名存入JobExecutionContext
        if (rowNum == 0) {
            StepExecution stepExecution = StepSynchronizationManager.getContext().getStepExecution();
            stepExecution.getJobExecution().getExecutionContext().put("dynamicColumnNames", columnNames);
        }
        return new Row(resultMap);
    };
}

// FlatFileItemWriter修改部分
writer.setHeaderCallback(writer -> {
    StepExecution stepExecution = StepSynchronizationManager.getContext().getStepExecution();
    List<String> columnNames = (List<String>) stepExecution.getJobExecution().getExecutionContext().get("dynamicColumnNames");
    if (columnNames != null) {
        writer.write(StringUtils.collectionToCommaDelimitedString(columnNames));
    }
});

方案对比 #

• 方案一：最可靠，提前获取元数据，不依赖Reader的执行顺序，完全兼容Spring Batch原生组件，代码简洁易维护。
• 方案二：无需额外执行SQL，但依赖RowMapper先于HeaderCallback执行，存在时序风险，适合简单场景。

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