Java并发编程之Executors与Thread Pool

怼烎@ 2022-06-02 05:19 219阅读 0赞

![这里写图片描述][SouthEast]  
上一章，我们谈到了JDK1.5引入的多线程编程框架**Executors**，其通过工作线程的方式对外提供线程池服务。一个**ExecutorService**实例就可以表示一个线程池，通过**Executors**，你可以创建以下集中类型线程池：

*  Single Thread Executor：仅包含一个线程的线程池，所有提交的任务顺序执行，创建方法：
    
    > **Executors.newSingleThreadExecutor()**
 *  Cached Thread Pool: 包含不等数目线程的线程池，JVM会根据实际执行情况创建尽可能多、满足并行场景的线程数，新任务会重复使用空闲线程，如果线程60秒内没有使用，则线程终止，并且从线程池中剥离。
    
    > **Executors.newCachedThreadPool()**
 *  Fixed Thread Pool：包含固定数目线程的线程池，如果没有空闲线程，新提交任务将会放到等待队列中。
    
    > **Executors.newFixedThreadPool()**
 *  Scheduled Thread Pool：用于任务调度的线程池。
    
    > **Executors.newScheduledThreadPool()**
 *  Single Thread Scheduled Pool：用于任务调度的线程池，但线程池仅包含一个线程。
    
    > **Executors.newSingleThreadScheduledExecutor()**

线程池创建好之后，你可以通过submit方法向线程池提交任务并返回一个Future对象，如果是Callable类型任务，该对象表示的是任务的未来执行状态。如果是Runnable类型任务，当任务执行完毕，Future对象返回为空。

private final class StringTask extends Callable<String>{ 
        public String call(){
            //Long operationsreturn"Run";
        }
    }
    
    ExecutorService pool = Executors.newFixedThreadPool(4);
    for(int i =0; i <10; i++){   
        pool.submit(new StringTask());
    }

当所有任务执行完毕之后，调用**pool.shutdown()**方法中止所有线程，否则未中止的线程可能导致JVM无法正常关闭的风险。你也可以通过调用shutdownNow()方法立即关闭线程池，一旦执行shutdownNow()命令，那么正在执行任务的线程会被立即中止，而尚未启动的任务则将不再被启动。

但是，在上面的示例代码中，你是无法获取到任务执行结果的，如果想要获取任务的执行结果，则通过下面的方法：

ExecutorService pool = Executors.newFixedThreadPool(4);
    List<Future<String>> futures =new ArrayList<Future<String>>(10);
    for (int i =0; i <10; i++){
        futures.add(pool.submit(new StringTask()));
    }
    for (Future<String> future : futures){
        String result = future.get();
        //Compute the result
    } 
    pool.shutdown();

这段代码略显复杂，而且还有一个小问题：我们知道，Future.get()方法是同步方法，假设我们提交的第一个任务执行时间过长，那么我们需要一直等到第一个方法执行完毕才能获取到别的任务的执行结果，这在实际的应用当中就显得不是特别友好。

因此，为了使得我们的“等待”更显优雅，Java为我们提供另一中解决方案**CompletionService**，因此，上面的方法我们可以改写如下：

ExecutorService threadPool = Executors.newFixedThreadPool(4);
    CompletionService<String> pool =new ExecutorCompletionService<String>(threadPool);
    for (int i =0; i <10; i++){ pool.submit(new StringTask()); }
    for (int i =0; i <10; i++){ String result = pool.take().get(); //Compute the result } 
    threadPool.shutdown();

**CompletionService**帮我们维护了线程池中任务完成的顺序，无需我们在额外的通过集合来维护任务提交列表，这样我们就可以按照任务完成的顺序来获取任务执行结果。忽略具体实现细节，我们可以用简单的程式化语言来表示**ExecutorService**与**CompletionService**之间的区别：

> **ExecutorService = incoming queue + worker threads**
> 
>   
> **CompletionService = incoming queue + worker threads + output queue**

### 示例代码 ###

public class CompletionServiceTest {
    
        static class CalcResult {
            long result;
    
            CalcResult(long l) {
                result = l;
            }
        }
    
        static class CallableTask implements Callable<CalcResult> {
            String taskName;
            long input1;
            int input2;
    
            CallableTask(String name, long v1, int v2) {
                taskName = name;
                input1 = v1;
                input2 = v2;
            }
    
            public CalcResult call() throws Exception {
                System.out.println("Task " + taskName + " Started -----");
                for (int i = 0; i < input2; i++) {
                    try {
                        Thread.sleep(200);
                    } catch (InterruptedException e) {
                        System.out.println("Task " + taskName + " Interrupted !! ");
                        e.printStackTrace();
                    }
                    input1 += i;
                }
                System.out.println("Task " + taskName + " Completed @@@@@@");
                return new CalcResult(input1);
            }
    
        }
    
    
        public static void main(String[] args) {
            ExecutorService taskExecutor = Executors.newFixedThreadPool(3);
            CompletionService<CalcResult> taskCompletionService =
                    new ExecutorCompletionService<>(taskExecutor);
    
            int submittedTasks = 5;
            for (int i = 0; i < submittedTasks; i++) {
                taskCompletionService.submit(new CallableTask(
                        String.valueOf(i),
                        (i * 10),
                        ((i * 10) + 10)
                ));
                System.out.println("Task " + String.valueOf(i) + " submitted");
            }
            for (int tasksHandled = 0; tasksHandled < submittedTasks; tasksHandled++) {
                try {
                    System.out.println("trying to take from Completion service");
                    Future<CalcResult> result = taskCompletionService.take();
                    System.out.println("result for a task available in queue.Trying to get()");
                    // above call blocks till at least one task is completed and results available for it
                    // but we don't have to worry which one
    
                    // process the result here by doing result.get()
                    CalcResult l = result.get();
                    System.out.println("Task " + String.valueOf(tasksHandled) + " Completed - results obtained : " + String.valueOf(l.result));
    
                } catch (InterruptedException e) {
                    // Something went wrong with a task submitted
                    System.out.println("Error Interrupted exception");
                    e.printStackTrace();
                } catch (ExecutionException e) {
                    // Something went wrong with the result
                    e.printStackTrace();
                    System.out.println("Error get() threw exception");
                }
            }
    
            taskExecutor.shutdown();
        }
    }

[https://dzone.com/articles/java-concurrency-%E2%80%93-part-7][https_dzone.com_articles_java-concurrency-_E2_80_93-part-7]  
[https://stackoverflow.com/questions/4912228/when-should-i-use-a-completionservice-over-an-executorservice][https_stackoverflow.com_questions_4912228_when-should-i-use-a-completionservice-over-an-executorservice]

[SouthEast]: /images/20220602/55fc47193708430da0611ca14aa89cfc.png
[https_dzone.com_articles_java-concurrency-_E2_80_93-part-7]: https://dzone.com/articles/java-concurrency-%E2%80%93-part-7
[https_stackoverflow.com_questions_4912228_when-should-i-use-a-completionservice-over-an-executorservice]: https://stackoverflow.com/questions/4912228/when-should-i-use-a-completionservice-over-an-executorservice