java如何处理每10秒钟通过ftp接收的50K文件

Question

我有5万台机器，每台机器都有一个唯一的id。 每10秒，机器将在位于ftp服务器的机器订阅源目录中发送一个文件。并非所有文件都同时接收

计算机将使用其id名称创建文件。 我需要处理所有收到的文件。若文件并没有在短时间内处理，那个么机器将发送新文件，覆盖现有文件，我将丢失现有数据

我的解决方案是

我已经创建了spring boot应用程序，它包含一个每1毫秒执行一次的调度程序，它将重命名接收到的文件，并将其复制到processing dir。当前日期时间将附加到每个文件

我还有一个用ApacheCamel编写的作业，它将每隔500毫秒轮询从processnig位置接收到的文件，并对其进行处理，并在数据库中插入数据。如果收到错误，它将在错误目录中移动文件

文件不大。它只包含一行信息

问题是，如果文件更少，那么它就做得很好。如果文件数量增加，则尽管文件有效，但它仍在错误文件夹中移动

当camel轮询文件时，找到长度为零的文件，然后将该文件复制到错误目录，然后它包含有效数据。有些人认为camel正在轮询未完全复制的文件

有谁知道这个问题的好办法吗

提前谢谢

Answer 1

我以前也遇到过类似的问题，但我使用了一套稍微不同的工具

我建议大家看看Apache Flume——这是一个轻量级java进程。这是我在我的情况下使用的。文档非常不错，所以你应该能够找到自己的方法，但我只是想简单介绍一下，让你开始

水槽有3个主要组件，每个组件都可以通过各种方式进行配置：

1. 来源-负责数据来源的组件
2. 通道缓冲组件
3. 接收器——这将代表数据需要到达的目的地

还有其他可选组件，比如拦截器，它主要用于拦截流和执行基本过滤、转换等

每一个都有各种各样的选项可供选择，但如果没有一个可用的选项适合您的用例，您可以编写自己的组件

现在，针对你的情况——以下是我可以想到的几个选项：

1. 由于您的文件位置几乎需要持续监控，您可能希望使用Flume的Spooling Directory Source，它将持续监视您的机器_feed目录，并在文件到达时将其拾取（您可以选择在文件被覆盖之前自己更改名称）

因此，我们的想法是拿起文件并将其交给处理目录，然后使用Apache Camel继续处理，就像您已经在做的那样

1. 另一个选项是（这是我建议考虑的）在一个Flume代理中完成所有操作

水槽设置可能如下所示：

• Spooling Directory Source
• 其中一个interceptors（可选：在将数据插入数据库之前进行处理。如果没有合适的可用选项，您甚至可以编写自己的自定义拦截器）
• 其中一个channels（Memory channel-可能是…）
• 最后是sinks中的一个（在您的情况下，这可能只需要是一个自定义接收器，以便在数据库中登录数据）

如果您确实需要编写一个自定义组件（拦截器或接收器），您可以查看其中一个默认组件的源代码以供参考。这是link to the source code repository

我知道我提出了一个全新的工具，这与我的想法略有不同，但这对我来说非常有效，因为该工具是一个重量非常轻的工具，具有相当简单的设置和配置

我希望这有帮助