在 Java 中使用 ThreadPoolExecutor 实现动态文件传输控制器

Question

我最近不得不实现一个控制器，将文件从 A 传输到 B。 大约有 8000 个文件，每个文件大小为 1-2 mb。

• 如果一个文件传输成功，创建另一个线程。 （目前增加 corePoolSize +1）
• 如果一个文件传输失败，关闭一个线程重试传输。（当前增加corePoolSize -1）
• 如果一个文件传输失败，一定时间内不要创建另一个线程。

这背后的想法是在不知道主机限制的情况下获得最大可能的连接数/最佳传输速度。

现在我的问题是，ThreadPoolExecutor 是实现此行为的最佳方式还是有更好的方式？

//Code simplified
//add all files to callables with type Future<Boolean>
while (true) {

   // entry = get the first result that's done.

   if (entry.getValue().get() == Boolean.TRUE) {
       results.remove(entry.getKey());
       if (results.size() > threadPool.getCorePoolSize()) {

           if (System.currentTimeMillis() >= nextAttempt)
               resizeThreadPool(+1);
       }
   } else {
       resizeThreadPool(-1);
       nextAttempt = System.currentTimeMillis() + someTimeinMs;
       entry.setValue(threadPool.submit(entry.getKey()));
   }
   if (results.isEmpty())
      return true;
}

编辑：作为参数给出了最小和最大线程数。

Answer 1

这是一个非常有趣的问题，因为每次文件传输的性能是可变的（读/写可能发生在本地磁盘或某个远程位置，并且并非所有磁盘/网络位置始终具有相同的有效带宽可用）。但是，我不确定使用传输成功/失败是否是决定何时增加或减少线程数的好指标。我怀疑由于使用线程过度配置读/写位置而不会发生故障，它们可能只会减慢速度，与一次复制一个文件相比会产生一些额外的开销。在这种情况下，您只会产生更多线程，直到您用完要复制的文件或内存不足，以先到者为准。

也就是说，我认为您最好从不同的角度解决问题。以下限制是我最关心的：

1. 您必须在某处从读取每个文件。
2. 您必须将每个文件写入某处。
3. 读/写位置是任意的。
4. 您不能在写入之前将每个文件都放入 RAM。
5. 您无法生成无限数量的线程（如果尝试，您将耗尽内存）。

考虑到这些限制，我将维护几个线程池，一个用于读取，一个用于写入，每个池轮询一个读/写请求队列。在最佳情况下，每发现一个新的读/写位置都会产生一个新线程，但上限为一定大小，因此您不会超过系统内存限制（或某个预定的可接受数量）。您还需要设置一些缓冲区大小，如果文件大于该缓冲区大小，您将读取前 N 个字节，然后在读取下一个 N 个字节之前将这些字节传递到写入队列。通过这种方式，您可以在完整读取大文件之前将其写入目的地，从而节省在写入之前读取整个文件所需的时间和内存。最后，您可以限制写入队列的大小，以防止您的程序在任何给定时间读取的数据超出程序的 RAM 分配容量。