动态查询时FlatFileHeaderCallback的使用

Question

我有一个 Spring Batch 应用程序，它从数据库读取数据并写入 CSV 文件。 CSV 文件需要一个标头，可以使用 Spring Batch 的 FlatFileItemWriter 中的 FlatFileHeaderCallback 接口提供标头。问题是从数据库读取数据的查询是动态的，因此列不是固定的，因此我无法在界面中对列名称进行硬编码。 读者有这种需求的rowmapper来动态读取每一行的数据。我想到在这里缓存列名，然后在界面中使用缓存的行：

public RowMapper<Row> rowMapper() {
        return (rs, rowNum) -> {
            ResultSetMetaData metaData = rs.getMetaData();
            int columnCount = metaData.getColumnCount();
            Map<String, Object> resultMap = new HashMap<>();
            for (int i = 1; i <= columnCount; i++) {
                String columnName = metaData.getColumnName(i);
                Object value = rs.getObject(i);
                resultMap.put(columnName, value);
            }
            return new Row(resultMap);
        };
    }

但是，这是在 FlatFileItemWriter 调用 doOpen 之后调用的，其中调用了 this.headerCallback.writeHeader(outputState.outputBufferedWriter)，因此我无法缓存列名称，然后以某种方式在回调中写入标题行。

一个解决方案是我在某处维护一个标志，然后在第一次调用编写器时在编写器中写入 csv 中的标题行后更改其状态。

有更好的方法吗？

更新： 使用以下代码更新了行映射器。它在持有者服务中缓存一次列名。

    @Bean
    public RowMapper<Row> rowMapper(ColumnNamesHolder columnNamesHolder) {
        return (rs, rowNum) -> {
            log.info("Reading a row...{}", 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);
                Object value = rs.getObject(i);
                resultMap.put(columnName, value);
                columnNames.add(columnName);
            }
            if(!columnNamesHolder.isHeaderWritten()) {
                columnNamesHolder.getColumnNames().clear();
                columnNamesHolder.getColumnNames().addAll(columnNames);
            }
            return new Row(resultMap);
        };
    }

扩展了我的编写器以更新 CSV 文件：

        @Override
        public String doWrite(Chunk items) {
            
            if(!columnNamesHolder.isHeaderWritten()) {
                try {
                    final StringBuilder header = new StringBuilder();
                    header.append(StringUtils.collectionToCommaDelimitedString(columnNamesHolder.getColumnNames()));
                    header.append(this.lineSeparator);
                    getOutputState().write(header.toString());
                    columnNamesHolder.setHeaderWritten(true);
                } catch (IOException e) {
                    throw new RuntimeException(e);
                }
            }

            return super.doWrite(items);
        }

这可行，但对我来说似乎很脏。有更好的方法吗？

Answer 1

由于您将 sql 查询作为作业参数传递，并且您预先知道列，因此您可以创建一个使用列动态配置的步骤范围标头回调。比如：

String columns = "id,name";
JobParameters jobParameters = new JobParametersBuilder()
   .addString("columns", columns)
   .addString("query", "select " + columns.split(",") + " from ...")
   .toJobParameters();

然后定义一个步骤范围的标头回调：

@Bean
@StepScope
public FlatFileHeaderCallback headerCallback(@Value("#{jobParameters['columns']}") String columns) {
        String[] columnsArray = columns.split(",");
        return writer -> writer.write(""); // write columns as needed
    }

这个回调可以在writer上设置。可以使用

query

作业参数配置阅读器。

有了这个，就不需要从映射器中的结果集中内省列并通过持有者对象与编写器共享它们。