使用 Spring Batch 和 Spring Cloud Data Flow 构建文件轮询/摄取任务

Question

我们计划创建一个新的处理机制，其中包括侦听几个目录e.g: /opt/dir1, /opt/dirN，并且对于在这些目录中创建的每个文档，启动一个例程来处理，将其注册表保存在数据库中（通过 REST 调用现有的 CRUD API）并生成一个协议文件到另一个目录。

出于测试目的，我没有使用任何现代（甚至体面的）框架/方法，只是一个带有 WatchService 实现的常规 SpringBoot 应用程序，它监听这些目录并在创建文件后立即轮询要处理的文件。它可以工作，但很明显，当我转向生产并开始接收要并行处理的数十个文件时，我肯定会对性能产生一些影响，这在我的示例中是不现实的。

经过一些研究和一些同事的提示，我发现 Spring Batch + Spring Cloud Data Flow 是满足我需求的最佳组合。但是，我以前从未处理过批处理或数据流，我有点困惑我应该什么以及如何构建这些块，以便让这个例程以最简单和最高效的方式进行。我有几个关于它的附加值和架构的问题，非常希望听到你的想法！

• 我设法创建并运行了一个基于 on this section of Spring Docs 的示例批处理文件摄取任务。每次在目录中创建文件时如何启动任务？我需要一个 Stream 吗？

• 如果我这样做了，我如何创建一个流应用程序，以编程方式为每个将其路径作为参数传递的新文件启动我的任务？我应该为此目的使用 RabbitMQ 吗？

• 如何为我的任务e.g directories path 保留一些变量？我可以让这些流和任务在 jar 之外的其他地方读取 application.yml 吗？

• 为什么我应该将 Spring Cloud Data Flow 与 Spring Batch 一起使用，而不仅仅是一个批处理应用程序？仅仅因为它跨越了每个文件的并行任务，还是我可以获得任何其他好处？

• 纯粹谈论性能，如果您只考虑顺序处理场景（我每小时只接收 1 个文件左右），那么这个解决方案与我的 WatchService + 普通处理实现相比如何？

另外，如果你们有任何关于如何以编程方式启动任务的指南或示例，我真的很感谢你们！我仍在寻找它，但似乎我做得不对。

感谢您的关注，我们非常感谢您的任何意见！

更新

我设法通过SCDF REST API 启动了我的任务，因此我可以使用 WatchService 通过 Feign 或 XXX 启动新任务来保留我原来的 SpringBoot 应用程序。我仍然知道这远非我应该在这里做的。经过更多研究后，我认为使用文件源和接收器创建流将是我的方式，除非有人有任何其他意见，但我无法将入站通道适配器设置为从多个目录轮询，我不能多个流，因为这个平台应该扩展到我们有数千个参与者（或从中轮询文件的目录）的地步。

Answer 1

这里有一些提示。

我设法根据 Spring Docs 的这一部分创建并运行示例批处理文件摄取任务。每次在目录中创建文件时如何启动任务？我需要一个 Stream 吗？

如果您必须在上游事件（例如：新文件）时自动启动它，是的，您可以通过流来做到这一点（请参阅example）。如果事件来自消息代理，您也可以直接在批处理作业中使用它们（例如：AmqpItemReader）。

如果我这样做，我如何创建一个流应用程序，以编程方式为每个将其路径作为参数传递的新文件启动我的任务？我应该为此使用 RabbitMQ 吗？

希望上面的例子能澄清它。如果您想以编程方式启动任务（不是通过 DSL/REST/UI），您可以使用 1.3 中添加的新 Java DSL 支持来实现。

如何为我的任务保留一些外部变量，例如目录路径？我可以让这些流和任务在 jar 之外的其他地方读取 application.yml 吗？

推荐的方法是使用配置服务器。根据正在编排的平台，您必须为任务及其子任务（包括批处理作业）提供配置服务器凭据。在 Cloud Foundry 中，我们只需将配置服务器服务实例绑定到每个任务，并在运行时自动解析外部化属性。

为什么我应该将 Spring Cloud Data Flow 与 Spring Batch 一起使用，而不仅仅是一个批处理应用程序？仅仅因为它跨越了每个文件的并行任务，还是我可以获得任何其他好处？

作为 Spring Batch Admin 的替代品，SCDF 为 Tasks/Batch-Jobs 提供监控和管理。错误时的执行、步骤、步骤进度和堆栈跟踪将被保留，并可从仪表板进行探索。您也可以直接使用 SCDF 的 REST 端点来检查这些信息。

纯粹谈论性能，如果您只考虑顺序处理场景（我每小时只接收 1 个文件左右），那么这个解决方案与我的 WatchService + 普通处理实现相比如何？

这是特定于实现的。我们没有任何基准可以分享。但是，如果需要性能，您可以探索 Spring Batch 中的 remote-partitioning 支持。您可以使用“n”个工作人员对摄取或数据处理任务进行分区，这样您就可以实现并行性。