写入未完成时如何检测我正在从文件中读取？

Question

我们有一个多线程程序，它执行以下操作：

thread_1 是硬盘监听器，用于检测创建的​​新文件。我们在 Java 7 中使用 WatchService api。当另一个程序创建新文件时，thread_1 检测并获取它并将其放入 PriorityBlockingQueue 前：

priorityBlockingQueue.add(FileObject)

FileObjComparator 是一个自定义对象实现比较器。它是按创建时间和FileObject 中的fileCreatedTime 字段排序的，我在检测到这个文件时从系统时间得到：

 public int compare(FileObject o1, FileObject o2) {
        return o1.getFileCreatedTime().compareTo(o2.getFileCreatedTime());
    }

priorityBlockingQueue 初始化为：

DataFileQueue.priorityBlockingQueue = new PriorityBlockingQueue<FileObject>(100000, new FileObjComparator());

并且Thread_2 将在此priorityBlockingQueue 中的最后一个文件旁边处理此文件

if(priorityBlockingQueue.size) > 1)
   process(priorityBlockingQueue.poll());

2 个线程并行运行，但是当我处理大量大文件时，有时Thread_2 会在一个正在写入的文件处理该文件。我检测到这一点是因为重新检查内容文件和处理结果。

此程序在 Centos 6.2 上运行，此硬盘分区以异步模式挂载。感谢您的帮助。

Answer 1

您的 Comparator 应按上次修改时间排序，而不是按创建时间排序。例如，我不知道您如何知道以 A、B 顺序打开的两个文件将完全以相同的顺序写入，除非您肯定知道文件生成是严格按顺序进行的。你还没说呢。

Answer 2

编辑更详细的答案。

问题是……

你写的：

在检测到这个文件时，我从系统时间得到的 FileObject 中的创建时间和 fileCreatedTime 字段排序： ....

thread_1 是硬盘的监听器，用于检测创建的​​新文件。我们在 Java 7 中使用 WatchService api。当一个新文件由另一个程序创建时。 ... thread_1 检测并获取它 并将其放入 PriorityBlockingQueue ex：

• 创建时间和“文件写入完成时间”可能有很大不同。 （取决于文件大小）。

例如：

打开文件管理器。开始下载大约 60 mb 的文件。注意创建时间。大约 3 分钟后，看看最后的时间。

要检测新文件，查看创建时间是 错误的时刻“将其放入 PriorityBlockingQueue ex：”

thraed_1 必须等到文件写入完成。然后他可以把它放到“a PriorityBlockingQueue ex：”

如何检测文件的写入是否完成？

3 个不太复杂的选项

• a.) 比较文件已创建和文件就绪时间。 或
• b.) 观察文件的大小正在稳步增长。如果 文件完成它停止增长。 或
• c.) 尝试将其移至临时文件夹。

你喜欢什么？

我更喜欢解决方案 c。

无法移动为写入而打开的文件。 第 3 方程序 关闭文件后可以移动它。

必要的步骤。

• thread_1 正在监视由第三方程序创建的文件。
• thread_1 试图将其移动到 xyztmp 文件夹（每 10 或 20 或 ... 秒）。
• thread_1 在 xyztmp 文件夹并将其放入 PriorityBlockingQueue ex.

---

解决方案 b。更复杂。

thread_1 将传入的文件名和大小放在一个控制数组中进行比较 3-5 次。（每 5 秒或更长时间）。

数组

(filenamexyz.dat, size1, size2, size3, ...).
(filenameabc.dat, size1, size2, size3, ...).
(filenamefgh.dat, size1, size2, size3, ...).
....

如果每 5 个比较大小由名称标识的文件相同，则第三方程序已完成对该文件的写入。

现在可以将其放入 PriorityBlockingQueue 例如：

让我们一步一步来看看

我们假设 thread_2 在 list.size 为 2 时启动！

• 第三方程序开始一一写入文件。
• 第三方程序开始写入 FILE_1。
• thread_1 检测到创建的FILE_1，将其放入列表中。
• 第 3 方程序完成了 FILE_1 的写入。
• 第 3 方程序开始写入 FILE_2。
• thread_1 检测到创建的 FILE_2，将其放入列表中。
• if(priorityBlockingQueue.size) > 1) TRUE
• thread_2 开始读取和处理列表 FILE_1 中的第一个文件。

---

• 第 3 方程序完成了 FILE_2 的写入。
• 第 3 方程序开始写入 FILE_3。
• thread_1检测到创建的FILE_3，放入列表中。
• thread_2 已完成处理 FILE_1。
• thread_2 从列表 FILE_2 中的下一个文件开始。

---

• 第 3 方程序已完成写入 FILE_3。
• 第 3 方程序开始写入 FILE_4。
• thread_1 检测到创建的 FILE_4，将其放入列表中。
• thread_2 已完成处理 FILE_2。

• thread_2 从列表 FILE_3 中的下一个文件开始。

  现在麻烦开始了

---

• 第三方程序完成了 FILE_4 的写入。
• 第三方程序开始写入 FILE_5。 （FILE_5 大于 FILE_4）。
• thread_1 检测到创建的FILE_5，将其放入列表中。
• thread_2 已完成处理 FILE_3。
• thread_2 从 FILE_4 列表中的下一个文件开始。
• thread_2 已完成处理 FILE_4。
• thread_2 从列表 FILE_5 中的下一个文件开始。
• thread_2 已完成处理 FILE_5。
• 第 3 方程序已完成写入 FILE_5。

如果第 3 方程序写入的文件较大，需要更多时间写入，并且 thread_2 已完成读取较小的 FILE_4。

thread_2 将下一个文件从列表中取出 - FILE_5，无论该文件是否已准备好读取。

FILE_5 是第 3 方程序仍在写入的文件。 FILE_5 是thread_2 正在读取和处理的文件。 thread_2 读取的字节数只是第三方程序此时写入的字节数。

Answer 3

如果您真的在处理倒数第二个文件，那么我很惊讶它的大小正在增长，除非多个进程或线程正在生成输入文件。确保创建文件的其他进程在写入下一个文件之前刷新并关闭每个文件。

• 您可以按块读取文件，然后返回一段时间以查看是否有任何其他数据添加到文件中，并在当时使用RandomAccessFile 处理它。如果您正在逐行阅读文件，那么不幸的是，您需要自己进行分页。如果文件是基于行的，那么您应该确保行终止字符关闭文件。

• 您可以尝试的另一件事是稍微延迟文件的处理，以让文件系统刷新其缓冲区。丑陋且不可靠，但可能是必要的。

• 如果您可以调整输出过程，那么您可以用魔术字符串结束文件，然后在看到魔术字符串之前不处理文件。

• 您可以让进程写入文件，将文件的大小写入带有“.size”扩展名（或其他内容）的单独文件中。大小文件将帮助您验证您读取的字符数是否正确。

• 如果您在 ~unix 系统上运行，另一件要尝试的事情是在开始从文件读取以同步文件系统之前Runtime.exec("/bin/sync");。问题是对此的支持高度依赖于操作系统。它也可以成为真正的性能杀手。他是我 Mac 上的手册页：

  可以调用同步实用程序来确保所有磁盘写入都已完成

Answer 4

您可以尝试使用信号量来组织对每个文件的访问，这样就不会获取任何文件 一次由多个线程写入。我认为每个文件对象都应该有它的 自己的信号量，每个线程都应该在写入之前尝试获取信号量 文件。