`

java并发编程-Executor框架

    博客分类：
  • Java
 
阅读更多

Executor框架是指java 5中引入的一系列并发库中与executor相关的一些功能类，其中包括线程池，Executor，Executors，ExecutorService，CompletionService，Future，Callable等。他们的关系为：


 

并发编程的一种编程方式是把任务拆分为一些列的小任务，即Runnable，然后在提交给一个Executor执行，Executor.execute(Runnalbe) 。Executor在执行时使用内部的线程池完成操作。

一、创建线程池

Executors类，提供了一系列工厂方法用于创先线程池，返回的线程池都实现了ExecutorService接口。

public static ExecutorService newFixedThreadPool(int nThreads)

创建固定数目线程的线程池。

public static ExecutorService newCachedThreadPool()

创建一个可缓存的线程池，调用execute 将重用以前构造的线程（如果线程可用）。如果现有线程没有可用的，则创建一个新线程并添加到池中。终止并从缓存中移除那些已有 60 秒钟未被使用的线程。

public static ExecutorService newSingleThreadExecutor()

创建一个单线程化的Executor。

public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize)

创建一个支持定时及周期性的任务执行的线程池，多数情况下可用来替代Timer类。

Java代码 复制代码 收藏代码
  1. Executor executor = Executors.newFixedThreadPool(10);   
  2. Runnable task = new Runnable() {   
  3.     @Override  
  4.     public void run() {   
  5.         System.out.println("task over");   
  6.     }   
  7. };   
  8. executor.execute(task);   
  9.   
  10. executor = Executors.newScheduledThreadPool(10);   
  11. ScheduledExecutorService scheduler = (ScheduledExecutorService) executor;   
  12. scheduler.scheduleAtFixedRate(task, 10, 10, TimeUnit.SECONDS);  
Java代码  收藏代码
  1. Executor executor = Executors.newFixedThreadPool(10);  
  2. Runnable task = new Runnable() {  
  3.     @Override  
  4.     public void run() {  
  5.         System.out.println("task over");  
  6.     }  
  7. };  
  8. executor.execute(task);  
  9.   
  10. executor = Executors.newScheduledThreadPool(10);  
  11. ScheduledExecutorService scheduler = (ScheduledExecutorService) executor;  
  12. scheduler.scheduleAtFixedRate(task, 10, 10, TimeUnit.SECONDS);  

 二、ExecutorService与生命周期

ExecutorService扩展了Executor并添加了一些生命周期管理的方法。一个Executor的生命周期有三种状态，运行，关闭 ，终止 。Executor创建时处于运行状态。当调用ExecutorService.shutdown()后，处于关闭状态，isShutdown()方法返回true。这时，不应该再想Executor中添加任务，所有已添加的任务执行完毕后，Executor处于终止状态，isTerminated()返回true。

如果Executor处于关闭状态，往Executor提交任务会抛出unchecked exception RejectedExecutionException。

Java代码 复制代码 收藏代码
  1. ExecutorService executorService = (ExecutorService) executor;   
  2. while (!executorService.isShutdown()) {   
  3.     try {   
  4.         executorService.execute(task);   
  5.     } catch (RejectedExecutionException ignored) {   
  6.            
  7.     }   
  8. }   
  9. executorService.shutdown();  
Java代码  收藏代码
  1. ExecutorService executorService = (ExecutorService) executor;  
  2. while (!executorService.isShutdown()) {  
  3.     try {  
  4.         executorService.execute(task);  
  5.     } catch (RejectedExecutionException ignored) {  
  6.           
  7.     }  
  8. }  
  9. executorService.shutdown();  

 三、使用Callable，Future返回结果

Future<V>代表一个异步执行的操作，通过get()方法可以获得操作的结果，如果异步操作还没有完成，则，get()会使当前线程阻塞。FutureTask<V>实现了Future<V>和Runable<V>。Callable代表一个有返回值得操作。

Java代码 复制代码 收藏代码
  1. Callable<Integer> func = new Callable<Integer>(){   
  2.     public Integer call() throws Exception {   
  3.         System.out.println("inside callable");   
  4.         Thread.sleep(1000);   
  5.         return new Integer(8);   
  6.     }          
  7. };         
  8. FutureTask<Integer> futureTask  = new FutureTask<Integer>(func);   
  9. Thread newThread = new Thread(futureTask);   
  10. newThread.start();   
  11.   
  12. try {   
  13.     System.out.println("blocking here");   
  14.     Integer result = futureTask.get();   
  15.     System.out.println(result);   
  16. } catch (InterruptedException ignored) {   
  17. } catch (ExecutionException ignored) {   
  18. }  
Java代码  收藏代码
  1. Callable<Integer> func = new Callable<Integer>(){  
  2.     public Integer call() throws Exception {  
  3.         System.out.println("inside callable");  
  4.         Thread.sleep(1000);  
  5.         return new Integer(8);  
  6.     }         
  7. };        
  8. FutureTask<Integer> futureTask  = new FutureTask<Integer>(func);  
  9. Thread newThread = new Thread(futureTask);  
  10. newThread.start();  
  11.   
  12. try {  
  13.     System.out.println("blocking here");  
  14.     Integer result = futureTask.get();  
  15.     System.out.println(result);  
  16. } catch (InterruptedException ignored) {  
  17. } catch (ExecutionException ignored) {  
  18. }  

 ExecutoreService提供了submit()方法，传递一个Callable，或Runnable，返回Future。如果Executor后台线程池还没有完成Callable的计算，这调用返回Future对象的get()方法，会阻塞直到计算完成。

例子：并行计算数组的和。

Java代码 复制代码 收藏代码
  1. package executorservice;   
  2.   
  3. import java.util.ArrayList;   
  4. import java.util.List;   
  5. import java.util.concurrent.Callable;   
  6. import java.util.concurrent.ExecutionException;   
  7. import java.util.concurrent.ExecutorService;   
  8. import java.util.concurrent.Executors;   
  9. import java.util.concurrent.Future;   
  10. import java.util.concurrent.FutureTask;   
  11.   
  12. public class ConcurrentCalculator {   
  13.   
  14.     private ExecutorService exec;   
  15.     private int cpuCoreNumber;   
  16.     private List<Future<Long>> tasks = new ArrayList<Future<Long>>();   
  17.   
  18.     // 内部类   
  19.     class SumCalculator implements Callable<Long> {   
  20.         private int[] numbers;   
  21.         private int start;   
  22.         private int end;   
  23.   
  24.         public SumCalculator(final int[] numbers, int start, int end) {   
  25.             this.numbers = numbers;   
  26.             this.start = start;   
  27.             this.end = end;   
  28.         }   
  29.   
  30.         public Long call() throws Exception {   
  31.             Long sum = 0l;   
  32.             for (int i = start; i < end; i++) {   
  33.                 sum += numbers[i];   
  34.             }   
  35.             return sum;   
  36.         }   
  37.     }   
  38.   
  39.     public ConcurrentCalculator() {   
  40.         cpuCoreNumber = Runtime.getRuntime().availableProcessors();   
  41.         exec = Executors.newFixedThreadPool(cpuCoreNumber);   
  42.     }   
  43.   
  44.     public Long sum(final int[] numbers) {   
  45.         // 根据CPU核心个数拆分任务，创建FutureTask并提交到Executor   
  46.         for (int i = 0; i < cpuCoreNumber; i++) {   
  47.             int increment = numbers.length / cpuCoreNumber + 1;   
  48.             int start = increment * i;   
  49.             int end = increment * i + increment;   
  50.             if (end > numbers.length)   
  51.                 end = numbers.length;   
  52.             SumCalculator subCalc = new SumCalculator(numbers, start, end);   
  53.             FutureTask<Long> task = new FutureTask<Long>(subCalc);   
  54.             tasks.add(task);   
  55.             if (!exec.isShutdown()) {   
  56.                 exec.submit(task);   
  57.             }   
  58.         }   
  59.         return getResult();   
  60.     }   
  61.   
  62.     /**  
  63.      * 迭代每个只任务，获得部分和，相加返回  
  64.      *   
  65.      * @return  
  66.      */  
  67.     public Long getResult() {   
  68.         Long result = 0l;   
  69.         for (Future<Long> task : tasks) {   
  70.             try {   
  71.                 // 如果计算未完成则阻塞   
  72.                 Long subSum = task.get();   
  73.                 result += subSum;   
  74.             } catch (InterruptedException e) {   
  75.                 e.printStackTrace();   
  76.             } catch (ExecutionException e) {   
  77.                 e.printStackTrace();   
  78.             }   
  79.         }   
  80.         return result;   
  81.     }   
  82.   
  83.     public void close() {   
  84.         exec.shutdown();   
  85.     }   
  86. }  
Java代码  收藏代码
  1. package executorservice;  
  2.   
  3. import java.util.ArrayList;  
  4. import java.util.List;  
  5. import java.util.concurrent.Callable;  
  6. import java.util.concurrent.ExecutionException;  
  7. import java.util.concurrent.ExecutorService;  
  8. import java.util.concurrent.Executors;  
  9. import java.util.concurrent.Future;  
  10. import java.util.concurrent.FutureTask;  
  11.   
  12. public class ConcurrentCalculator {  
  13.   
  14.     private ExecutorService exec;  
  15.     private int cpuCoreNumber;  
  16.     private List<Future<Long>> tasks = new ArrayList<Future<Long>>();  
  17.   
  18.     // 内部类  
  19.     class SumCalculator implements Callable<Long> {  
  20.         private int[] numbers;  
  21.         private int start;  
  22.         private int end;  
  23.   
  24.         public SumCalculator(final int[] numbers, int start, int end) {  
  25.             this.numbers = numbers;  
  26.             this.start = start;  
  27.             this.end = end;  
  28.         }  
  29.   
  30.         public Long call() throws Exception {  
  31.             Long sum = 0l;  
  32.             for (int i = start; i < end; i++) {  
  33.                 sum += numbers[i];  
  34.             }  
  35.             return sum;  
  36.         }  
  37.     }  
  38.   
  39.     public ConcurrentCalculator() {  
  40.         cpuCoreNumber = Runtime.getRuntime().availableProcessors();  
  41.         exec = Executors.newFixedThreadPool(cpuCoreNumber);  
  42.     }  
  43.   
  44.     public Long sum(final int[] numbers) {  
  45.         // 根据CPU核心个数拆分任务，创建FutureTask并提交到Executor  
  46.         for (int i = 0; i < cpuCoreNumber; i++) {  
  47.             int increment = numbers.length / cpuCoreNumber + 1;  
  48.             int start = increment * i;  
  49.             int end = increment * i + increment;  
  50.             if (end > numbers.length)  
  51.                 end = numbers.length;  
  52.             SumCalculator subCalc = new SumCalculator(numbers, start, end);  
  53.             FutureTask<Long> task = new FutureTask<Long>(subCalc);  
  54.             tasks.add(task);  
  55.             if (!exec.isShutdown()) {  
  56.                 exec.submit(task);  
  57.             }  
  58.         }  
  59.         return getResult();  
  60.     }  
  61.   
  62.     /** 
  63.      * 迭代每个只任务，获得部分和，相加返回 
  64.      *  
  65.      * @return 
  66.      */  
  67.     public Long getResult() {  
  68.         Long result = 0l;  
  69.         for (Future<Long> task : tasks) {  
  70.             try {  
  71.                 // 如果计算未完成则阻塞  
  72.                 Long subSum = task.get();  
  73.                 result += subSum;  
  74.             } catch (InterruptedException e) {  
  75.                 e.printStackTrace();  
  76.             } catch (ExecutionException e) {  
  77.                 e.printStackTrace();  
  78.             }  
  79.         }  
  80.         return result;  
  81.     }  
  82.   
  83.     public void close() {  
  84.         exec.shutdown();  
  85.     }  
  86. }  

 Main

Java代码 复制代码 收藏代码
  1. int[] numbers = new int[] { 1, 2, 3, 4, 5, 6, 7, 8, 10, 11 };   
  2. ConcurrentCalculator calc = new ConcurrentCalculator();   
  3. Long sum = calc.sum(numbers);   
  4. System.out.println(sum);   
  5. calc.close();  
Java代码  收藏代码
  1. int[] numbers = new int[] { 1, 2, 3, 4, 5, 6, 7, 8, 10, 11 };  
  2. ConcurrentCalculator calc = new ConcurrentCalculator();  
  3. Long sum = calc.sum(numbers);  
  4. System.out.println(sum);  
  5. calc.close();  

 四、CompletionService

在刚在的例子中，getResult()方法的实现过程中，迭代了FutureTask的数组，如果任务还没有完成则当前线程会阻塞，如果我们希望任意字任务完成后就把其结果加到result中，而不用依次等待每个任务完成，可以使CompletionService。生产者submit()执行的任务。使用者take()已完成的任务，并按照完成这些任务的顺序处理它们的结果 。也就是调用CompletionService的take方法是，会返回按完成顺序放回任务的结果，CompletionService内部维护了一个阻塞队列BlockingQueue，如果没有任务完成，take()方法也会阻塞。修改刚才的例子使用CompletionService：

Java代码 复制代码 收藏代码
  1. public class ConcurrentCalculator2 {   
  2.   
  3.     private ExecutorService exec;   
  4.     private CompletionService<Long> completionService;   
  5.   
  6.   
  7.     private int cpuCoreNumber;   
  8.   
  9.     // 内部类   
  10.     class SumCalculator implements Callable<Long> {   
  11.         ......   
  12.     }   
  13.   
  14.     public ConcurrentCalculator2() {   
  15.         cpuCoreNumber = Runtime.getRuntime().availableProcessors();   
  16.         exec = Executors.newFixedThreadPool(cpuCoreNumber);   
  17.         completionService = new ExecutorCompletionService<Long>(exec);   
  18.   
  19.   
  20.     }   
  21.   
  22.     public Long sum(final int[] numbers) {   
  23.         // 根据CPU核心个数拆分任务，创建FutureTask并提交到Executor   
  24.         for (int i = 0; i < cpuCoreNumber; i++) {   
  25.             int increment = numbers.length / cpuCoreNumber + 1;   
  26.             int start = increment * i;   
  27.             int end = increment * i + increment;   
  28.             if (end > numbers.length)   
  29.                 end = numbers.length;   
  30.             SumCalculator subCalc = new SumCalculator(numbers, start, end);    
  31.             if (!exec.isShutdown()) {   
  32.                 completionService.submit(subCalc);   
  33.   
  34.   
  35.             }   
  36.                
  37.         }   
  38.         return getResult();   
  39.     }   
  40.   
  41.     /**  
  42.      * 迭代每个只任务，获得部分和，相加返回  
  43.      *   
  44.      * @return  
  45.      */  
  46.     public Long getResult() {   
  47.         Long result = 0l;   
  48.         for (int i = 0; i < cpuCoreNumber; i++) {               
  49.             try {   
  50.                 Long subSum = completionService.take().get();   
  51.                 result += subSum;              
  52.             } catch (InterruptedException e) {   
  53.                 e.printStackTrace();   
  54.             } catch (ExecutionException e) {   
  55.                 e.printStackTrace();   
  56.             }   
  57.         }   
  58.         return result;   
  59.     }   
  60.   
  61.     public void close() {   
  62.         exec.shutdown();   
  63.     }   
  64. }  
Java代码  收藏代码
  1. public class ConcurrentCalculator2 {  
  2.   
  3.     private ExecutorService exec;  
  4.     private CompletionService<Long> completionService;  
  5.   
  6.   
  7.     private int cpuCoreNumber;  
  8.   
  9.     // 内部类  
  10.     class SumCalculator implements Callable<Long> {  
  11.         ......  
  12.     }  
  13.   
  14.     public ConcurrentCalculator2() {  
  15.         cpuCoreNumber = Runtime.getRuntime().availableProcessors();  
  16.         exec = Executors.newFixedThreadPool(cpuCoreNumber);  
  17.         completionService = new ExecutorCompletionService<Long>(exec);  
  18.   
  19.   
  20.     }  
  21.   
  22.     public Long sum(final int[] numbers) {  
  23.         // 根据CPU核心个数拆分任务，创建FutureTask并提交到Executor  
  24.         for (int i = 0; i < cpuCoreNumber; i++) {  
  25.             int increment = numbers.length / cpuCoreNumber + 1;  
  26.             int start = increment * i;  
  27.             int end = increment * i + increment;  
  28.             if (end > numbers.length)  
  29.                 end = numbers.length;  
  30.             SumCalculator subCalc = new SumCalculator(numbers, start, end);   
  31.             if (!exec.isShutdown()) {  
  32.                 completionService.submit(subCalc);  
  33.   
  34.   
  35.             }  
  36.               
  37.         }  
  38.         return getResult();  
  39.     }  
  40.   
  41.     /** 
  42.      * 迭代每个只任务，获得部分和，相加返回 
  43.      *  
  44.      * @return 
  45.      */  
  46.     public Long getResult() {  
  47.         Long result = 0l;  
  48.         for (int i = 0; i < cpuCoreNumber; i++) {              
  49.             try {  
  50.                 Long subSum = completionService.take().get();  
  51.                 result += subSum;             
  52.             } catch (InterruptedException e) {  
  53.                 e.printStackTrace();  
  54.             } catch (ExecutionException e) {  
  55.                 e.printStackTrace();  
  56.             }  
  57.         }  
  58.         return result;  
  59.     }  
  60.   
  61.     public void close() {  
  62.         exec.shutdown();  
  63.     }  
  64. }  

 

 

 

 

 

1. ExecutorService

Java从1.5开始正式提供了并发包,而这个并发包里面除了原子变量,synchronizer,并发容器,另外一个非常重要的特性就是线程池.对于线程池的意义,我们这边不再多说.

上图是线程池的主体类图,ThreadPoolExecutor是应用最为广泛的一个线程池实现(我也将在接下来的文字中详细描述我对这个类的理解和执行机制),ScheduledThreadPoolExecutor则在ThreadPoolExecutor上提供了定时执行的等附加功能,这个可以从ScheduledExecutorService接口的定义中看出来.Executors则类似工厂方法,提供了几个非常常用的线程池初始化方法.

ThreadPoolExecutor

这个类继承了AbstractExecutorService抽象类, AbstractExecutorService主要的职责有2部分,一部分定义和实现提交任务的方法(3个submit方法的实现) ,实例化FutureTask并且交给子类执行,另外一部分实现invokeAny,invokeAll方法.留给子类的方法为execute方法,也就是Executor接口定义的方法.

//实例化一个FutureTask,交给子类的execute方法执行.这种设计能够保证callable和runnable的执行接口方法的一致性(FutureTask包装了这个差别)
public <T> Future<T> submit(Runnable task, T result) {
    if (task == null) throw new NullPointerException();
    RunnableFuture<T> ftask = newTaskFor(task, result);
    execute(ftask);
    return ftask;
}

protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) {
    return new FutureTask<T>(runnable, value);
}

关于FutureTask这个类的实现,我在前面的JAVA LOCK代码浅析有讲过其实现原理,主要的思想就是关注任务完成与未完成的状态,任务提交线程get()结果时被park住,等待任务执行完成被唤醒,任务执行线程在任务执行完毕后设置结果,并且unpark对应线程并且让其得到执行结果.

回到ThreadPoolExecutor类.ThreadPoolExecutor需要实现除了我们刚才说的execute(Runnable command)方法外,还得实现ExecutorService接口定义的部分方法.但ThreadPoolExecutor所提供的不光是这些,以下根据我的理解来列一下它所具有的特性
1. execute流程
2. 池
3. 工作队列
4. 饱和拒绝策略
5. 线程工厂
6. beforeExecute和afterExecute扩展

execute方法的实现有个机制非常重要,当当前线程池线程数量小于corePoolSize,那么生成一个新的worker并把提交的任务置为这个工作线程的头一个执行任务,如果大于corePoolSize,那么会试着将提交的任务塞到workQueue里面供线程池里面的worker稍后执行,并不是直接再起一个worker,但是当workQueue也满,并且当前线程池小于maxPoolSize,那么起一个新的worker并将该任务设为该worker执行的第一个任务执行,大于maxPoolSize,workQueue也满负荷,那么调用饱和策略里面的行为.

worker线程在执行完一个任务之后并不会立刻关闭,而是尝试着去workQueue里面取任务,如果取不到,根据策略关闭或者保持空闲状态.所以submit任务的时候,提交的顺序为核心线程池——工作队列——扩展线程池.

池包括核心池,扩展池(2者的线程在同一个hashset中,这里只是为了方便才这么称呼,并不是分离的),核心池在池内worker没有用完的情况下,只要有任务提交都会创建新的线程,其代表线程池正常处理任务的能力.扩展池,是在核心线程池用完,并且工作队列也已排满任务的情况下才会开始初始化线程,其代表的是线程池超出正常负载时的解决方案,一旦任务完成,并且试图从workQueue取不到任务,那么会比较当前线程池与核心线程池的大小,大于核心线程池数的worker将被销毁.

Runnable getTask() {
    for (;;) {
        try {
            int state = runState;
            //>SHUTDOWN就是STOP或者TERMINATED
            //直接返回
            if (state > SHUTDOWN)
                return null;
            Runnable r;
            //如果是SHUTDOWN状态,那么取任务,如果有
              //将剩余任务执行完毕,否则就结束了
            if (state == SHUTDOWN)  // Help drain queue
                r = workQueue.poll();
            //如果不是以上状态的(也就是RUNNING状态的),那么如果当前池大于核心池数量,
            //或者允许核心线程池取任务超时就可以关闭,那么从任务队列取任务,
            //如果超出keepAliveTime,那么就返回null了,也就意味着这个worker结束了
            else if (poolSize > corePoolSize || allowCoreThreadTimeOut)
                r = workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS);
            //如果当前池小于核心池,并且不允许核心线程池取任务超时就关闭,那么take(),直到拿到任务或者被interrupt
            else
                r = workQueue.take();
            //如果经过以上判定,任务不为空,那么返回任务
            if (r != null)
                return r;
            //如果取到任务为空,那么判定是否可以退出
            if (workerCanExit()) {
                //如果整个线程池状态变为SHUTDOWN或者TERMINATED,那么将所有worker interrupt (如果正在执行,那继续让其执行)
                if (runState >= SHUTDOWN) // Wake up others
                    interruptIdleWorkers();
                return null;
            }
            // Else retry
        } catch (InterruptedException ie) {
            // On interruption, re-check runState
        }
}
    }

//worker从workQueue中取不到数据的时候调用此方法,以决定自己是否跳出取任务的无限循环,从而结束此worker的运行
private boolean workerCanExit() {
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    boolean canExit;
    try {
        /**//*
        *线程池状态为stop或者terminated,
        *或者任务队列里面任务已经为空,
        *或者允许线程池线程空闲超时(实现方式是从工作队列拿最多keepAliveTime的任务,超过这个时间就返回null了)并且
         *当前线程池大于corePoolSize(>1)
        *那么允许线程结束
        *static final int RUNNING    = 0;
        *static final int SHUTDOWN   = 1;
        *static final int STOP       = 2;
        *static final int TERMINATED = 3;
        */
        canExit = runState >= STOP ||
        workQueue.isEmpty() ||
       (allowCoreThreadTimeOut &&
        poolSize > Math.max(1,corePoolSize));
    } finally {
        mainLock.unlock();
    }
    return canExit;
}

当提交任务是,线程池都已满,并且工作队列也无空闲位置的情况下,ThreadPoolExecutor会执行reject操作,JDK提供了四种reject策略,包括AbortPolicy(直接抛RejectedException Exception),CallerRunsPolicy(提交任务线程自己执行,当然这时剩余任务也将无法提交),DiscardOldestPolicy(将线程池的workQueue任务队列里面最老的任务剔除,将新任务丢入),DiscardPolicy(无视,忽略此任务,并且立即返回).实例化ThreadPoolExecutor时,如果不指定任何饱和策略,默认将使用AbortPolicy.

个人认为这些饱和策略并不十分理想,特别是在应用既要保证快速,又要高可用的情况下,我的想法是能够加入超时等待策略,也就是提交线程时线程池满,能够park住提交任务的线程,一旦有空闲,能在第一时间通知到等待线程. 这个实际上和主线程执行相似,但是主线程执行期间即使线程池有大量空闲也不会立即可以提交任务,效率上后者可能会比较低,特别是执行慢速任务.

实例化Worker的时候会调用ThreadFactory的addThread(Runnable r)方法返回一个Thread,这个线程工厂是可以在ThreadPoolExecutor实例化的时候指定的,如果不指定,那么将会使用DefaultThreadFactory, 这个也就是提供给使用者命名线程,线程归组,是否是demon等线程相关属性设置的机会.

beforeExecute和afterExecute是提供给使用者扩展的,这两个方法会在worker runTask之前和run完毕之后分别调用.JDK注释里 Doug Lea(concurrent包作者)展示了beforeExecute一个很有趣的示例.代码如下.

class PausableThreadPoolExecutor extends ThreadPoolExecutor {
    private boolean isPaused;
    private ReentrantLock pauseLock = new ReentrantLock();
    private Condition unpaused = pauseLock.newCondition();

public PausableThreadPoolExecutor() { super(); }

protected void beforeExecute(Thread t, Runnable r) {
    super.beforeExecute(t, r);
    pauseLock.lock();
    try {
        while (isPaused) unpaused.await();
    } catch (InterruptedException ie) {
        t.interrupt();
    } finally {
        pauseLock.unlock();
    }
}

public void pause() {
    pauseLock.lock();
    try {
        isPaused = true;
    } finally {
        pauseLock.unlock();
    }
}

public void resume() {
    pauseLock.lock();
    try {
        isPaused = false;
        unpaused.signalAll();
    } finally {
        pauseLock.unlock();
    }
}
  }

使用这个线程池,用户可以随时调用pause中止剩余任务执行,当然也可以使用resume重新开始执行剩余任务.

ScheduledThreadPoolExecutor

ScheduledThreadPoolExecutor是一个很实用的类,它的实现核心是基于DelayedWorkQueue.从ScheduledThreadPoolExecutor的继承结构上来看,各位应该能够看出些端倪来,就是ScheduledThreadPoolExecutor将ThreadPoolExecutor中的任务队列设置成了DelayedWorkQueue,这也就是说,线程池Worker从任务队列中取的一个任务,需要等待这个队列中最短超时任务的超时,也就是实现定时的效果.所以ScheduledThreadPoolExecutor所做的工作其实是比较少的.主要就是实现任务的实例化并加入工作队列,以及支持scheduleAtFixedRate和scheduleAtFixedDelay这种周期性任务执行.

public ScheduledThreadPoolExecutor(int corePoolSize,ThreadFactory threadFactory) {
           super(corePoolSize, Integer.MAX_VALUE, 0, TimeUnit.NANOSECONDS,new DelayedWorkQueue(), threadFactory);
}

对于scheduleAfFixedRate和scheduleAtFiexedDelay这种周期性任务支持,是由ScheduledThreadPoolExecutor内部封装任务的ScheduledFutureTask来实现的.这个类在执行任务后,对于周期性任务,它会处理周期时间,并将自己再次丢入线程池的工作队列,从而达到周期执行的目的.

private void runPeriodic() {
          boolean ok = ScheduledFutureTask.super.runAndReset();
          boolean down = isShutdown();
          // Reschedule if not cancelled and not shutdown or policy allows
      if (ok && (!down ||(getContinueExistingPeriodicTasksAfterShutdownPolicy() && !isStopped()))) {
                long p = period;
                if (p > 0)
                      time += p;
                else
                      time = triggerTime(-p);

                ScheduledThreadPoolExecutor.super.getQueue().add(this);
         }
         // This might have been the final executed delayed
        // task.  Wake up threads to check.
        else if (down)
              interruptIdleWorkers();
}

2. CompletionService

ExecutorCompletionService

CompletionService定义了线程池执行任务集,可以依次拿到任务执行完毕的Future,ExecutorCompletionService是其实现类,先举个例子,如下代码,这个例子中,需要注意ThreadPoolExecutor核心池一定保证能够让任务提交并且马上执行,而不是放到等待队列中去,那样次序将会无法控制,CompletionService也将失去效果(其实核心池中的任务完成顺序还是准确的).

public static void main(String[] args) throws InterruptedException, ExecutionException{
    ThreadPoolExecutor es=new ThreadPoolExecutor(10, 15, 2000, TimeUnit.MILLISECONDS, new ArrayBlockingQueue<Runnable>(10),new ThreadPoolExecutor.AbortPolicy());
    CompletionService<String> cs=new ExecutorCompletionService<String>(es);
    cs.submit(new Callable<String>() {
     @Override
     public String call() throws Exception {
         Thread.currentThread().sleep(1000);
         return "i am sleeped 1000 milliseconds";
     }
    });

    cs.submit(new Callable<String>() {
     @Override
     public String call() throws Exception {
         Thread.currentThread().sleep(5000);
         return "i am sleeped 5000 milliseconds";
     }
    });

    cs.submit(new Callable<String>() {
     @Override
     public String call() throws Exception {
         Thread.currentThread().sleep(4000);
         return "i am sleeped 4000 milliseconds";
     }
    });

    cs.submit(new Callable<String>() {
     @Override
         public String call() throws Exception {
          Thread.currentThread().sleep(2000);
              return "i am sleeped 2000 milliseconds";
      }
});

    for(int i=0;i<4;i++){
        Future<String> fu=cs.take();
        System.out.println(fu.get());
    }
}

执行结果:
i am sleeped 1000 milliseconds
i am sleeped 2000 milliseconds
i am sleeped 4000 milliseconds
i am sleeped 5000 milliseconds
从执行结果看来,我们发现先完成的任务先被拿出来了,直到所有任务被执行完毕,也就是CompletionService的效果达到了.

ExecutorCompletionService并不复杂,关键的一个点就是它的内部类QueueingFuture继承了FutureTask类,并且实现了done()方法,done()方法是在线程池任务执行完毕,最后调用FutureTask的方法(这在 JAVA LOCK代码浅析(http://www.blogjava.net/BucketLi/archive/2010/09/30/333471.html)一文中对于FutureTask代码解析有提到)

QueueingFuture的done()方法实现是将执行完的任务(FutureTask)丢入全局的完成队列中(completionQueue),那么take是从这个blockingqueue中取元素.也就是任务完成就会有元素,即生产者消费者.

这种实现的思想是将原本在单个FutureTask上的等待转化为在BlockingQueue上的等待,即对全部FutureTask的等待,从而达到哪个先完成,哪个就可取执行结果的效果.

private class QueueingFuture extends FutureTask<Void> {
    QueueingFuture(RunnableFuture<V> task) {
        super(task, null);
        his.task = task;
    }
    protected void done() { completionQueue.add(task); }
    private final Future<V> task;
}

总结:
JUC提供的线程池体系核心是在ThreadPoolExecutor, 而ScheduledThreadPoolExecutor和ExecutorCompletionService只是对其扩展,这里没有去细讲Executors这个便捷类,这个类提供很多便捷的线程池构建方法.各位使用的时候不妨去看下.

分享到：
java解压rar文件
评论
发表评论

您还没有登录,请您登录后再发表评论

相关推荐

    《Java并发编程的艺术》

    《Java并发编程的艺术》内容涵盖Java并发编程机制的底层实现原理、Java内存模型、Java并发编程基础、Java中的锁、并发容器和框架、原子类、并发工具类、线程池、Executor框架等主题，每个主题都做了深入的讲解，同时...

    Java并发编程实战

    6.2　Executor框架 6.2.1　示例：基于Executor的Web服务器 6.2.2　执行策略 6.2.3　线程池 6.2.4　Executor的生命周期 6.2.5　延迟任务与周期任务 6.3　找出可利用的并行性 6.3.1　示例：串行的页面渲染器 6.3.2　...

    Java并发编程的艺术

    , 《Java并发编程的艺术》内容涵盖Java并发编程机制的底层实现原理、Java内存模型、Java并发编程基础、Java中的锁、并发容器和框架、原子类、并发工具类、线程池、Executor框架等主题，每个主题都做了深入的讲解，...

    《Java并发编程的艺术》源代码

    Java并发编程的艺术 作者:方腾飞 魏鹏 程晓明　著 丛书名:Java核心技术系列 出版日期 :2015-07-25 ISBN:978-7-111-50824-3 第1章介绍Java并发编程的挑战，向读者说明进入并发编程的世界可能会遇到哪些问题，以及如何...

    Java并发编程实践 PDF 高清版

    本书的读者是那些具有一定Java编程经验的程序员、希望了解Java SE 5，6在线程技术上的改进和新特性的程序员，以及Java和并发编程的爱好者。 目录 代码清单 序 第1章 介绍 1.1 并发的（非常）简短历史 1.2 线程的...

    Java 并发编程实战

    6.2　Executor框架 6.2.1　示例：基于Executor的Web服务器 6.2.2　执行策略 6.2.3　线程池 6.2.4　Executor的生命周期 6.2.5　延迟任务与周期任务 6.3　找出可利用的并行性 6.3.1　示例：串行的页面渲染器 6.3.2　...

    Java并发编程原理与实战

    Executor框架详解.mp4 实战：简易web服务器（一）.mp4 实战：简易web服务器（二）.mp4 JDK8的新增原子操作类LongAddr原理与使用.mp4 JDK8新增锁StampedLock详解.mp4 重排序问题.mp4 happens-before简单概述.mp4 锁的...

    Java并发编程的艺术_非扫描

    Java并发编程的艺术_非扫描本书特色本书结合JDK的源码介绍了Java并发框架、线程池的实现原理，帮助读者做到知其所以然。本书对原理的剖析不仅仅局限于Java层面，而是深入到JVM，甚至CPU层面来进行讲解，帮助读者从更...

    龙果 java并发编程原理实战

    龙果 java并发编程原理实战 第2节理解多线程与并发的之间的联系与区别 [免费观看] 00:11:59分钟 | 第3节解析多线程与多进程的联系以及上下文切换所导致资源浪费问题 [免费观看] 00:13:03分钟 | 第4节学习并发的四...

    JAVA并发编程实践_中文版(1-16章全)_1/4

    6.2 executor 框架 6.3 寻找可强化的并行性 第7章 取消和关闭 7.1 任务取消 7.2 停止基于线程的服务 7.3 处理反常的线程终止 7.4 jvm关闭 第8章 应用线程池 8.1 任务与执行策略问的隐性耦合 8.2 定制线程池的大小 ...

    Java 7并发编程实战手册

    java7在并发编程方面，带来了很多令人激动的新功能，这将使你的应用程序具备更好的并行任务性能。 《Java 7并发编程实战手册》是Java 7并发编程的实战指南，介绍了Java 7并发API中大部分重要而有用的机制。全书分为9...

    针对于Executor框架,Java API,线程共享数据

    Executor框架是Java并发编程中的一个重要工具，它提供了一种管理线程池的方式，使得我们可以更方便地管理线程的生命周期和执行线程任务。 原子操作是指不可被中断的操作，要么全部执行成功，要么全部不执行。原子...

    龙果java并发编程完整视频

    第53节Executor框架详解00:36:54分钟 | 第54节实战：简易web服务器（一）00:55:34分钟 | 第55节实战：简易web服务器（二）00:24:36分钟 | 第56节JDK8的新增原子操作类LongAddr原理与使用00:17:45分钟 | 第57节...

    汪文君高并发编程实战视频资源下载.txt

    │ Java并发编程.png │ ppt+源码.rar │ 高并发编程第二阶段01讲、课程大纲及主要内容介绍.wmv │ 高并发编程第二阶段02讲、介绍四种Singleton方式的优缺点在多线程情况下.wmv │ 高并发编程第二阶段03讲、...

    Java 并发编程原理与实战视频

    java并发编程原理实战 第2节理解多线程与并发的之间的联系与区别 [免费观看] 00:11:59分钟 | 第3节解析多线程与多进程的联系以及上下文切换所导致资源浪费问题 [免费观看] 00:13:03分钟 | 第4节学习并发的四个...

Magicbox
Global site tag (gtag.js) - Google Analytics