你如何“忽略” java.util.concurrent.Future 对象？

Question

你能发现错误吗？这将抛出一个java.lang.OutOfMemoryError。

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class TestTheads {

    public static void main(String[] args) {

        ExecutorService executorService = Executors.newFixedThreadPool(1);
        while(true) {
            executorService.submit(new Runnable() {
                public void run() {
                    try {
                        Thread.sleep(10);
                    } catch (InterruptedException e) {
                    }
                }
            });
        }
    }

}

错误是我调用executorService.submit() 而不是executorService.execute()，因为submit() 返回一个我忽略的Future 对象。使用execute()，这个程序实际上将永远运行。

但是，并不总是拥有execute() 方法的奢侈，例如使用ScheduledExecutorService 时：

public static void main(String[] args) {
    // this will FAIL because I ignore the ScheduledFuture object
    ScheduledExecutorService executorService = Executors.newScheduledThreadPool(2);
    while(true) {
        executorService.scheduleWithFixedDelay(new Runnable() {
            public void run() {
                try {
                    Thread.sleep(10);
                } catch (InterruptedException e) {
                }
            }
        }, 1, 1, TimeUnit.SECONDS);
    }
}

对于不返回任何内容、仅计算的任务应该如何处理？

任何想法将不胜感激！

编辑：ThreadPoolExecutors purge() 看起来很有希望，但它只会清除已取消的任务。

Answer 1

返回的Future 对象被ExecutorService 强引用直到它被执行。（它实际上是一个FutureTask 实例，它委托给你的Runnable。）一旦它执行，它将被垃圾回收，因为调用者没有引用它。也就是说，内存问题与Future的处理无关。

如果您的内存不足，那是因为工作队列有数百万个任务排队。与任何队列一样，除非平均消耗率超过平均生产率，否则队列将被填满。队列的内容会消耗内存。

使用有界队列，这将有效地限制任务排队或获得更多内存。

此代码将“永远”运行：

  ExecutorService executorService = Executors.newFixedThreadPool(1);
  while(true) {
    executorService.submit(new Runnable() {
      public void run() {
        try {
          Thread.sleep(10);
        } catch (InterruptedException e) { }
      }
    });
    Thread.sleep(12);
  }

不同之处不在于对生成的Future 实例的处理，而是任务以​​可以处理的速率排队。

Answer 2

返回的 Futures 并不是你的问题。问题是，对于您提交的每个 Runnable，ExecutorService 将存储它们以供以后处理。每次您使用 Runnable（或 Future）调用 submit 方法时，ExecutorService 都会将该 runnable 推送到工作队列。 Runnable 将坐在那里，直到线程可以从队列中选择该 Runnable（稍后）。如果所有工作线程都忙，那么 ExecutorService 将简单地将可运行对象放入所述队列。

所以你的问题是你只有一个线程试图拉出一个被另一个线程无限添加的队列。它的添加速度要快得多，然后工作线程可以处理每个 Runnable。

编辑：我给出的代码示例确实抛出了一个 RejectedExecutionException，所以如果你要选择，节流机制必须略有不同。

就更好的解决方案而言，就像我在评论中提到的那样；如果您希望以工作线程无法跟上队列的方式填充 ExecutorService，您可以在请求进入时对其进行序列化和反序列化（构建您自己的 ThreadPoolExecutor），但我会确保需要这样一个案例是绝对必要的。

请记住，工作完成后，这些 Future 将被丢弃并收集垃圾。因此，如果您每秒执行一个 Future 并在一秒钟内执行，Future 本身将被删除，您将不会遇到内存问题。但是如果你每秒执行一个 Future 而线程每 3 秒执行一个 Future，那将会绘制并发出。

编辑：我分析了您正在运行的程序的堆，问题正是如此。 ExecutorService 创建的 FutureTask 位于工作队列中，直到工作线程将其取出

Class Name                                                       | Shallow Heap | Retained Heap | Percentage 
------------------------------------------------------------------------------------------------------------
java.util.concurrent.ThreadPoolExecutor @ 0x78513c5a0            |          104 | 2,051,298,872 |     99.99% 
|- java.util.concurrent.LinkedBlockingQueue @ 0x785140598        |           80 | 2,051,298,216 |     99.99% 
|  |- java.util.concurrent.LinkedBlockingQueue$Node @ 0x785142dd8|           32 | 2,051,297,696 |     99.99% 
------------------------------------------------------------------------------------------------------------

堆分析一下，你可以想象有很多LinkedBlockingQueue$Node