从spring批处理ItemProcessor返回多个项目

Question

我正在编写一个Spring批处理作业,在我的一个步骤中,我有以下处理器代码:

@Component
public class SubscriberProcessor implements ItemProcessor<NewsletterSubscriber, Account>, InitializingBean {

    @Autowired
    private AccountService service;

    @Override public Account process(NewsletterSubscriber item) throws Exception {
        if (!Strings.isNullOrEmpty(item.getId())) {
            return service.getAccount(item.getId());
        }
        // search with email address
        List<Account> accounts = service.findByEmail(item.getEmail());
        checkState(accounts.size() <= 1, "Found more than one account with email %s", item.getEmail());
        return accounts.isEmpty() ? null : accounts.get(0);
    }

    @Override public void afterPropertiesSet() throws Exception {
        Assert.notNull(service, "account service must be set");
    }
}

上面的代码工作,但我发现,有些情况下有一个以上的一些优势情况下Account每个NewsletterSubscriber被允许的.所以我需要删除状态检查并将多个传递Account给项目编写者.

我找到的一个解决方案是更改两者ItemProcessor并ItemWriter处理List<Account>类型而不是,Account但这有两个缺点:

• 由于编写器中的嵌套列表,代码和测试更加难以编写和维护
• 最重要的Account是,可以在同一个事务中写入多个对象,因为给写入者的列表可能包含多个帐户,我想避免这种情况.

有没有办法,可能使用一个监听器,或者更换一些弹簧批处理使用的内部组件以避免处理器中的列表？

更新

我已经在春季Jira上为这个问题打开了一个问题.

我正在研究isComplete和getAdjustedOutputs方法,FaultTolerantChunkProcessor其中标记为扩展点,SimpleChunkProcessor以查看是否可以以某种方式使用它们来实现我的目标.

任何提示都是受欢迎的.

Answer 1

Item Processor接受一件事,并返回一个列表

MyItemProcessor implements ItemProcessor<SingleThing,List<ExtractedThingFromSingleThing>> {
    public List<ExtractedThingFromSingleThing> process(SingleThing thing) {
    //parse and convert to list
    }
}

将下游作者包裹起来以解决问题.这样,这个编写器下游的东西不必使用列表.

@StepScope
public class ItemListWriter<T> implements ItemWriter<List<T>> {
    private ItemWriter<T> wrapped;

    public ItemListWriter(ItemWriter<T> wrapped) {
        this.wrapped = wrapped;
    }

    @Override
    public void write(List<? extends List<T>> items) throws Exception {
        for (List<T> subList : items) {
            wrapped.write(subList);
        }
    }
}

Answer 2

没有一种方法可以ItemProcessor在不深入杂草的情况下每次调用返回一个以上的项目到Spring Batch 中。如果你真的想知道 anItemProcessor和ItemWriterexit的关系在哪里（不推荐），看看ChunkProcessor接口的实现。虽然简单的情况 ( SimpleChunkProcessor) 并没有那么糟糕，但如果您使用任何容错逻辑（通过 跳过/重试FaultTolerantChunkProcessor），它会变得非常笨拙。

一个更简单的选择是ItemReader在返回项目之前将此逻辑移至执行此扩充的 。将ItemReader您正在使用的任何内容包装在一个自定义ItemReader实现中，该实现在返回项目之前执行服务查找。在这种情况下，不是NewsletterSubscriber从读取器返回 a，而是Account根据先前的信息返回 a 。