multithreading - Spring 批处理 : problems (mix data) when converting to multithread

Question

也许这是一个经常出现的问题，但我需要根据我的上下文进行一些自定义。

我正在使用 Spring Batch 3.0.1.RELEASE

我有一个简单的工作，有一些步骤。一个步骤是这样的 block :

    <tasklet transaction-manager="myTransactionManager">
<batch:chunk reader="myReader" processor="myProcessor" writer="myWriter" commit-interval="${commit.interval}">
</batch:chunk>

<bean id="myProcessor" class="org.springframework.batch.item.support.CompositeItemProcessor" scope="step">
<property name="delegates">
    <list>
        <bean class="...MyFirstProcessor">
        </bean>
        <bean class="...MySecondProcessor">
        </bean>
    </list>
</property>

读者 : JdbcCursorItemReader

处理器 : CompositeProcessor 和我的代表

作家 : CompositeWriter 和我的代表

有了这个配置，我的工作就完美了。

现在，我想将其转换为多线程作业。
关注 documentation to basic multi-thread jobs ，我在 tasklet 中包含了一个 SympleAsyncTaskExecutor，但它失败了。

我读过 JdbcCursorItemReader 在多线程执行时不能正常工作(对吗？)。我把reader改成了JdbcPagingItemReader，真是一场噩梦:作业没有失败，写过程还可以，但是线程间数据混杂，客户数据不正确和连贯(客户有服务，地址，其他人等)。

那么，为什么会发生呢？我怎样才能更改为多线程作业？

复合处理器和编写器是否适合多线程？

如何制作自定义线程安全复合处理器？

也许它可能是 JDBC 阅读器:是否有任何线程安全的多线程 JDBC 阅读器？

我对此非常困惑和困惑，因此非常感谢任何帮助。
非常感谢。

[编辑 - 已解决]

好吧，解决我的问题的正确和合适的方法是从一开始就为多线程和线程安全执行设计工作。习惯性的先练习单线程单步执行，了解和了解Spring Batch的概念；但是如果你认为你正在离开这个阶段，那么像不可变对象(immutable对象)、线程安全列表、映射等这样的考虑必须提出来。

我的问题当前状态的当前修复是我稍后描述的下一个。在测试了 Martin 的建议并考虑了 Michael 的指导方针之后，我终于尽可能地解决了我的问题。接下来的步骤不是很好的做法，但我无法从一开始就重建我的工作:

将 itemReader 更改为 JdbcPagingItemReader，并将 setState 设置为 false。

通过 CopyOnWriteArrayList 更改列表。

通过 ConcurrentHashMap 改变 HashMap。

在每个委托(delegate)处理器中，通过传递上下文(实现 ApplicationContextAware)并获取 bean 的唯一实例(将每个注入(inject)的 bean 配置为 scope="prototype")，获取每个 bean 属性的新实例(幸运的是，只有一个注入(inject)的 bean) )。

因此，如果委托(delegate) bean 是:

<bean class="...MyProcessor">
<property name="otherBean"  ref="otherBeanID" />

改成:

<bean class="...MyProcessor">
<property name="otherBean"  value="otherBeanID" />

并且，在 MyProcessor 中，从上下文中获取 otherBeanID 的单个实例； otherBeanID 必须配置为 scope="protoype"。

正如我之前所说，它们的风格不好，但这是我最好的选择，我可以断言每个线程都有自己的和不同的项目实例和其他 bean 实例。

它证明了一些类没有为正确的多线程执行而设计。

马丁，迈克尔，感谢您的支持。

我希望它对任何人都有帮助。

Answer 1

您在问题中提出了很多问题(将来，请将此类问题分解为多个更具体的问题)。但是，逐项:

是 JdbcCursorItemReader线程安全？
作为documentation states ， 它不是。原因是 JdbcCursorItemReader包装单个 ResultSet这不是线程安全的。

复合处理器和编写器是否适合多线程？ CompositeItemProcessor只要委托(delegate) ItemProcessor，Spring Batch 提供的就被认为是线程安全的。实现也是线程安全的。您没有提供与您的实现或其配置相关的代码，因此我无法验证它们的线程安全性。但是，鉴于您所描述的症状，我的直觉是您的代码中存在某种形式的线程安全问题。

您也无法识别 ItemWriter您正在使用的实现或其配置，因此那里也可能存在与线程相关的问题。

如果您使用有关您的实现和配置的更多信息更新您的问题，我们可以提供更多见解。

如何制作自定义线程安全复合处理器？
实现任何 ItemProcessor 时需要考虑两件事:

使其线程安全:遵循基本的线程安全规则(阅读本书 Java Concurrency In Practice 了解有关该主题的圣经)将允许您通过添加任务执行器来扩展组件。

使其具有幂等性:在跳过/重试处理期间，可能会重新处理项目。通过制作您的ItemProcessor实现幂等，这将防止这种多次通过处理器的副作用。

也许它可能是 JDBC 阅读器:是否有任何线程安全的多线程 JDBC 阅读器？
正如您所指出的， JdbcPaginingItemReader是线程安全的，并在 documentation 中注明.当使用多个线程时，每个 block 都在它自己的线程中执行。如果您已将页面大小配置为与提交间隔匹配，则意味着每个页面都在同一个线程中处理。

单步缩放的其他选项
当您走上实现单个多线程步骤的道路时，可能会有更好的选择。 Spring Batch 提供了 5 个核心扩展选项:

多线程步骤 - 正如您现在正在尝试的那样。

并行步骤 - 使用 Spring Batch 的拆分功能，您可以并行执行多个步骤。假设您正在使用复合 ItemProcessor和复合 ItemWriter在同一步骤中，这可能是需要探索的东西(将您当前的复合场景分解为多个并行步骤)。

异步 ItemProcessor/ItemWriters - 此选项允许您在不同的线程中执行处理器逻辑。处理器关闭线程并返回 Future到AsyncItemWriter这将阻塞到 Future返回被写入。

分区 - 这是将数据划分为称为分区的 block ，由子步骤并行处理。每个分区都由一个实际的、独立的步骤处理，因此使用步骤范围的组件可以防止线程安全问题(每个步骤都有自己的实例)。分区处理可以通过线程在本地执行，也可以跨多个 JVM 远程执行。

远程分 block - 此选项将处理器逻辑分配给其他 JVM 进程。它真的应该只在 ItemProcessor 时使用。逻辑是流程中的瓶颈。

您可以在此处阅读 Spring Batch 文档中的所有这些选项: http://docs.spring.io/spring-batch/trunk/reference/html/scalability.html

线程安全是一个复杂的问题。只需将多个线程添加到过去在单线程环境中工作的代码通常会发现代码中的问题。