Spring Batch如何对嵌套对象StoreTransaction应用Chunk处理

你当前配置以整个PurchaseHistoryList为Chunk处理单位，要实现将Chunk逻辑应用到嵌套的StoreTransaction列表（比如Chunk Size=1时，Writer每次仅接收含单个StoreTransaction的PurchaseHistoryList），有两种可行方案：

方案一：通过ItemProcessor拆分完整PurchaseHistoryList #

保持原StaxEventItemReader读取整个PurchaseHistoryList，自定义ItemProcessor将其拆分为多个仅含单个StoreTransaction的PurchaseHistoryList实例，Spring Batch会自动展开Processor返回的集合，将每个拆分后的对象作为独立Item进入Chunk流程。

自定义Processor代码 #

import org.springframework.batch.item.ItemProcessor;
import java.util.ArrayList;
import java.util.List;

public class PurchaseHistorySplitterProcessor implements ItemProcessor<PurchaseHistoryList, List<PurchaseHistoryList>> {

    @Override
    public List<PurchaseHistoryList> process(PurchaseHistoryList input) {
        List<PurchaseHistoryList> splitLists = new ArrayList<>();
        List<StoreTransaction> transactions = input.getStoreTransactions();

        for (StoreTransaction transaction : transactions) {
            PurchaseHistoryList singleTxList = new PurchaseHistoryList();
            // 复制原PurchaseHistoryList的公共属性（如批次ID、用户ID等）
            singleTxList.setBatchId(input.getBatchId());
            singleTxList.setStoreTransactions(List.of(transaction));
            splitLists.add(singleTxList);
        }
        return splitLists;
    }
}

Batch配置（Java Config） #

@Configuration
@EnableBatchProcessing
public class BatchConfiguration {

    @Autowired
    private JobBuilderFactory jobBuilderFactory;

    @Autowired
    private StepBuilderFactory stepBuilderFactory;

    @Bean
    public StaxEventItemReader<PurchaseHistoryList> purchaseHistoryReader() {
        StaxEventItemReader<PurchaseHistoryList> reader = new StaxEventItemReader<>();
        reader.setResource(new ClassPathResource("purchase-history.xml"));
        reader.setFragmentRootElementName("purchaseHistoryList");
        reader.setUnmarshaller(jaxbMarshaller());
        return reader;
    }

    @Bean
    public Jaxb2Marshaller jaxbMarshaller() {
        Jaxb2Marshaller marshaller = new Jaxb2Marshaller();
        marshaller.setClassesToBeBound(PurchaseHistoryList.class, StoreTransaction.class);
        return marshaller;
    }

    @Bean
    public PurchaseHistorySplitterProcessor splitterProcessor() {
        return new PurchaseHistorySplitterProcessor();
    }

    @Bean
    public ItemWriter<PurchaseHistoryList> purchaseHistoryWriter() {
        return items -> {
            for (PurchaseHistoryList list : items) {
                StoreTransaction tx = list.getStoreTransactions().get(0);
                // 执行写入逻辑，比如持久化到数据库
                System.out.printf("Processing transaction ID: %s%n", tx.getTransactionId());
            }
        };
    }

    @Bean
    public Step processTxStep() {
        return stepBuilderFactory.get("processTxStep")
                .<PurchaseHistoryList, PurchaseHistoryList>chunk(1) // Chunk Size设为1，每次处理单个拆分后的对象
                .reader(purchaseHistoryReader())
                .processor(splitterProcessor())
                .writer(purchaseHistoryWriter())
                .build();
    }

    @Bean
    public Job purchaseHistoryJob() {
        return jobBuilderFactory.get("purchaseHistoryJob")
                .start(processTxStep())
                .build();
    }
}

方案二：直接读取StoreTransaction片段（适合大文件） #

如果XML文件体积较大，一次性加载整个PurchaseHistoryList会占用过多内存，可以直接配置StaxEventItemReader读取单个StoreTransaction节点，再通过Processor封装为PurchaseHistoryList对象。

调整后的Reader和Processor #

@Bean
public StaxEventItemReader<StoreTransaction> storeTransactionReader() {
    StaxEventItemReader<StoreTransaction> reader = new StaxEventItemReader<>();
    reader.setResource(new ClassPathResource("purchase-history.xml"));
    reader.setFragmentRootElementName("storeTransaction"); // 对应XML中单个交易的节点名
    reader.setUnmarshaller(jaxbMarshaller());
    return reader;
}

@Bean
public ItemProcessor<StoreTransaction, PurchaseHistoryList> txToListProcessor() {
    return transaction -> {
        PurchaseHistoryList list = new PurchaseHistoryList();
        // 可通过StepExecution上下文传入公共属性，比如批次ID
        list.setBatchId("BATCH_001");
        list.setStoreTransactions(List.of(transaction));
        return list;
    };
}

// 调整Step配置
@Bean
public Step processTxStep() {
    return stepBuilderFactory.get("processTxStep")
            .<StoreTransaction, PurchaseHistoryList>chunk(1)
            .reader(storeTransactionReader())
            .processor(txToListProcessor())
            .writer(purchaseHistoryWriter())
            .build();
}

关键注意点 #

• 方案一中，Spring Batch会自动展开Processor返回的List<PurchaseHistoryList>，每个元素作为独立Item进入Chunk，因此Chunk Size设为1时，Writer每次仅接收一个含单个交易的PurchaseHistoryList。
• 若PurchaseHistoryList包含公共属性（如批次ID），务必在拆分/封装时复制这些属性，避免丢失上下文信息。
• 大文件场景优先选择方案二，减少内存占用。

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