如何使用 Spring Batch 正确聚合和拆分作业

Question

我需要实现一个批处理作业来拆分 XML、处理这些部分并在之后聚合它们。聚合需要进一步处理。

帐户部分的处理确实很昂贵。在下图中，处理部分对一个人的每个帐户执行操作（一个人的帐户数量会有所不同）。

示例输入文件：

<?xml version="1.0" encoding="UTF-8"?>
<Persons>
    <Person>
        <Name>Max Mustermann</Name>
        <Accounts>
            <Account>maxmustermann</Account>
            <Account>esel</Account>
            <Account>affe</Account>
        </Accounts>
    </Person>
    <Person>
        <Name>Petra Pux</Name>
        <Accounts>
            <Account>petty</Account>
            <Account>petra</Account>
        </Accounts>
    </Person>
        <Person>
        <Name>Einsiedler Bob</Name>
        <Accounts>
            <Account>bob</Account>
        </Accounts>
    </Person>
</Persons>

对于每个人的每个帐户，请执行以下操作： 调用 REST 服务，例如

GET /account/{person}/{account}/logins

结果为每个包含聚合登录名的人员调用休息服务-xml：

POST /analysis/logins/{person}

<Person>
    <Name>Max Mustermann</Name>
    <Accounts>
        <Account>
            <LoginCount>22</LoginCount>
            <Name>maxmustermann</Name>
        </Account>
        <Account>
            <LoginCount>42</LoginCount>
            <Name>esel</Name>
        </Account>
        <Account>
            <LoginCount>13</LoginCount>
            <Name>affe</Name>
        </Account>
    </Accounts>
</Person>

我对 API 没有任何影响，所以我需要更新人员包。

如何实现并行化并因此构建我的 spring 批处理应用程序？

我找到了一些起点，但没有一个是真正令人满意的。

我应该在一个步骤中处理帐户数据，返回属于每个帐户的项目并在下一步中轮询每个项目并聚合这些，还是应该在帐户步骤中实现并行化并聚合它在这一步里面，引入下一步进一步处理？

第一种方法的问题：我应该如何知道所有项目都已到达开始聚合？

第二种方法的问题（或者更确切地说是关于）的问题：手动实现并行化（例如 Java Futures）而不是将其留给 Spring Batch 是否常见？

Spring Batch 是什么样的方式？

谢谢，托马斯

Answer 1

您没有在问题中解释很多事情，例如 - 您的出发点以及为什么不令人满意？

此外，您应该已经展示了一个示例输入和输出 xml。您的最后一段非常难以阅读和理解，所以请写得更好。

话虽如此，我想就 Java 而言，您输入的类型为 List<Person>，其中 Person 就像 -

public class Person{ private List<Account> accounts; .... .... }

我仍然不知道您想要的输出，也不确定您所说的聚合是什么意思？

请解释这些要点，我将修改我的答案。

据我了解的聚合要求，Spring 批处理已经写入 chunks，因此该部分应该已经在没有任何显式聚合的情况下得到处理。

如果我正确理解了您的输入结构和处理需求，您应该使用Spring Batch Partitioning 对您的List<Person>（从XML 读取的记录）进行分区，或者您可以直接在XML 上进行分区（Using split command and SystemCommandTasklet）或您可以将您的 XML 分成多个较小的 xml，并且可以使用 MultiResourcePartitioner

不必一次性启动所有分区步骤，因为您可以使用concurrencyLimit 或TaskExecutor 并行启动一小组分区步骤。

可以制定多种策略，通过多个线程写入单个文件或通过多个线程写入多个文件等。

标准 Spring Batch 方式是在 y 个并行线程中读取 x 个人员（即每个线程有 x 个人员），并行处理每个人，然后以块的形式写入单个目标或多个目标。

将所有人员划分为 N - 分区，在 ItemProcessor 中为每个人调用 REST 服务，并在处理器中准备目标 Person 对象。所需的输出人数将作为chunk-size 设置发送给作者。在步骤@StepScope 中标记从属步骤组件ItemReader、ItemProcessor 和ItemWriter，每个从属步骤将在其自己的线程中运行。

希望对您有所帮助，让我知道您面临哪些具体挑战。