java - Spring Batch 中的多线程

Question

我有一个 Spring Batch 应用程序。此应用有一个配置为多线程的步骤。

<bean id="simpleTaskExecutor"
    class="org.springframework.core.task.SimpleAsyncTaskExecutor">
    <property name="concurrencyLimit" value="10"/>
</bean>

<batch:job id="rabbitReadJob" job-repository="jobRepository">
    <batch:step id="step1">
        <batch:tasklet task-executor="simpleTaskExecutor" throttle-limit="10">
            <batch:chunk    
                    reader="CarItemReader"
                    processor="CarItemProcessor"
                    writer="CarItemWriter"
                    commit-interval="2">
            </batch:chunk>
        </batch:tasklet>
    </batch:step>
</batch:job>

提交间隔为 2，然后项目读取器的线程读取 2 个项目并发送给进程。 item reader 的其他线程读取其他 2 个 itens 并发送到进程。两个线程的 Itens 同时执行该过程，但我们需要等待第一个项目完成才能执行第二个。为简单起见，我将发布执行日志。

INFO  06-01 12:44:53,344 - >>>>>>>>>> start processor id 1
INFO  06-01 12:44:53,355 - >>>>>>>>>> start processor id 3
INFO  06-01 12:44:53,362 - >>>>>>>>>> start processor id 5
INFO  06-01 12:44:53,456 - >>>>>>>>>> finish processor id 5
INFO  06-01 12:44:53,456 - >>>>>>>>>> finish processor id 3
INFO  06-01 12:44:53,456 - >>>>>>>>>> start processor id 6
INFO  06-01 12:44:53,456 - >>>>>>>>>> start processor id 4
INFO  06-01 12:44:53,467 - >>>>>>>>>> finish processor id 1
INFO  06-01 12:44:53,467 - >>>>>>>>>> start processor id 2
INFO  06-01 12:44:53,520 - >>>>>>>>>> finish processor id 6
INFO  06-01 12:44:53,520 - >>>>>>>>>> finish processor id 4
INFO  06-01 12:44:53,520 - >>>>>>>>>> writer id 3
INFO  06-01 12:44:53,520 - >>>>>>>>>> writer id 5
INFO  06-01 12:44:53,520 - >>>>>>>>>> writer id 4
INFO  06-01 12:44:53,520 - >>>>>>>>>> writer id 6
INFO  06-01 12:44:53,560 - >>>>>>>>>> finish processor id 2
INFO  06-01 12:44:53,560 - >>>>>>>>>> writer id 1
INFO  06-01 12:44:53,560 - >>>>>>>>>> writer id 2

为了理解，第一个线程进程 id 1 和 id 2。问题是 id 2 仅在 id 1 完成后才启动进程。看看第二个线程(id 3 和 4)和第三个线程(id 5 和 6)发生了同样的情况。

我想对 block 进行多线程处理。这可能吗？

Answer 1

多线程步骤在它自己的线程中执行每个 block 。因此，在您的情况下，如果您的 commit-interval 为 2 并且按顺序读取项目，id 1 和 2 将在一个线程上按顺序处理，id 3 和 4 将在一个线程上第二个线程，依此类推。

block 内多线程的唯一选择是使用 AsyncItemProcessor 和 AsyncItemWriter。 AsyncItemProcessor 包装了一个常规的 ItemProcessor 并在返回一个 Future 的新线程中执行它。 Future 然后传递给 AsyncItemWriter，它等待 Future 返回，然后将结果委托(delegate)给 ItemWriter 它包裹着。这种方法允许您对 block 的处理部分进行多线程处理。

让每个项目在它自己的线程中处理的唯一替代方法是将 commit-interval 更改为 1。

您可以在此处的文档中阅读有关 AsyncItemProcessor 和 AsyncItemWriter 的更多信息:http://docs.spring.io/spring-batch/trunk/reference/html/springBatchIntegration.html#asynchronous-processors