Spring Batch如何在ItemProcessor层面实现对象缓冲批量处理？

无需修改现有ItemReader/ItemWriter，通过Spring Batch原生的Chunk批量处理特性即可实现需求，以下是具体实现方案：

方案一：基于Chunk的批量ItemProcessor（推荐） #

Spring Batch支持直接使用ItemProcessor<List<Product>, List<Product>>类型的处理器，配合Step的chunkSize配置，自动将指定数量的Product收集为列表后传入处理器，完美适配你的批量HTTP请求需求。

1. 实现批量HTTP处理器 #

import org.springframework.batch.item.ItemProcessor;
import org.springframework.web.client.RestTemplate;
import java.util.List;
import java.util.stream.Collectors;

public class BatchHttpProcessor implements ItemProcessor<List<Product>, List<Product>> {

    private final RestTemplate restTemplate;

    public BatchHttpProcessor(RestTemplate restTemplate) {
        this.restTemplate = restTemplate;
    }

    @Override
    public List<Product> process(List<Product> products) throws Exception {
        // 提取批量请求所需的Product ID列表
        List<Long> productIds = products.stream()
                .map(Product::getId)
                .collect(Collectors.toList());

        // 发送批量HTTP请求（根据实际接口调整请求参数和响应类型）
        List<ProductApiResponse> apiResponses = restTemplate.postForObject(
                "https://your-api-endpoint/batch-process",
                productIds,
                List.class
        );

        // 将API响应映射回原有Product对象（更新业务属性）
        return products.stream()
                .map(product -> {
                    ProductApiResponse response = apiResponses.stream()
                            .filter(res -> res.getId().equals(product.getId()))
                            .findFirst()
                            .orElseThrow(() -> new RuntimeException("未找到产品ID对应的响应: " + product.getId()));
                    // 示例：更新产品状态、处理时间等字段
                    product.setProcessedStatus(response.getStatus());
                    product.setProcessedTime(response.getProcessedAt());
                    return product;
                })
                .collect(Collectors.toList());
    }

    // 接口响应DTO（根据实际接口定义调整）
    private static class ProductApiResponse {
        private Long id;
        private String status;
        private String processedAt;

        // getter/setter
        public Long getId() { return id; }
        public void setId(Long id) { this.id = id; }
        public String getStatus() { return status; }
        public void setStatus(String status) { this.status = status; }
        public String getProcessedAt() { return processedAt; }
        public void setProcessedAt(String processedAt) { this.processedAt = processedAt; }
    }
}

2. 配置Step并指定Chunk大小 #

将Step的chunkSize设置为10，Spring Batch会自动收集10个Product作为一个Chunk传入批量处理器，处理完成后直接传给原有ItemWriter：

import org.springframework.batch.core.Step;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.step.builder.StepBuilder;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.web.client.RestTemplate;

@Configuration
public class BatchConfig {

    @Bean
    public Step productProcessingStep(JobRepository jobRepository, PlatformTransactionManager transactionManager,
                                      RestTemplate restTemplate) {
        return new StepBuilder("productProcessingStep", jobRepository)
                .<Product, Product>chunk(10, transactionManager) // 每次攒10个产品作为一个处理单元
                .reader(new CustomReader()) // 复用原有Reader，无需修改
                .processor(new BatchHttpProcessor(restTemplate)) // 使用批量处理器
                .writer(new CustomWriter()) // 复用原有Writer，无需修改
                .build();
    }
}

方案优势 #

• 零侵入：完全复用现有ItemReader、ItemWriter和业务逻辑，无需修改任何原有代码
• 内存可控：Chunk大小固定为10，不会出现内存激增问题
• 事务兼容：利用Spring Batch原生事务机制，批量处理失败时整个Chunk会自动回滚
• 边界处理：自动处理最后一批不足10个的产品，无需额外逻辑

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