如何使用Spring Batch批量读取、处理并发送数据列表至第三方

方案1：自定义ItemReader直接读取完整列表 #

绕过默认单条读取的逻辑，自己实现一个ItemReader直接从数据库读取整个数据列表，这样后续的Processor就能直接拿到完整列表做处理，最后由Writer发送给第三方。

代码示例：

@Component
public class FullListItemReader implements ItemReader<List<YourEntity>> {

    private final YourDataRepository repository;
    private boolean hasRead = false;

    public FullListItemReader(YourDataRepository repository) {
        this.repository = repository;
    }

    @Override
    public List<YourEntity> read() throws Exception {
        // 仅读取一次，避免重复读取
        if (!hasRead) {
            hasRead = true;
            // 从数据库拉取完整列表
            return repository.findAll();
        }
        // 返回null标识读取结束
        return null;
    }
}

对应的Processor和Writer也需要适配列表类型：

// 处理完整列表的Processor
@Component
public class ListItemProcessor implements ItemProcessor<List<YourEntity>, List<ThirdPartyDto>> {

    @Override
    public List<ThirdPartyDto> process(List<YourEntity> items) throws Exception {
        // 对整个列表做统一处理：格式转换、数据校验等
        return items.stream()
                .map(this::convertToThirdPartyDto)
                .filter(ThirdPartyDto::isValid)
                .collect(Collectors.toList());
    }

    private ThirdPartyDto convertToThirdPartyDto(YourEntity entity) {
        // 实体转第三方DTO的逻辑
        return new ThirdPartyDto();
    }
}

// 发送完整列表的Writer
@Component
public class ThirdPartyBatchWriter implements ItemWriter<List<ThirdPartyDto>> {

    private final ThirdPartyApiClient apiClient;

    public ThirdPartyBatchWriter(ThirdPartyApiClient apiClient) {
        this.apiClient = apiClient;
    }

    @Override
    public void write(List<? extends List<ThirdPartyDto>> lists) throws Exception {
        // 这里lists只会有一个元素（因为Reader只返回一次完整列表）
        List<ThirdPartyDto> processedList = lists.get(0);
        apiClient.sendBatchData(processedList);
    }
}

方案2：利用Chunk机制在Writer中统一处理列表 #

如果不想修改Reader，保持默认单条读取的逻辑，可以在Writer中接收Spring Batch的Chunk列表（即一次批量读取的所有数据），直接在Writer里完成列表处理和发送。这种方式更适合数据量较大、需要分Chunk处理但仍需按Chunk整体发送的场景。

代码示例：

@Component
public class ChunkBatchWriter implements ItemWriter<YourEntity> {

    private final ThirdPartyApiClient apiClient;

    public ChunkBatchWriter(ThirdPartyApiClient apiClient) {
        this.apiClient = apiClient;
    }

    @Override
    public void write(List<? extends YourEntity> items) throws Exception {
        // 对当前Chunk的整个列表做处理
        List<ThirdPartyDto> processedList = items.stream()
                .map(this::convertToDto)
                .collect(Collectors.toList());
        // 发送整个Chunk列表到第三方
        apiClient.sendBatchData(processedList);
    }

    private ThirdPartyDto convertToDto(YourEntity entity) {
        // 转换逻辑
        return new ThirdPartyDto();
    }
}

然后在Step配置中设置合适的Chunk大小（比如设置为你需要一次性发送的列表长度）：

@Bean
public Step dataSendStep(JobRepository jobRepository, PlatformTransactionManager transactionManager,
                         ItemReader<YourEntity> defaultItemReader, ChunkBatchWriter batchWriter) {
    return new StepBuilder("dataSendStep", jobRepository)
            .<YourEntity, YourEntity>chunk(1000, transactionManager) // 按1000条为一个Chunk
            .reader(defaultItemReader)
            .writer(batchWriter)
            .build();
}

注意事项 #

• 如果数据量极大，一次性读取完整列表会导致内存溢出，建议采用方案2分Chunk处理，或与第三方协商支持分批发送。
• 若必须严格发送全部数据组成的单列表，可在Step中通过ExecutionContext累积所有数据，最后在Step的afterStep回调中统一处理发送，但需注意重启时的状态恢复问题。

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