Java的Executor框架和线程池实现原理， callable， Future， Runnable

Java的Executor框架和线程池实现原理， callable， Future， Runnable

https://blog.csdn.net/tuke_tuke/article/details/51353925

一，Java的Executor框架

1，Executor接口

1.  
   public interface Executor {
2.  
   void execute(Runnable command);
3.  
   }

Executor接口是Executor框架中最基础的部分，定义了一个用于执行Runnable的execute方法，它没有实现类只有另一个重要的子接口ExecutorService

2，ExecutorService接口

1.  
   //继承自Executor接口
2.  
   public interface ExecutorService extends Executor {
3.  
   /**
4.  
   * 关闭方法，调用后执行之前提交的任务，不再接受新的任务
5.  
   */
6.  
   void shutdown();
7.  
   /**
8.  
   * 从语义上可以看出是立即停止的意思，将暂停所有等待处理的任务并返回这些任务的列表
9.  
   */
10.  
    List<Runnable> shutdownNow();
11.  
    /**
12.  
    * 判断执行器是否已经关闭
13.  
    */
14.  
    boolean isShutdown();
15.  
    /**
16.  
    * 关闭后所有任务是否都已完成
17.  
    */
18.  
    boolean isTerminated();
19.  
    /**
20.  
    * 中断
21.  
    */
22.  
    boolean awaitTermination(long timeout, TimeUnit unit)
23.  
    throws InterruptedException;
24.  
    /**
25.  
    * 提交一个Callable任务
26.  
    */
27.  
    <T> Future<T> submit(Callable<T> task);
28.  
    /**
29.  
    * 提交一个Runable任务，result要返回的结果
30.  
    */
31.  
    <T> Future<T> submit(Runnable task, T result);
32.  
    /**
33.  
    * 提交一个任务
34.  
    */
35.  
    Future<?> submit(Runnable task);
36.  
    /**
37.  
    * 执行所有给定的任务，当所有任务完成，返回保持任务状态和结果的Future列表
38.  
    */
39.  
    <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
40.  
    throws InterruptedException;
41.  
    /**
42.  
    * 执行给定的任务，当所有任务完成或超时期满时（无论哪个首先发生），返回保持任务状态和结果的 Future 列表。
43.  
    */
44.  
    <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,
45.  
    long timeout, TimeUnit unit)
46.  
    throws InterruptedException;
47.  
    /**
48.  
    * 执行给定的任务，如果某个任务已成功完成（也就是未抛出异常），则返回其结果。
49.  
    */
50.  
    <T> T invokeAny(Collection<? extends Callable<T>> tasks)
51.  
    throws InterruptedException, ExecutionException;
52.  
    /**
53.  
    * 执行给定的任务，如果在给定的超时期满前某个任务已成功完成（也就是未抛出异常），则返回其结果。
54.  
    */
55.  
    <T> T invokeAny(Collection<? extends Callable<T>> tasks,
56.  
    long timeout, TimeUnit unit)
57.  
    throws InterruptedException, ExecutionException, TimeoutException;
58.  
    }

ExecutorService接口继承自Executor接口，定义了终止、提交,执行任务、跟踪任务返回结果等方法

1，execute（Runnable command）：履行Ruannable类型的任务,

2，submit（task）：可用来提交Callable或Runnable任务，并返回代表此任务的Future对象
3，shutdown（）：在完成已提交的任务后封闭办事，不再接管新任务,

4，shutdownNow（）：停止所有正在履行的任务并封闭办事。
5，isTerminated（）：测试是否所有任务都履行完毕了。,

6，isShutdown（）：测试是否该ExecutorService已被关闭

3，Executors的静态方法：负责生成各种类型的ExecutorService线程池实例

+newFixedThreadPool(numberOfThreads:int):（固定线程池）ExecutorService 创建一个固定线程数量的线程池，并行执行的线程数量不变，线程当前任务完成后，可以被重用执行另一个任务
+newCachedThreadPool():（可缓存线程池）ExecutorService 创建一个线程池，按需创建新线程，就是有任务时才创建，空闲线程保存60s，当前面创建的线程可用时，则重用它们

+new SingleThreadExecutor();（单线程执行器）线程池中只有一个线程，依次执行任务

+new ScheduledThreadPool()：线程池按时间计划来执行任务，允许用户设定执行任务的时间

+new SingleThreadScheduledExcutor();线程池中只有一个线程，它按规定时间来执行任务

4，Runnable、Callable、Future接口

Runnable接口：

1.  
   // 实现Runnable接口的类将被Thread执行，表示一个基本的任务
2.  
   public interface Runnable {
3.  
   // run方法就是它所有的内容，就是实际执行的任务
4.  
   public abstract void run();
5.  
   }

Callable接口：与Runnable接口的区别在于它接收泛型，同时它执行任务后带有返回内容

1.  
   // Callable同样是任务，与Runnable接口的区别在于它接收泛型，同时它执行任务后带有返回内容
2.  
   public interface Callable<V> {
3.  
   // 相对于run方法的带有返回值的call方法
4.  
   V call() throws Exception;
5.  
   }

Runnable接口和Callable接口的实现类，都可以被ThreadPoolExecutor和ScheduledThreadPoolExecutor执行，他们之间的区别是Runnable不会返回结果，而Callable可以返回结果。

Executors可以把一个Runnable对象转换成Callable对象：

public static Callable<Object> callable(Runnbale task);

Executors把一个Runnable和一个待返回的结果包装成一个Callable的API：

public static<T> Callable<T> callable(Runnbale task,T result);

当把一个Callable对象(Callable1,Callable2)提交给ThreadPoolExecutor和ScheduledThreadPoolExecutor执行时，submit(...)会向我们返回一个FutureTask对象。我们执行FutureTask.get()来等待任务执行完成，当任务完成后，FutureTask.get()将返回任务的结果。

Future接口：

1.  
   // Future代表异步任务的执行结果
2.  
   public interface Future<V> {
3.  
    
4.  
   /**
5.  
   * 尝试取消一个任务，如果这个任务不能被取消（通常是因为已经执行完了），返回false，否则返回true。
6.  
   */
7.  
   boolean cancel(boolean mayInterruptIfRunning);
8.  
    
9.  
   /**
10.  
    * 返回代表的任务是否在完成之前被取消了
11.  
    */
12.  
    boolean isCancelled();
13.  
     
14.  
    /**
15.  
    * 如果任务已经完成，返回true
16.  
    */
17.  
    boolean isDone();
18.  
     
19.  
    /**
20.  
    * 获取异步任务的执行结果（如果任务没执行完将等待）
21.  
    */
22.  
    V get() throws InterruptedException, ExecutionException;
23.  
     
24.  
    /**
25.  
    * 获取异步任务的执行结果（有最常等待时间的限制）
26.  
    *
27.  
    * timeout表示等待的时间，unit是它时间单位
28.  
    */
29.  
    V get(long timeout, TimeUnit unit)
30.  
    throws InterruptedException, ExecutionException, TimeoutException;
31.  
    }

Future就是对于具体的Runnable或者Callable任务的执行结果进行取消、查询是否完成、获取结果。必要时可以通过get方法获取执行结果，该方法会阻塞直到任务返回结果

在Future接口中声明了5个方法，下面依次解释每个方法的作用：
+cancel方法用来取消任务，如果取消任务成功则返回true，如果取消任务失败则返回false。参数mayInterruptIfRunning表示是否允许取消正在执行却没有执行完毕的任务，如果设置true，则表示可以取消正在执行过程中的任务。如果任务已经完成，则无论mayInterruptIfRunning为true还是false，此方法肯定返回false，即如果取消已经完成的任务会返回false；如果任务正在执行，若mayInterruptIfRunning设置为true，则返回true，若mayInterruptIfRunning设置为false，则返回false；如果任务还没有执行，则无论mayInterruptIfRunning为true还是false，肯定返回true。
+isCancelled方法表示任务是否被取消成功，如果在任务正常完成前被取消成功，则返回 true。
+isDone方法表示任务是否已经完成，若任务完成，则返回true；
+get()方法用来获取执行结果，这个方法会产生阻塞，会一直等到任务执行完毕才返回；
+get(long timeout, TimeUnit unit)用来获取执行结果，如果在指定时间内，还没获取到结果，就直接返回null。
也就是说Future提供了三种功能：
　　1）判断任务是否完成；
　　2）能够中断任务；
　　3）能够获取任务执行结果。

FutureTask:

通常使用FutureTask来处理我们的任务。FutureTask类同时又实现了Runnable接口，所以可以直接提交给Executor执行。

1.  
   FutureTask提供了2个构造器：
2.  
    
3.  
   public FutureTask(Callable<V> callable) {
4.  
   }
5.  
   public FutureTask(Runnable runnable, V result) {
6.  
   }
7.  
   　　//事实上，FutureTask是Future接口的一个唯一实现类。

使用FutureTask实现超时执行的代码如下：

1.  
   xecutorService executor = Executors.newSingleThreadExecutor();
2.  
   FutureTask<String> future =
3.  
   new FutureTask<String>(new Callable<String>() {//使用Callable接口作为构造参数
4.  
   public String call() {
5.  
   //真正的任务在这里执行，这里的返回值类型为String，可以为任意类型
6.  
   }});
7.  
   executor.execute(future);
8.  
   //在这里可以做别的任何事情
9.  
   try {
10.  
    result = future.get(5000, TimeUnit.MILLISECONDS); //取得结果，同时设置超时执行时间为5秒。同样可以用future.get()，不设置执行超时时间取得结果
11.  
    } catch (InterruptedException e) {
12.  
    futureTask.cancel(true);
13.  
    } catch (ExecutionException e) {
14.  
    futureTask.cancel(true);
15.  
    } catch (TimeoutException e) {
16.  
    futureTask.cancel(true);
17.  
    } finally {
18.  
    executor.shutdown();
19.  
    }

不直接构造Future对象，也可以使用ExecutorService.submit方法来获得Future对象，submit方法即支持以 Callable接口类型，也支持Runnable接口作为参数，具有很大的灵活性。使用示例如下：

1.  
   ExecutorService executor = Executors.newSingleThreadExecutor();
2.  
   FutureTask<String> future =　executor.submit(
3.  
   new Callable<String>() {//使用Callable接口作为构造参数
4.  
   public String call() {
5.  
   //真正的任务在这里执行，这里的返回值类型为String，可以为任意类型
6.  
   }});
7.  
   //在这里可以做别的任何事情
8.  
   //同上面取得结果的代码

线程池实现原理详解：

ThreadPoolExecutor是线程池的实现类：

1.  
   public ThreadPoolExecutor(int corePoolSize,
2.  
   int maximumPoolSize,
3.  
   long keepAliveTime,
4.  
   TimeUnit unit,
5.  
   BlockingQueue<Runnable> workQueue,
6.  
   ThreadFactory threadFactory,
7.  
   RejectedExecutionHandler handler)

（1）corePoolSize（线程池的基本大小）：当提交一个任务到线程池时，线程会创建一个线程来执行任务，

即使有其他空闲的基本线程还是会去创建线程，等到需要执行的任务数大于线程池基本大小corePoolSize时就不会再创建。

（2）maximumPoolSize（线程池最大大小）：线程池允许最大线程数。如果阻塞队列满了，并且已经创建的线程数小于最大线程数，则线程池会再创建新的线程执行。因为线程池执行任务时是线程池基本大小满了，后续任务进入阻塞队列，阻塞队列满了，在创建线程。

（3）keepAliveTime（线程活动保持时间）：空闲线程的保持存活时间。
（4）TimeUnit（线程活动保持时间的单位）：

        TimeUnit.DAYS; //天
        TimeUnit.HOURS; //小时
        TimeUnit.MINUTES; //分钟
        TimeUnit.SECONDS; //秒
        TimeUnit.MILLISECONDS; //毫秒
        TimeUnit.MICROSECONDS; //微妙
        TimeUnit.NANOSECONDS; //纳秒

（5）workQueue（任务队列）：用于保存等待执行的任务的阻塞队列。一个阻塞队列，用来存储等待执行的任务：数组，链表，不存元素的阻塞队列

      5.1）ArrayBlockingQueue;数组结构的有界阻塞队列，先进先出FIFO
      5.2）LinkedBlockingQueue;链表结构的无界阻塞队列。先进先出FIFO排序元素，静态方法Executors.newFixedThreadPool使用这个方法

      5.3）SynchronousQueue;不存储元素的阻塞队列，就是每次插入操作必须等到另一个线程调用移除操作，静态方法Executors.newCachedThreadPool使用这个方法

 （6）threadFactory：用于设置创建线程的工厂，可以通过线程工厂给每个创建出来的线程设置更有意义的名字

 （7）handler（饱和策略）：表示当拒绝处理任务时的策略。当队列和线程池都满了，说明线程池处于饱和状态，那么必须采取一种策略处理提交的新任务。这个策略默认情况下是AbortPolicy，表示无法处理新任务时抛出异常。

     ThreadPoolExecutor.AbortPolicy:丢弃任务并抛出RejectedExecutionException异常。

     ThreadPoolExecutor.DiscardPolicy：也是丢弃任务，但是不抛出异常。

     ThreadPoolExecutor.DiscardOldestPolicy：丢弃队列最前面的任务，然后重新尝试执行任务（重复此过程）

    ThreadPoolExecutor.CallerRunsPolicy：由调用线程处理该任务

我们尽量优先使用Executors提供的静态方法来创建线程池，如果Executors提供的方法无法满足要求，再自己通过ThreadPoolExecutor类来创建线程池   

1.  
   Executors.newFixedThreadPool(int); //创建固定容量大小的缓冲池
2.  
   Executors.newCachedThreadPool(); //创建一个缓冲池，缓冲池容量大小为Integer.MAX_VALUE
3.  
   Executors.newSingleThreadExecutor(); //创建容量为1的缓冲池

下面是这三个静态方法的具体实现;

1.  
   public static ExecutorService newFixedThreadPool(int nThreads) {
2.  
   return new ThreadPoolExecutor(nThreads, nThreads,
3.  
   0L, TimeUnit.MILLISECONDS,
4.  
   new LinkedBlockingQueue<Runnable>());
5.  
   }
6.  
   public static ExecutorService newSingleThreadExecutor() {
7.  
   return new FinalizableDelegatedExecutorService
8.  
   (new ThreadPoolExecutor(1, 1,
9.  
   0L, TimeUnit.MILLISECONDS,
10.  
    new LinkedBlockingQueue<Runnable>()));
11.  
    }
12.  
    public static ExecutorService newCachedThreadPool() {
13.  
    return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
14.  
    60L, TimeUnit.SECONDS,
15.  
    new SynchronousQueue<Runnable>());
16.  
    }

从它们的具体实现来看，它们实际上也是调用了ThreadPoolExecutor，只不过参数都已配置好了。

newFixedThreadPool创建的线程池corePoolSize和maximumPoolSize值是相等的(n,n)，它使用的LinkedBlockingQueue；

newSingleThreadExecutor将corePoolSize和maximumPoolSize都设置为1(1,1)，也使用的LinkedBlockingQueue；

newCachedThreadPool将corePoolSize设置为0，将maximumPoolSize设置为Integer.MAX_VALUE，使用的SynchronousQueue，也就是说来了任务就创建线程运行，当线程空闲超过60秒，就销毁线程。

实际中，如果Executors提供的三个静态方法能满足要求，就尽量使用它提供的三个方法，因为自己去手动配置ThreadPoolExecutor的参数有点麻烦，要根据实际任务的类型和数量来进行配置。

1）newFixedThreadPool:（固定线程池）

1.  
   public static ExecutorService newFixedThreadPool(int nThreads) {
2.  
   return new ThreadPoolExecutor(nThreads, nThreads,
3.  
   0L, TimeUnit.MILLISECONDS,
4.  
   new LinkedBlockingQueue<Runnable>());
5.  
   }

线程池corePoolSize和maximumPoolSize值是相等的(n,n),把keepAliveTime设置0L，意味着多余的空闲线程会被立即终止。

newFixedThreadPool的execute方法执行过程：

1，如果当前运行线程数少于corePoolSize，则创建新线程来执行任务（优先满足核心池）

2，当前运行线程数等于corePoolSize时，将任务加入LinkedBlockingQueue链式阻塞队列（核心池满了在进入队列）

3，当线程池的任务完成之后，循环反复从LinkedBlockingQueue队列中获取任务来执行

2）newSingleThreadExecutor：(单线程执行器)

newSingleThreadExecutor是使用单个worker线程的Executors.

1.  
   public static ExecutorService newSingleThreadExecutor() {
2.  
   return new FinalizableDelegatedExecutorService
3.  
   (new ThreadPoolExecutor(1, 1,
4.  
   0L, TimeUnit.MILLISECONDS,
5.  
   new LinkedBlockingQueue<Runnable>()));
6.  
   }

newSingleThreadExecutor的execute方法执行过程如下：

  1，当前运行的线程数少于corePoolSize（即当前线程池中午运行的线程），则创建一个新的线程来执行任务

  2，当线程池中有一个运行的线程时，将任务加入阻塞队列

  3，当线程完成任务时，会无限反复从链式阻塞队列中获取任务来执行

3,）newCachedThreadPool:可缓存线程池

1.  
   public static ExecutorService newCachedThreadPool() {
2.  
   return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
3.  
   60L, TimeUnit.SECONDS,
4.  
   new SynchronousQueue<Runnable>());
5.  
   }

newCachedThreadPool是一个根据需要创建线程的线程池。

newCachedThreadPool的corePoolSize设置0，即核心池是空，maxmumPoolSize设置为Integer.MAX_VALUE,即maxmumPool是无界的。keepAliveTime设置60L,当空闲线程等待新任务最长时间是60s，超过60s就终止

三个线程池的特点：

1、newFixedThreadPool创建一个指定工作线程数量的线程池。每当提交一个任务就创建一个工作线程，如果工作线程数量达到线程池初始的最大数corePoolSize，则将提交的任务存入到池队列中。

2、newCachedThreadPool创建一个可缓存的线程池。这种类型的线程池特点是：
1).工作线程的创建数量几乎没有限制(其实也有限制的,数目为Interger. MAX_VALUE), 这样可灵活的往线程池中添加线程。
2).如果长时间没有往线程池中提交任务，即如果工作线程空闲了指定的时间(默认为1分钟)，则该工作线程将自动终止。终止后，如果你又提交了新的任务，则线程池重新创建一个工作线程。

3、newSingleThreadExecutor创建一个单线程化的Executor，即只创建唯一的工作者线程来执行任务，如果这个线程异常结束，会有另一个取代它，保证顺序执行(我觉得这点是它的特色)。单工作线程最大的特点是可保证顺序地执行各个任务，并且在任意给定的时间不会有多个线程是活动的 

线程池的处理流程：

线程池执行示意图：

1，首先线程池判断基本线程池是否已满（< corePoolSize ？）？没满，创建一个工作线程来执行任务。满了，则进入下个流程。

2，其次线程池判断工作队列是否已满？没满，则将新提交的任务存储在工作队列里。满了，则进入下个流程。

3，最后线程池判断整个线程池是否已满（< maximumPoolSize ？）？没满，则创建一个新的工作线程来执行任务，满了，则交给饱和策略来处理这个任务。

总结：线程池优先要创建出基本线程池大小（corePoolSize）的线程数量，没有达到这个数量时，每次提交新任务都会直接创建一个新线程，当达到了基本线程数量后，又有新任务到达，优先放入等待队列，如果队列满了，才去创建新的线程（不能超过线程池的最大数maxmumPoolSize）

向线程池提交任务的两种方式：

1）通过execute()方法

1.  
   ExecutorService threadpool= Executors.newFixedThreadPool(10);
2.  
   threadpool.execute(new Runnable(){...});

这种方式提交没有返回值，也就不能判断任务是否被线程池执行成功。

2）通过submit()方法

1.  
   Future<?> future = threadpool.submit(new Runnable(){...});
2.  
   try {
3.  
   Object res = future.get();//获取任务执行结果
4.  
   } catch (InterruptedException e) {
5.  
   // 处理中断异常
6.  
   e.printStackTrace();
7.  
   } catch (ExecutionException e) {
8.  
   // 处理无法执行任务异常
9.  
   e.printStackTrace();
10.  
    }finally{
11.  
    // 关闭线程池
12.  
    executor.shutdown();
13.  
    }

使用submit 方法来提交任务，它会返回一个Future对象，通过future的get方法来获取返回值，get方法会阻塞住直到任务完成，而使用get(long timeout, TimeUnit unit)方法则会阻塞一段时间后立即返回，这时有可能任务没有执行完。

线程池的关闭：

• shutdown()：不会立即终止线程池，而是再也不会接受新的任务，要等所有任务缓存队列中的任务都执行完后才终止
• shutdownNow()：立即终止线程池，再也不会接受新的任务，并尝试打断正在执行的任务，并且清空任务缓存队列，返回尚未执行的任务

线程池本身的状态

1.  
   volatile int runState;
2.  
   static final int RUNNING = 0; //运行状态
3.  
   static final int SHUTDOWN = 1; //关闭状态
4.  
   static final int STOP = 2; //停止
5.  
   static final int TERMINATED = 3; //终止，终结

1，当创建线程池后，初始时，线程池处于RUNNING状态；
2，如果调用了shutdown()方法，则线程池处于SHUTDOWN状态，此时线程池不能够接受新的任务，它会等待所有任务执行完毕，最后终止；
3，如果调用了shutdownNow()方法，则线程池处于STOP状态，此时线程池不能接受新的任务，并且会去尝试终止正在执行的任务，返回没有执行的任务列表；
4，当线程池处于SHUTDOWN或STOP状态，并且所有工作线程已经销毁，任务缓存队列已经清空或执行结束后，线程池被设置为TERMINATED状态。

参考:http://blog.csdn.net/shakespeare001/article/details/51330745

http://singleant.iteye.com/blog/1423931

http://blog.csdn.net/it_man/article/details/7193727