Spring Batch Split Processor优化方案咨询：寻求基于记录属性实现差异化处理器链分发的更优方案

嘿，这个场景其实Spring Batch官方早就给了更贴合的解决方案，完全不用绕到ItemProcessListener里做过滤逻辑——既不优雅还容易把业务逻辑耦合到监听组件里。下面给你两个高效合理的实现思路，都是业内常用的：

Spring Batch专门为这种「按记录属性路由到不同处理链」的场景提供了ClassifierCompositeItemProcessor和ClassifierCompositeItemWriter，完全贴合你的需求，而且遵循框架设计原则，解耦性和扩展性都拉满。

步骤1：定义记录类型 #

假设你有一个基础记录类BaseRecord，包含类型标识字段，ARecord和BRecord继承它：

public abstract class BaseRecord {
    private String type;
    // getter/setter
}

public class ARecord extends BaseRecord {
    // A类型记录的专属字段
}

public class BRecord extends BaseRecord {
    // B类型记录的专属字段
}

步骤2：配置分类处理器 #

先实现一个Classifier，用来根据记录类型选择对应的处理器：

public class RecordProcessorClassifier implements Classifier<BaseRecord, ItemProcessor<BaseRecord, ? extends BaseRecord>> {
    private ItemProcessor<BaseRecord, ARecord> processorA;
    private ItemProcessor<BaseRecord, BRecord> processorB;

    @Override
    public ItemProcessor<BaseRecord, ? extends BaseRecord> classify(BaseRecord record) {
        if ("A".equals(record.getType())) {
            return processorA;
        } else if ("B".equals(record.getType())) {
            return processorB;
        }
        throw new IllegalArgumentException("Unknown record type: " + record.getType());
    }

    // 生成getter/setter
}

然后把这个分类器注入到ClassifierCompositeItemProcessor中：

@Bean
public ClassifierCompositeItemProcessor<BaseRecord, BaseRecord> compositeProcessor() {
    ClassifierCompositeItemProcessor<BaseRecord, BaseRecord> composite = new ClassifierCompositeItemProcessor<>();
    composite.setClassifier(recordProcessorClassifier());
    return composite;
}

@Bean
public RecordProcessorClassifier recordProcessorClassifier() {
    RecordProcessorClassifier classifier = new RecordProcessorClassifier();
    classifier.setProcessorA(processorA());
    classifier.setProcessorB(processorB());
    return classifier;
}

// 定义A/B处理器的Bean
@Bean
public ItemProcessor<BaseRecord, ARecord> processorA() {
    return record -> {
        // 这里写A类型记录的处理逻辑
        ARecord aRecord = new ARecord();
        // 属性转换、业务处理...
        return aRecord;
    };
}

@Bean
public ItemProcessor<BaseRecord, BRecord> processorB() {
    return record -> {
        // 这里写B类型记录的处理逻辑
        BRecord bRecord = new BRecord();
        // 属性转换、业务处理...
        return bRecord;
    };
}

步骤3：配置分类Writer #

同样的思路，实现一个Writer的分类器，把处理后的记录路由到对应Writer：

public class RecordWriterClassifier implements Classifier<BaseRecord, ItemWriter<? super BaseRecord>> {
    private ItemWriter<ARecord> writerA;
    private ItemWriter<BRecord> writerB;

    @Override
    public ItemWriter<? super BaseRecord> classify(BaseRecord record) {
        if (record instanceof ARecord) {
            return writerA;
        } else if (record instanceof BRecord) {
            return writerB;
        }
        throw new IllegalArgumentException("Unknown processed record type: " + record.getClass());
    }

    // 生成getter/setter
}

@Bean
public ClassifierCompositeItemWriter<BaseRecord> compositeWriter() {
    ClassifierCompositeItemWriter<BaseRecord> composite = new ClassifierCompositeItemWriter<>();
    composite.setClassifier(recordWriterClassifier());
    return composite;
}

@Bean
public RecordWriterClassifier recordWriterClassifier() {
    RecordWriterClassifier classifier = new RecordWriterClassifier();
    classifier.setWriterA(writerA());
    classifier.setWriterB(writerB());
    return classifier;
}

// 定义A/B Writer的Bean
@Bean
public ItemWriter<ARecord> writerA() {
    return items -> {
        // A类型记录的写入逻辑（比如DB插入、文件写入）
        items.forEach(item -> System.out.println("写入A类型记录: " + item));
    };
}

@Bean
public ItemWriter<BRecord> writerB() {
    return items -> {
        // B类型记录的写入逻辑
        items.forEach(item -> System.out.println("写入B类型记录: " + item));
    };
}

步骤4：组装Step #

最后把Reader、分类处理器、分类Writer串成一个Step：

@Bean
public Step classificationStep(JobRepository jobRepository, PlatformTransactionManager transactionManager) {
    return new StepBuilder("classificationStep", jobRepository)
            .<BaseRecord, BaseRecord>chunk(10, transactionManager) // 按批量处理
            .reader(yourItemReader()) // 替换成你自己的Reader Bean
            .processor(compositeProcessor())
            .writer(compositeWriter())
            .build();
}

如果你的业务逻辑比较简单，不想引入太多组件，也可以用「处理器返回包装对象，Writer统一分发」的方式：

步骤1：定义包装类 #

用一个包装类记录记录类型和处理后的结果：

public class RecordWrapper<T> {
    private String type;
    private T processedRecord;

    // 构造器、getter/setter
    public RecordWrapper(String type, T processedRecord) {
        this.type = type;
        this.processedRecord = processedRecord;
    }
}

步骤2：实现拆分处理器 #

在处理器里根据类型处理并返回包装对象：

@Bean
public ItemProcessor<BaseRecord, RecordWrapper<?>> splitProcessor() {
    return record -> {
        if ("A".equals(record.getType())) {
            ARecord processed = processorA().process(record);
            return new RecordWrapper<>("A", processed);
        } else if ("B".equals(record.getType())) {
            BRecord processed = processorB().process(record);
            return new RecordWrapper<>("B", processed);
        }
        throw new IllegalArgumentException("未知记录类型: " + record.getType());
    };
}

步骤3：实现分发Writer #

在Writer里按类型分组，分别调用对应Writer：

@Bean
public ItemWriter<RecordWrapper<?>> splitWriter() {
    return items -> {
        List<ARecord> aItems = new ArrayList<>();
        List<BRecord> bItems = new ArrayList<>();
        
        // 批量分组
        for (RecordWrapper<?> wrapper : items) {
            if ("A".equals(wrapper.getType())) {
                aItems.add((ARecord) wrapper.getProcessedRecord());
            } else if ("B".equals(wrapper.getType())) {
                bItems.add((BRecord) wrapper.getProcessedRecord());
            }
        }
        
        // 分别写入
        writerA().write(aItems);
        writerB().write(bItems);
    };
}

组装Step #

这个方案的Step组装更简单：

@Bean
public Step simpleSplitStep(JobRepository jobRepository, PlatformTransactionManager transactionManager) {
    return new StepBuilder("simpleSplitStep", jobRepository)
            .<BaseRecord, RecordWrapper<?>>chunk(10, transactionManager)
            .reader(yourItemReader())
            .processor(splitProcessor())
            .writer(splitWriter())
            .build();
}

你之前用ItemProcessListener.afterProcess()做过滤的方式，其实违背了Listener的设计初衷——Listener是用来做监听回调（比如日志、监控、异常兜底）的，不是用来承载业务路由逻辑的。这样做会导致：

• 业务逻辑和监听逻辑耦合，违反单一职责原则
• 没法很好地处理批量数据，效率不如官方Composite组件
• 代码可读性差，后续维护成本高

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