如何在多线程配置下按原序读取处理FlatFileItemReader的文件条目？

核心思路 #

多线程处理时，读取阶段保持单线程以保证顺序，给每条记录附加唯一的顺序标识；处理阶段用多线程并行执行；最后在写入前按顺序标识重新排序，确保输出和文件初始顺序一致。

具体实现步骤 #

1. 给读取的条目添加顺序号 #

先创建一个包装类，绑定原始条目与顺序号：

public class OrderedItem<T> {
    private final long sequenceId;
    private final T item;

    public OrderedItem(long sequenceId, T item) {
        this.sequenceId = sequenceId;
        this.item = item;
    }

    // getter方法
    public long getSequenceId() { return sequenceId; }
    public T getItem() { return item; }
}

再实现一个Reader代理，在读取时为每个条目分配递增的顺序号：

public class SequencingItemReader<T> implements ItemReader<OrderedItem<T>> {
    private final ItemReader<T> delegate;
    private AtomicLong sequenceGenerator = new AtomicLong(0);

    public SequencingItemReader(ItemReader<T> delegate) {
        this.delegate = delegate;
    }

    @Override
    public OrderedItem<T> read() throws Exception {
        T item = delegate.read();
        if (item == null) {
            return null;
        }
        return new OrderedItem<>(sequenceGenerator.incrementAndGet(), item);
    }
}

将这个代理包装在你的FlatFileItemReader外层，读取阶段必须保持单线程。

2. 多线程处理条目 #

使用AsyncItemProcessor实现多线程处理，它会返回Future对象，不会破坏顺序标识：

@Bean
public AsyncItemProcessor<OrderedItem<YourDomain>, OrderedItem<YourDomain>> asyncItemProcessor() {
    AsyncItemProcessor<OrderedItem<YourDomain>, OrderedItem<YourDomain>> processor = new AsyncItemProcessor<>();
    processor.setDelegate(yourActualItemProcessor()); // 传入你的业务ItemProcessor
    processor.setTaskExecutor(taskExecutor()); // 配置你的线程池
    return processor;
}

3. 按顺序写入结果 #

使用AsyncItemWriter，写入前先按顺序号排序：

@Bean
public AsyncItemWriter<OrderedItem<YourDomain>> asyncItemWriter() {
    AsyncItemWriter<OrderedItem<YourDomain>> writer = new AsyncItemWriter<>();
    writer.setDelegate(items -> {
        // 按sequenceId排序条目
        List<OrderedItem<YourDomain>> sortedItems = new ArrayList<>(items);
        sortedItems.sort(Comparator.comparingLong(OrderedItem::getSequenceId));
        // 提取原始业务条目，交给实际的ItemWriter处理
        List<YourDomain> actualItems = sortedItems.stream()
                .map(OrderedItem::getItem)
                .collect(Collectors.toList());
        yourActualItemWriter().write(actualItems);
    });
    return writer;
}

4. 配置Step #

读取阶段不要配置多线程，处理和写入使用上述异步组件：

@Bean
public Step orderedMultiThreadedStep() {
    return stepBuilderFactory.get("orderedMultiThreadedStep")
            .<OrderedItem<YourDomain>, OrderedItem<YourDomain>>chunk(100)
            .reader(new SequencingItemReader<>(flatFileItemReader()))
            .processor(asyncItemProcessor())
            .writer(asyncItemWriter())
            .build();
}

注意事项 #

• 读取阶段必须保持单线程，否则顺序号的连续性和正确性无法保证。
• 线程池大小需根据系统资源合理配置，避免过度并发导致性能下降。
• 处理逻辑中的共享资源需保证线程安全，避免状态混乱。

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