无需POJO的Spring Batch分块处理实现方案咨询

实现步骤 #

1. 自定义数据库读取器（无POJO读取） #

使用JdbcCursorItemReader，通过自定义RowMapper将数据库每行结果转为Map<String, Object>，避免编写POJO类。同时设置fetchSize提升批量读取性能：

@Bean
public JdbcCursorItemReader<Map<String, Object>> dbReader(DataSource dataSource) {
    JdbcCursorItemReader<Map<String, Object>> reader = new JdbcCursorItemReader<>();
    reader.setDataSource(dataSource);
    reader.setSql("SELECT * FROM target_table"); // 替换为目标表查询语句
    reader.setRowMapper((resultSet, rowNum) -> {
        Map<String, Object> rowMap = new HashMap<>();
        ResultSetMetaData metaData = resultSet.getMetaData();
        int columnCount = metaData.getColumnCount();
        // 遍历ResultSet所有列，存入Map
        for (int i = 1; i <= columnCount; i++) {
            String columnName = metaData.getColumnName(i);
            Object value = resultSet.getObject(i);
            rowMap.put(columnName, value);
        }
        return rowMap;
    });
    reader.setFetchSize(1000); // 根据数据库性能调整，批量拉取数据
    return reader;
}

2. 数据转换处理器（Map转JSON） #

利用Jackson的ObjectMapper将Map转为JSON字符串，作为Elasticsearch的文档内容：

@Bean
public ItemProcessor<Map<String, Object>, String> jsonProcessor(ObjectMapper objectMapper) {
    return rowMap -> {
        try {
            return objectMapper.writeValueAsString(rowMap);
        } catch (JsonProcessingException e) {
            throw new RuntimeException("转换行数据到JSON失败", e);
        }
    };
}

如果需要为Elasticsearch文档指定_id或自定义结构，可以在这里扩展逻辑（比如从Map中提取主键字段作为_id，包装成符合ES要求的JSON结构）。

3. Elasticsearch批量写入器 #

使用Elasticsearch的RestHighLevelClient实现批量写入，接收分块后的JSON字符串列表，组装BulkRequest提交：

@Bean
public ItemWriter<String> esWriter(RestHighLevelClient restHighLevelClient) {
    return items -> {
        BulkRequest bulkRequest = new BulkRequest();
        for (String jsonDoc : items) {
            // 替换为目标索引名，可根据业务动态调整
            IndexRequest indexRequest = new IndexRequest("target_index")
                    .source(jsonDoc, XContentType.JSON);
            // 可选：指定文档ID，例如从JSON中提取主键
            // String docId = objectMapper.readTree(jsonDoc).get("id").asText();
            // indexRequest.id(docId);
            bulkRequest.add(indexRequest);
        }
        if (!bulkRequest.requests().isEmpty()) {
            restHighLevelClient.bulk(bulkRequest, RequestOptions.DEFAULT);
        }
    };
}

4. 组装Spring Batch Job与Step #

将Reader、Processor、Writer组合成分块处理的Step，并配置多线程执行提升性能：

@Bean
public Step dataToEsStep(JobRepository jobRepository, PlatformTransactionManager transactionManager,
                         JdbcCursorItemReader<Map<String, Object>> dbReader,
                         ItemProcessor<Map<String, Object>, String> jsonProcessor,
                         ItemWriter<String> esWriter) {
    return new StepBuilder("dataToEsStep", jobRepository)
            .<Map<String, Object>, String>chunk(1000, transactionManager) // 分块大小，按需调整
            .reader(dbReader)
            .processor(jsonProcessor)
            .writer(esWriter)
            .taskExecutor(new SimpleAsyncTaskExecutor()) // 多线程处理，注意数据库连接池大小匹配
            .build();
}

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

性能优化要点 #

• 分块大小调整：根据数据库和ES的承载能力，将chunk大小设置在1000-5000之间，平衡内存占用与IO效率。
• 数据库连接池配置：多线程场景下，确保连接池最大连接数大于等于Step的并发线程数，避免连接耗尽。
• Elasticsearch批量优化：设置bulkRequest.setRefreshPolicy(WriteRequest.RefreshPolicy.NONE)减少ES实时刷新开销；同时调整ES集群的thread_pool.bulk参数提升批量处理能力。
• 分区处理（可选）：如果数据量极大，可将数据按主键范围分片，使用Spring Batch的分区功能并行处理不同分片数据，进一步缩短耗时。

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