1

我目前有一个运行良好的 Spring Integration 流程（参见图表链接）。我想在我当前的配置之上添加 Batch 以允许使用指数回退、断路器模式重试并将作业持久化到数据库以重新启动。

集成流由一个接收 Message<MyObj> 的网关组成，该网关最终被路由到将 Message<MyObj> 转换为 Message<String> 的 Transformer。然后，聚合器接受 Message<String> 并最终释放连接的 Message<String>（使用大小释放策略和带有超时的 MessageGroupStoreReaper）。然后，连接的字符串是使用 SFTP 出站通道适配器上传的文件的有效负载。

我已经搜索、通读文档、查看了大量示例，但我不知道如何将流程的最后一步封装到批处理作业中。如果在上传过程中出现 SFTP 连接问题或引发其他异常，我需要能够重试上传字符串（作为文件的有效负载）。我还希望能够在出现故障时重新启动（使用数据库支持的 JobRepository），所以我认为使用 Retry Advice 是不够的。

请解释并帮助我了解如何将这些部分连接在一起以及使用哪个（作业启动网关、MessageToJobRequest 转换器、ItemReader、ItemWriter??）。我也不确定如何访问每个 Message<String> 并发送到 Job、Step 或 Tasklet 内的 SFTP 通道适配器。

电流：http://i.stack.imgur.com/GnurV.png

4

1 回答 1

1

首先让我们看看我们如何在没有 Batch 的情况下克服您的要求。

  1. <int-sftp:outbound-channel-adapter>有<request-handler-advice-chain>，你可以在其中配置RequestHandlerRetryAdvice和RequestHandlerCircuitBreakerAdvice。

  2. 要实现该restartable选项，您可以将该适配器的输入通道设置为具有message-store.

现在关于批处理。

要从集成流程开始工作，您应该编写一些MessageToJobRequest并<batch-int:job-launching-gateway>在此之后使用。在这里，当然，您可以将您payload的jobParameters.

要将消息从 Job 发送到某个通道（例如发送到 sftp 适配器），您可以使用org.springframework.batch.integration.chunk.ChunkMessageChannelItemWriter.

在这里阅读更多：http: //docs.spring.io/spring-batch/reference/html/springBatchIntegration.html

于 2014-09-10T07:04:49.660 回答