java - 从 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但这有两个缺点:

• 由于 writer 中的嵌套列表，代码和测试更难编写和维护
• 最重要的不止一个 Account对象可能会写入同一个事务中，因为提供给 writer 的列表可能包含多个帐户，我想避免这种情况。

有什么办法，可能是使用监听器，或者替换spring批处理使用的一些内部组件来避免处理器中的列表？

更新

我打开了an issue on spring 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);
        }
    }
}