多行聚合ItemReader不按照Spring批处理样本中的建议工作

Question

我正在尝试使用多行IteamReader来跟踪spring批处理示例https://github.com/spring-projects/spring-batch/tree/main/spring-batch-samples#multiline

我遇到如下编译错误-

我确信有一些与泛型相关的东西，因为它在寻找实现ItemReader的类，但是Aggregate ItemReader实现了ItemReader

public class AggregateItemReader<T> implements ItemReader<List<T>> {

你可以在这里找到我的代码-https://github.com/arpit9mittal/spring-batch-demo/blob/master/src/main/java/my/demo/batch/BatchConfiguration.java

更新：

为了调用ItemStreamReader open（）方法，我抑制了泛型并更新了AggregateItemReader，如下所示。

public class AggregateItemReader<T> implements ItemStreamReader<List<T>> {
private static final Log LOG = LogFactory.getLog(AggregateItemReader.class);

private ItemStreamReader<AggregateItem<T>> itemReader;

我注意到ItemWriter正在编写记录列表，而不是每行记录

[Trade: [isin=UK21341EAH45,quantity=978,price=98.34,customer=customer1], Trade: [isin=UK21341EAH46,quantity=112,price=18.12,customer=customer2]]
[Trade: [isin=UK21341EAH47,quantity=245,price=12.78,customer=customer3], Trade: [isin=UK21341EAH48,quantity=108,price=9.25,customer=customer4], Trade: [isin=UK21341EAH49,quantity=854,price=23.39,customer=customer5]]
[Trade: [isin=UK21341EAH47,quantity=245,price=12.78,customer=customer6], Trade: [isin=UK21341EAH48,quantity=108,price=9.25,customer=customer7], Trade: [isin=UK21341EAH49,quantity=854,price=23.39,customer=customer8]]

当我尝试添加处理器时，它会抱怨处理器无法将列表转换为交易对象。

@Bean
public ItemProcessor<Trade, Trade> processor() {
    return new ItemProcessor<Trade, Trade>() {

        @Override
        public Trade process(Trade item) throws Exception {
            item.setProcessed(true);
            return item;
        }
    };
}

@SuppressWarnings({ "rawtypes", "unchecked" })
@Bean
public Step multilineStep(
        AggregateItemReader reader, 
        ItemProcessor processor,
        FlatFileItemWriter writer,
        StepItemReadListener itemReadListener) {
    return stepBuilderFactory.get("multiLineStep")
            .chunk(1)
            .reader(reader)
            .writer(writer)
            .processor(processor)
            .build();
}

错误：

java.lang.ClassCastException: java.util.ArrayList cannot be cast to my.demo.batch.multiline.Trade
    at my.demo.batch.BatchConfiguration$2.process(BatchConfiguration.java:1) ~[main/:na]
    at org.springframework.batch.core.step.item.SimpleChunkProcessor.doProcess(SimpleChunkProcessor.java:134) ~[spring-batch-core-4.3.3.jar:4.3.3]
    at org.springframework.batch.core.step.item.SimpleChunkProcessor.transform(SimpleChunkProcessor.java:319) ~[spring-batch-core-4.3.3.jar:4.3.3]
    at org.html" target="_blank">springframework.batch.core.step.item.SimpleChunkProcessor.process(SimpleChunkProcessor.java:210) ~[spring-batch-core-4.3.3.jar:4.3.3]
    at org.springframework.batch.core.step.item.ChunkOrientedTasklet.execute(ChunkOrientedTasklet.java:77) ~[spring-batch-core-4.3.3.jar:4.3.3]

帮助：

1. 我们如何在不抑制泛型的情况下让它工作

Answer 1

您需要在批次级别上处理的项目类型保持一致。根据您的步骤定义，它是交易。通过调用

问题是你的阅读器是ItemReader类型的

如果您想使用Aggregate ItemReader，您需要将其包装到一个自定义阅读器中，该阅读器作为适配器工作，实际上返回Trade项，而不是List

例如，自定义read方法可以如下所示：

public Trade read() throws Exception {
    if (queue.isEmpty()) {
        List<Trade> trades = aggregateItemReader.read();
        if (trades != null) {
            queue.addAll(trades);
        }
    }
    return queue.poll();
}

将队列初始化为

private Deque<Trade> queue = new ArrayDeque<>();