Spring 批量使用或如何在 Job 中启动 Jobs

Question

TL;DR：应该如何使用 Spring Batch Job 创建 Spring Batch Jobs？ 事务边界似乎是问题所在。这似乎是一个 经典问题，但又来了：

我有以下用例：我需要轮询 FTP 服务器并存储找到 XML 文件作为数据库中的 blob。 XML 有 0...N 个感兴趣的条目 I 需要发送到外部 Web Service 并存储 回复。响应可以是不可重试或可重试的，我需要 存储每个请求及其响应以供审核。

域/JPA 模型如下：Batch（包含 XML blob）包含 0-N 个 BatchRow 对象。 BatchRow 包含要发送到 Web 的数据 服务，它还包含 1...N BatchRowHistory 持有状态的对象 有关 Web 服务调用的信息。

我被要求使用 Spring Batch（Spring Integration 自从这种整合案例以来，可能还有其他可能性）。现在 我一直在努力使用不同的方法，我发现这项任务很多 更复杂，因此更困难，恕我直言。

我已将任务拆分为以下工作：

工作1：

• Step11：获取文件并以blob形式存储到数据库中。

• Step12：将 XML 拆分为条目并将这些条目存储到 db。

• Step13：创建 Job2 并为存储在其中的每个条目启动它 步骤 12。标记 Job2 在域模型中创建标志 条目数据库。

工作2：

• Step21：为每个条目调用Web服务并将结果存储到db。重试并 跳过逻辑就在这里。 Job2 类型可能需要手动重启等。

这个结构背后的逻辑是Job1定期运行 预定（每分钟一次左右）。 Job2 只要有就运行 那些作业，他们要么成功，要么重试限制已到 他们失败了。领域模型基本上只存储结果和 Spring Batch 负责运行该节目。手动重新启动 等可以通过 Spring Batch Admin 处理（至少我希望如此）。还 Job2 在 JobParameters 映射中有 BatchRow 的 id，因此可以 在 Spring Batch Admin 中查看。

问题 1：这种工作结构有意义吗？创造新的 db中每一行的Spring Batch Jobs，它似乎打败了 目的并在某种程度上重新发明轮子？

问题 2：如何在 Step13 中创建那些 Job2 条目？

我在事务和 JobRepository 方面遇到了第一个问题，但成功了 使用以下设置启动几个作业：

<batch:step id="Step13" parent="stepParent">
 <batch:tasklet>
   <batch:transaction-attributes propagation="NEVER"/>
   <batch:chunk reader="rowsWithoutJobReader" processor="batchJobCreator" writer="itemWriter"
                commit-interval="10" />
 </batch:tasklet>
</batch:step>

<bean id="stepParent" class="org.springframework.batch.core.step.item.FaultTolerantStepFactoryBean" abstract="true"/>

请注意，commit-interval="10" 表示这最多可以创建 10 个 目前的工作就是这样......因为batchJobCreator调用 JobLauncher.run 方法，它很流畅，但 itemWriter 不能 使用更新的信息将 BatchRows 写回数据库（布尔值 jobCreated 标志打开）。明显的原因是传播。在事务属性中永远不会，但没有它我无法使用 jobLauncher 创建工作。

因为更新没有传递到数据库，我再次得到相同的 BatchRows 并且 他们把日志弄得乱七八糟：

org.springframework.batch.retry.RetryException: Non-skippable exception in recoverer while processing; nested exception is org.springframework.batch.core.repository.JobExecutionAlreadyRunningException: A job execution for this job is already running: JobInstance: id=1, version=0, JobParameters=[{batchRowId=71}], Job=[foo.bar]
        at org.springframework.batch.core.step.item.FaultTolerantChunkProcessor$2.recover(FaultTolerantChunkProcessor.java:278)
        at org.springframework.batch.retry.support.RetryTemplate.handleRetryExhausted(RetryTemplate.java:420)
        at org.springframework.batch.retry.support.RetryTemplate.doExecute(RetryTemplate.java:289)
        at org.springframework.batch.retry.support.RetryTemplate.execute(RetryTemplate.java:187)
        at org.springframework.batch.core.step.item.BatchRetryTemplate.execute(BatchRetryTemplate.java:215)
        at org.springframework.batch.core.step.item.FaultTolerantChunkProcessor.transform(FaultTolerantChunkProcessor.java:287)
        at org.springframework.batch.core.step.item.SimpleChunkProcessor.process(SimpleChunkProcessor.java:190)
        at org.springframework.batch.core.step.item.ChunkOrientedTasklet.execute(ChunkOrientedTasklet.java:74)
        at org.springframework.batch.core.step.tasklet.TaskletStep$ChunkTransactionCallback.doInTransaction(TaskletStep.java:386)
        at org.springframework.transaction.support.TransactionTemplate.execute(TransactionTemplate.java:130)
        at org.springframework.batch.core.step.tasklet.TaskletStep$2.doInChunkContext(TaskletStep.java:264)
        at org.springframework.batch.core.scope.context.StepContextRepeatCallback.doInIteration(StepContextRepeatCallback.java:76)
        at org.springframework.batch.repeat.support.RepeatTemplate.getNextResult(RepeatTemplate.java:367)
        at org.springframework.batch.repeat.support.RepeatTemplate.executeInternal(RepeatTemplate.java:214)
        at org.springframework.batch.repeat.support.RepeatTemplate.iterate(RepeatTemplate.java:143)
        at org.springframework.batch.core.step.tasklet.TaskletStep.doExecute(TaskletStep.java:250)
        at org.springframework.batch.core.step.AbstractStep.execute(AbstractStep.java:195)
        at org.springframework.batch.core.job.SimpleStepHandler.handleStep(SimpleStepHandler.java:135)
        at org.springframework.batch.core.job.flow.JobFlowExecutor.executeStep(JobFlowExecutor.java:61)
        at org.springframework.batch.core.job.flow.support.state.StepState.handle(StepState.java:60)
        at org.springframework.batch.core.job.flow.support.SimpleFlow.resume(SimpleFlow.java:144)
        at org.springframework.batch.core.job.flow.support.SimpleFlow.start(SimpleFlow.java:124)
        at org.springframework.batch.core.job.flow.FlowJob.doExecute(FlowJob.java:135)
        at org.springframework.batch.core.job.AbstractJob.execute(AbstractJob.java:293)
        at org.springframework.batch.core.launch.support.SimpleJobLauncher$1.run(SimpleJobLauncher.java:120)
        at java.lang.Thread.run(Thread.java:680)

这意味着已经在 Spring Batch 中创建了作业并且它 尝试在以后执行 Step13 时再次创建这些文件。一世 可以规避在 Job2/Step21 但我觉得有点笨拙和错误。

问题 3：我有更多的领域对象驱动方法；我有春天 使用非常精细的 JPQL 查询扫描域表的批处理作业 和 JPAItemReaders。这种方法的问题在于 不要使用 Spring Batch 的更精细的功能。历史和重试逻辑是 问题。我需要将重试逻辑编码到 JPQL 查询中 直接（例如，如果 BatchRow 有超过 3 个 BatchRowHistory 元素失败，需要手动重新检查）。 我应该 咬紧牙关继续这种方法，而不是试图 为每个 Web 服务调用创建单独的 Spring Batch Job？

软件信息（如果需要）：Spring Batch 2.1.9、Hibernate 4.1.2、Spring 3.1.2，Java 6。

提前谢谢你，很抱歉，蒂莫，长篇大论

编辑 1： 我认为我需要创造新工作的原因是：

• 读取器返回 null 或抛出异常时循环

• 事务开始

• 读取器 - 处理器 - 整个 N 行的写入器循环

• 批量大小为 N 的事务结束

每个失败的条目都是问题；我想要手动重启 执行（作业是唯一可以在 Spring 中重新启动的 批处理管理员，对吗？）批处理中的每一行，以便我可以使用 Spring Batch Admin 查看失败的作业（及其作业参数 其中包含来自域 db 的行 id）并重新启动这些等。我该如何 在不产生作业和存储的情况下完成这种行为 域数据库的历史记录？

Answer 1

好吧，我讨厌回答问题……但我需要知道一些事情吗？

1) 如果您的输入文件是 XML，为什么不在它们上使用 StaxEventItemReader 并简单地将您的条目保留在步骤 1 中？

2) 从一个步骤开始第二个工作！！！！我什至不知道它是否应该工作......但IMO ..它闻起来;-)

为什么不直接定义另一个步骤，使用 JdbcCursorItemReader 读取条目并调用 ItemProcessor 中的 Web 服务，然后将结果写入数据库？

也许我不明白您为每次调用 Web 服务创建不同工作的要求！！！

我做了与您的用例类似的事情，并且是使用这种情况完成的：

工作 1： 第1步：读取xml，处理pojo->domain obj，将domain obj写入DB

工作 2： 第 1 步：从 db 中读取 obj，进程 = 调用 WS，在 DB 中写入响应

这很简单，效果很好（包括可重启和跳过功能）

希望对你有帮助

问候