Spring Batch Boot - 从CSV读取数据时根据状态码将数据写入多个位置？

Question

我正在开发 Spring Boot Batch 示例。在此示例中，从CSV 读取数据并基于状态代码值（如SUCCESS、REJECT、PENDING、COMPLETED）我想将其写入 4 个不同的位置（MySQL、XML、Postgres , Salesforce DB) 在写 ?.我可以看到我们只能将Customers 的一种类型的列表传递给CompositeItemWriter 方法。如何将 4 个不同的 Customer 列表传递给 CompositeItemWriter ？

在这里我可以考虑使用CompositeItemProcessor 和CompositeItemWriter。在CompositeItemProcessor 中，我将决定分离所有这些处理器中的记录，现在的问题是如何将SUCCESS 传递给MYSQL，REJECT 传递给XML，PENDING 传递给Postgres和COMPLETED 到Salesforce DB?

@Bean
public CompositeItemProcessor<Customer, Customer> compositeItemProcessor() throws Exception{
    List<ItemProcessor<Customer, Customer>> delegates = new ArrayList<>();
    delegates.add(new FilteringSuccessProcessor());
    delegates.add(new FilteringRejectProcessor());
    delegates.add(new FilteringPendingProcessor());
    delegates.add(new FilteringCompletedProcessor());

    CompositeItemProcessor<Customer, Customer> processor = new CompositeItemProcessor<>();
    processor.setDelegates(delegates);
    processor.afterPropertiesSet();

    return processor;
}

FilteringSuccessProcessor.java

像这样我已经创建了处理器的所有状态代码和

public class FilteringSuccessProcessor implements ItemProcessor<Customer, Customer> {

    @Override
    public Customer process(Customer item) throws Exception {
        if(item.getStatus == "SUCCESS"){
            return item;
        }
        else
            return null;
    }
}

现在我的问题是 - 我们如何将 4 个不同的数据结果传递给 CompositeItemWriter 以将其写入 4 个不同的地方？

我虽然也使用CustomerClassifier，但是如何将四个不同的列表传递给同一个作者？

有什么指导吗？这是一个非常有趣的用例，但看起来很关键。

Answer 1

您可以使用 classifierCompositeItemWriter 并在其中定义您的路由策略：

它是一个 ItemWriter，它将使用分类器根据您的状态来处理写入不同的地方。

    @Bean
    public ClassifierCompositeItemWriter<Item> classifierCompositeItemWriter(ItemWriter<Item> POSTGRESWriter, ItemWriter<Item> SQLWiter) {
        ClassifierCompositeItemWriter<Item> classifierCompositeItemWriter = new ClassifierCompositeItemWriter<>();
        classifierCompositeItemWriter.setClassifier((Classifier<Item, ItemWriter<? super Item>>) item-> {
            if (item.getStatus("Success")) {
                return POSTGRESWriter;
            } else {
                return SQLWiter;
            }
        });
        return classifierCompositeItemWriter;
    }

您可以按照以下方式配置您的步骤：

    @Bean
    public Step process() {
        return stepBuilderFactory.get("dataExtraprocessctionStep")
                .<Item, Item>chunk(1)
                .reader(itemReader())
                .writer(classifierCompositeItemWriter(POSTGRESWriter(), SQLWriter()))
                .stream(POSTGRESWriter())
                .stream(SQLWriter())
                .build();
    }