java - 如何正确使用Java Executor？

Question

我在我的多线程应用程序中使用了 Java Executors，但我似乎无法弄清楚何时最好使用以下每种方式:

1.

ExecutorService executor=Executors.newFixedThreadPool(50);
executor.execute(new A_Runner(... some parameter ...));
executor.shutdown();
while (!executor.isTerminated()) { Thread.sleep(100); }

2.

int Page_Count=200;
ExecutorService executor=Executors.newFixedThreadPool(50);
doneSignal=new CountDownLatch(Page_Count);
for (int i=0;i<Page_Count;i++) executor.execute(new A_Runner(doneSignal, ... some parameter ...));
doneSignal.await();
executor.shutdown();
while (!executor.isTerminated()) { Thread.sleep(100); }

3.

int Executor_Count=30;
ThreadPoolExecutor executor=new ThreadPoolExecutor(Executor_Count,Executor_Count*2,1,TimeUnit.SECONDS,new LinkedBlockingQueue());
List<Future<String>> futures=new ArrayList<>(3330);

for (int i=0;i<50;i++) futures.add(executor.submit(new A_Runner(... some parameter ...));
executor.shutdown();
while (!executor.isTerminated()) { executor.awaitTermination(1,TimeUnit.SECONDS); }
for (Future<String> future : futures)
{
    String f=future.get();
    // ...
}

具体来说，在 [2] 中，如果我跳过 doneSignal 会怎样，那么它会像 [1] 那样，那么 doneSignal 有什么用呢？

此外，在 [3] 中，如果我添加一个 doneSignal 会怎么样？或者有可能吗？

我想知道的是:这些方法是否可以互换，或者在某种情况下我应该使用上面的特定类型？

Answer 1

1. ExecutorService

   ExecutorService executor=Executors.newFixedThreadPool(50);

   它简单易用。它隐藏了 ThreadPoolExecutor 的底层细节。

   当 Callable/Runnable 任务的数量很少并且在无界队列中堆积任务不会增加内存和降低系统性能时，更喜欢这个。如果您有 CPU/Memory 限制，请使用具有容量限制的 ThreadPoolExecutor 和 RejectedExecutionHandler 来处理拒绝任务。

2. CountDownLatch

   您已经使用给定计数初始化了 CountDownLatch。通过调用 countDown() 方法减少此计数。我假设您稍后在 Runnable 任务中调用递减。等待此计数达到零的线程可以调用其中一种 await() 方法。调用 await() 会阻塞线程，直到计数达到零。 此类使 java 线程能够等待，直到其他线程集完成其任务。

   用例:

   1. Achieving Maximum Parallelism:有时我们希望同时启动多个线程以达到最大并行度

   2. 等待N个线程执行完毕

   3. 死锁检测。

      看看这个article由 Lokesh Gupta 提供更多详细信息。

3. ThreadPoolExecutor :它提供了更多的控制来微调各种线程池参数。如果您的应用程序受到 Activity Runnable/Callable 任务数量的限制，您应该通过设置最大容量来使用有界队列。一旦队列达到最大容量，您就可以定义 RejectionHandler。 Java提供了四种RejectedExecutionHandler policies .

   1. 在默认的 ThreadPoolExecutor.AbortPolicy 中，处理程序在拒绝时抛出运行时 RejectedExecutionException。

   2. 在ThreadPoolExecutor.CallerRunsPolicy中，调用execute自身的线程运行任务。这提供了一种简单的反馈控制机制，可以减慢提交新任务的速度。

   3. 在 ThreadPoolExecutor.DiscardPolicy 中，无法执行的任务会被简单地丢弃。

   4. 在ThreadPoolExecutor.DiscardOldestPolicy中，如果执行器没有关闭，则丢弃工作队列头部的任务，然后重试执行(可能会再次失败，导致这个要重复。)

      如果你想模拟CountDownLatch行为，你可以使用invokeAll()方法。

4. 您没有引用的另一种机制是 ForkJoinPool

   ForkJoinPool 是在 Java 7 中加入的。 ForkJoinPool 类似于Java ExecutorService 但有一点不同。 ForkJoinPool 实现了任务很容易将他们的工作分解成更小的任务，然后也提交给 ForkJoinPool。当空闲工作线程从繁忙的工作线程队列中窃取任务时，任务窃取发生在 ForkJoinPool 中。

   Java 8 在 ExecutorService 中又引入了一个 API创建工作窃取池。您不必创建 RecursiveTask 和 RecursiveAction，但仍然可以使用 ForkJoinPool。

     public static ExecutorService newWorkStealingPool()
   

   Creates a work-stealing thread pool using all available processors as its target parallelism level.

   默认以CPU核数为参数。

这四种机制互为补充。根据您要控制的粒度级别，您必须选择正确的粒度级别。