spring批处理程序链

Question

我需要依次执行七个不同的流程（一个接一个）。数据存储在Mysql中。我正在考虑以下选项，如果我错了，或者有更好的解决方案，请纠正我。

要求：

>

需要分块处理数据。

我的解决方案和问题：数据读取：

1. 使用JdbcCursorItemReader读取数据，因为这是性能最好的db阅读器-但是，SQL非常复杂，所以我可能不得不考虑使用JdbcTemboard的自定义ItemReader？这让我在处理数据时更加灵活。

工艺流程：

>

定义七个步骤和块，使用databean在步骤之间共享数据。但是，这不是一个好主意，因为数据在块中处理，在每个块之后，step1编写器将在数据库中创建一组新的数据。当此数据在其他步骤中共享时，数据完整性将成为一个问题。

使用StepExecutionContext在步骤之间共享数据。但这可能会影响性能，因为这涉及批处理作业存储库。

仅定义一个步骤、一个ItemReader和一系列流程（七个流程），并创建一个ItemWriter，将处理后的数据写入DB。但是，我无法管理或监控每个不同的流程，所有这些都将在一个步骤中完成。

Answer 1

建议：

1） 使用JdbcCursorItemReader读取数据。

所有开箱即用的组件都是一个不错的选择，因为它们已经实现了ItemStream接口，使您的步骤可以重新启动。但就像您提到的，有时，请求只是为了复杂，或者像我一样，您已经有了可以重用的服务或DAO。

我建议您使用ItemReaderAdapter。它允许您配置委托服务以调用以获取数据。

    <bean id="MyReader" class="xxx.adapters.MyItemReaderAdapter">
       <property name="targetObject" ref="AnExistingDao" />
       <property name="targetMethod" value="next" />        
   </bean>

请注意，目标方法必须尊重ItemReaders的读取契约（当没有更多数据时返回null）

如果您的作业不需要重新启动，您可以简单地使用类：org。springframework。一批项目适配器。ItemReaderAdapter

但是，如果需要重新启动作业，可以创建自己的ItemReaderAdapter，如下所示：

public class MyItemReaderAdapter<T> extends AbstractMethodInvokingDelegator<T> implements ItemReader<T>, ItemStream {




private long currentCount = 0;

private final String CONTEXT_COUNT_KEY = "count"; 

/**
 * @return return value of the target method.
 */
public T read() throws Exception {

    super.setArguments(new Long[]{currentCount++});
    return invokeDelegateMethod();
}
@Override
public void open(ExecutionContext executionContext)
        throws ItemStreamException {
    currentCount = executionContext.getLong(CONTEXT_COUNT_KEY,0);

}


@Override
public void update(ExecutionContext executionContext) throws ItemStreamException {
    executionContext.putLong(CONTEXT_COUNT_KEY, currentCount);
    log.info("Update Stream current count : " + currentCount);
}


@Override
public void close() throws ItemStreamException {
    // TODO Auto-generated method stub

}

}

因为开箱即用的itemReaderAdapter不可重新启动，所以您只需创建自己的实现ItemStream

2） 关于7个步骤vs 1个步骤。

在这一点上，我将使用compositeProcessor的1个步骤。7步选项只会给IMO带来问题。

1） 7步数据库：因此，您的写入程序在一个数据库中提交，直到第7步。。然后，第7步编写器尝试提交到真实的数据库并弹出错误！！！全部丢失，批处理必须从步骤1重新启动！！

2） 上下文的7个步骤：可能更好，因为状态将保存在spring批处理元数据中。。但在springBatch的元数据中存储大数据并不是一个好的做法！！

3）是去IMO的方式。；-）

Answer 2

组织。springframework。一批项目支持CompositeItemProcessor是Spring批处理框架中的一个现成组件，它将支持类似于第二个选项的需求。这将允许您执行以下操作；-在从数据库读取数据的设计/解决方案中保持分离（itemreader）-保持每个处理器关注点和配置的分离-允许任何单个处理器通过返回null来“关闭”数据块，而不管以前的进程如何

CompositeItemProcessor迭代委托循环，因此它与动作模式“类似”。它在您描述的场景中非常有用，并且仍然允许您利用区块优势（异常、重试、提交策略等）