在单独的线程中处理文件写入的正确方法

Question

我希望多次写入文件（100k+），写入将通过不稳定的网络进行。因此，为此，我正在考虑使用 Java ExecutorService 来帮助生成线程，但我不太确定哪种设置组合会正确地发生以下情况：

1. 一次只允许 1 次写入（当然顺序很重要）
2. 让写操作有足够的时间来进行每次写操作（比如 5 秒），此时就可以放弃了
3. 如果写入速度很慢，让 Executor 将写入收集到队列中并等待。
4. 在线程队列为空之前，不允许整个程序退出。
5. 按作者分隔线程。即，如果相同的写入器出现在此函数中，则将其放入自己的队列中。如果有不同的写入器指针进入，则给它自己的队列（无需将不同的写入器放在同一个队列中）。

我相信这可以通过结合执行器功能以及主程序对象上的.wait() 和.notify() 命令来完成。但是，我只是不确定如何精确地使用 executor API 来完成这项工作。

这是我得到的：

private void writeToFileInSeperateThread(final PrintWriter writer, final String text) {
  ExecutorService executor = Executors.newSingleThreadExecutor();
  try {
    executor.submit(new Thread(new Runnable() {
      public void run() {
        writer.println(text);
      }
    })).get(5L, TimeUnit.SECONDS);
  } catch (Exception e) {
    e.printStackTrace();
  }
  executor.shutdown();
}

该方法将在单个进程中被调用 100k+ 次，所以我不确定是否应该每次都创建一个新的 ExcutorService 实例，还是使用同一个实例？ （在尝试使用同一个指令时，我不断收到我认为与 .newSingleThreadExecutor() 指令有关的异常。

希望保持 Java 5 兼容，但 Java 6 没问题。在 Windows XP/7 上运行。

更新： 这似乎在初始测试中起到了作用：

  private class WriterStringPair {
    public final PrintWriter writer;
    public final String text;

    public WriterStringPair(PrintWriter writer, String text) {
      this.writer = writer;
      this.text = text;
    }
  }

  private void writeTextInSeperateThread(Writer writer, String text) {
    try {
      textQueue.offer(new WriterStringPair(writer, text), 300L, TimeUnit.SECONDS);
    } catch (InterruptedException e) {
      errOut.println(e);
      e.printStackTrace();
    }
  }

  final BlockingQueue<WriterStringPair> textQueue = new ArrayBlockingQueue<WriterStringPair>(500);

  private void setWritingThread() {
    new Thread((new Runnable() {
      public void run() {
        WriterStringPair q;
        while (!shutdown && !Thread.currentThread().isInterrupted()) {
          try {
            q = textQueue.poll(1L, TimeUnit.SECONDS);
            if (q != null) {
              q.writer.write(q.text + "\n");
              q.writer.flush();
            }
          } catch (Exception e) {
            e.printStackTrace();
          }
        }
      }
    })).start();
  }

Answer 1

如果不了解有关您在“不稳定”网络上编写文件的更多详细信息及其含义，我们很难提供具体信息。但这里有一些事情需要考虑。

我会计算出有多少并发写入者可以在此处为您提供最佳性能 - 或者在目标上提供最可靠的输出。然后，您应该启动固定数量的这些写入程序，每个写入程序都从一个共享的BlockingQueue 使用（或者如果重要的话，每个写入程序一个队列）。您应该很快超过您的 IO 或网络带宽，因此从 5 个左右的写入器开始并根据需要增加或减少应该可以工作。

public void run() {
   writer.println(text);
}

是的，就每行工作而言，您不想做这种事情。最好将String text 放入BlockingQueue<String>，然后让您的编写器Runnable 类在ExecutorService 中运行，从该队列中出列，并且仅在队列为空或设置shutdown 布尔值时停止.

正如 Peter 所提到的，您需要小心使用排队的文本字符串填充内存。如果输入文本很大，您应该将BlockingQueue 的限制设置为几百左右。

我不确定是否应该每次都创建一个新的 ExecutorService 实例，还是使用同一个实例？

当然，您应该有一个服务并且不一遍又一遍地创建一个。

我相信这可以通过结合执行器功能以及主程序对象上的 .wait() 和 .notify() 命令来完成。

如果你写得正确，你不应该需要使用等待和通知。我有一个volatile boolean shutdown = false，你们所有的作家都会看。他们每个人都通过查看关机从文本队列中出列。比如：

while (!shutdown && !Thread.currentThread().isInterrupgted()) {
    String text = textQueue.poll(1, TimeUnit.SECONDS);
    if (text != null) {
        // write the text
    }
}

如果写入失败或发生什么，您可以重试或任何必要的操作。

Answer 2

几个问题

• println 不会告诉您是否存在 IOException，因此如果您想要一些防止错误的保护，这将无济于事。
• 为每一行启动一个 ExecutorService 非常慢，比提交任务慢很多。
• 创建大量任务不仅会很慢，而且如果这样做的话，可能会耗尽你所有的内存。
• 您向 ExecutorService 提交 Runnable，而不是 Threads
• shutdown 不会停止线程，例如，如果它在写入时阻塞。这可能会导致许多线程试图同时写入。

我建议将数据保存到 JMS 等本地系统或数据库或文件（例如 Java-Chronicle），并在可用时将数据复制到 NFS。

这是假设您无法修复 NFS，因此它不是片状的。