java1.5之线程池

小鱼儿 2021-05-12 11:51 404阅读 0赞

Java从1.5版本开始，为简化多线程并发编程，引入全新的并发编程包:java.util.concurrent及其并发编程框架（Executor框架）。

![Center][]

1、Executor接口：线程池的顶级接口，但是严格意义上讲Executor并不是一个线程池，该接口只定义了一个方法execute(Runnable r) ，至于任务怎么执行，什么时候执行，以哪种方式执行，是否有返回值，实在当前线程中执行还是在其他线程中执行，都不做要求。接口最伟大的地方就是把任务的任务的提交和任务的运行解耦开了。

2、ExecutorService接口：真正的线程池接口，继承Executor，扩展了一些方法，其中：  
1. Future<?> submit(Runnable task)：提交Runnable任务到线程池，返回Future对象；  
2. <T> Future<T> submit(Callable<T> task)：提交Callable任务到线程池，返回Future对象；  
3. **invokeAll**(collection of tasks)/invokeAll(collection of tasks, long timeout, TimeUnit unit)：阻塞的方法，其参数就是一组任务，并返回一组Future。参数和返回这2个集合具有相同的结构。invokeAll会按照任务集合中的顺序将所有的Future添加到返回的集合中。只有当所有的任务都执行完毕时，或者调用线程被中断，又或者超出指定时限时，invokeAll方法才会返回。当invokeAll返回之后每个任务要么返回，要么取消，此时客户端可以调用get/isCancelled来判断具体是什么情况。  
4. **invokeAny**(collection of tasks)/invokeAny(collection of tasks, long timeout, TimeUnit unit)：阻塞的方法，不会返回 Future 对象，而是返回集合中某一个Callable 对象的结果，而且无法保证调用之后返回的结果是哪一个 Callable，如果一个任务运行完毕或者抛出异常，方法会取消其它的 Callable 的执行。和invokeAll区别是只要有一个任务执行完了，就把结果返回，并取消其他未执行完的任务；同样，也带有超时功能；  
5. shutdown()：在完成已提交的任务后关闭服务，不再接受新任；  
6. shutdownNow()：停止所有正在执行的任务并关闭服务；  
7. isTerminated()：测试是否所有任务都执行完毕了；  
8. isShutdown()：测试是否该ExecutorService已被关闭。

submit方法都是向线程池中提交任务，它们的区别在于Runnable在执行完毕后没有结果，Callable执行完毕后有一个结果。这在多个线程中传递状态和结果是非常有用的。另外他们的相同点在于都返回一个Future对象。Future对象可以阻塞线程直到运行完毕（获取结果，如果有的话），也可以取消任务执行，当然也能够检测任务是否被取消或者是否执行完毕。  
  
        在没有Future之前我们检测一个线程是否执行完毕通常使用Thread.join()或者用一个死循环加状态位来描述线程执行完毕。现在有了更好的方法能够阻塞线程，检测任务执行完毕甚至取消执行中或者未开始执行的任务。

3、ScheduledExecutorService接口：继承ExecutorService，并且提供了按时间安排执行任务的功能，它提供的方法主要有：  
1. schedule(task, initDelay): 安排所提交的Callable或Runnable任务在initDelay指定的时间后执行；  
2. scheduleAtFixedRate()：安排所提交的Runnable任务按指定的间隔重复执行；  
        3.scheduleWithFixedDelay()：安排所提交的Runnable任务在每次执行完后，等待delay所指定的时间后重复执行；

4、Executors工具类：虽然以上提到的接口有其实现类，但是为了方便，我们建议使用Executors的工具类来得到Executor接口的具体对象，需要注意的是，Executors是一个类，不是Executor的复数形式，Executor提供了一下一些静态方法:  
1. callable(Runnable task):将一个Runnable的任务转化成callable的任务；  
2. public static ExecutorService newSingleThreadExecutor()：创建一个单线程化的Executor。  
3. public static ExecutorService newFixedThreadPool(int nThreads)：创建固定数目线程的线程池。  
4. public static ExecutorService newCachedThreadPool()：创建一个可缓存的线程池，调用execute将重用以前构造的线程（如果线程可用）。如果现有线程没有可用的，则创建一个新线程并添加到池中。终止并从缓存中移除那些已有 60 秒钟未被使用的线程。  
5. public static  ScheduledExecutorService  newSingleThreadScheduledExecutor()：产生一个ScheduledExecutorService对象，这个对象的线程池大小为1，若任务多于一个，任务将按先后顺序执行；  
6. public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize)：创建一个支持定时及周期性的任务执行的线程池，多数情况下可用来替代Timer类。  
  
  
注：submit()和execute()方法区别：  
ExecutorService、ScheduledExecutorService接口的submit()和execute()方法都是把任务提交到线程池中，但二者的区别是  
1. 接收的参数不一样，execute只能接收Runnable类型、submit可以接收Runnable和Callable两种类型；  
2. submit有返回值，而execute没有返回值；  
3. submit方便Exception处理；

5、ThreadPoolExecutor和ScheduledThreadPoolExecutor 实现类：          这两个类分别实现了ExecutorService和ScheduledExecutorService接口，通过他们可以灵活的构造出一个线程池对象。根据上面的类图可以知道Executors工具类关联了这两个实现类，从源码也可以看出来Executors工具类内部就是通过这两个实现类来默认构造的线程池对象。  
public static ExecutorService newCachedThreadPool() \{  
        return new ThreadPoolExecutor(0, Integer.MAX\_VALUE,  
                                      60L, TimeUnit.SECONDS,  
                                      new SynchronousQueue<Runnable>());  
\}  
ThreadPoolExecutor构造函数：  
public ThreadPoolExecutor(     
            int corePoolSize,     
            int maximumPoolSize,     
            long keepAliveTime,     
            TimeUnit unit,     
            BlockingQueue<Runnable>  workQueue,  
RejectedExecutionHandler  handler)    
参数含义：  
corePoolSize 指的是保留的线程池大小。   
maximumPoolSize 指的是线程池的最大大小。   
keepAliveTime 指的是空闲线程结束的超时时间。   
unit 是一个枚举，表示 keepAliveTime 的单位。   
workQueue 表示存放任务的队列。  
Handler 表示线程池对拒绝任务的处理策略。  
  
  
我们可以从线程池的工作过程中了解这些参数的意义。线程池的工作过程如下：  
• 线程池刚创建时，里面没有一个线程。任务队列是作为参数传进来的。不过，就算队列里面有任务，线程池也不会马上执行它们。  
• 当调用 execute() 方法添加一个任务时，线程池会做如下判断：(一个任务通过 execute(Runnable)方法被添加到线程池，任务就是一个 Runnable类型的对象，任务的执行方法就是 Runnable类型对象的run()方法。)  
      a. 如果正在运行的线程数量小于 corePoolSize，那么马上创建线程运行这个任务；  
      b. 如果正在运行的线程数量大于或等于 corePoolSize，那么将这个任务放入队列。  
      c. 如果这时候队列满了，而且正在运行的线程数量小于 maximumPoolSize，那么还是要创建线程运行这个任务；  
      d. 如果队列满了，而且正在运行的线程数量大于或等于 maximumPoolSize，那么线程池会抛出异常，告诉调用者“我不能再接受任务了”。  
• 当一个线程完成任务时，它会从队列中取下一个任务来执行。  
• 当一个线程无事可做，超过一定的时间（keepAliveTime）时，线程池会判断，如果当前运行的线程数大于 corePoolSize，那么这个线程就被停掉。所以线程池的所有任务完成后，它最终会收缩到 corePoolSize 的大小。  
  
这样的过程说明，并不是先加入任务就一定会先执行。假设队列大小为 10，corePoolSize 为 3，maximumPoolSize 为 6，那么当加入 20 个任务时，执行的顺序就是这样的：首先执行任务 1、2、3，然后任务 4~13 被放入队列。这时候队列满了，任务 14、15、16 会被马上执行，而任务 17~20 则会抛出异常。最终顺序是：1、2、3、14、15、16、4、5、6、7、8、9、10、11、12、13。

6、ExecutorCompletionService：  
          使用ExecutorService类的时候，我们常维护一个list保存submit的callable task所返回的Future对象。然后在主线程中遍历这个list并调用Future的get()方法取到Task的返回值。  
       CompletionService和ExecutorService的主要的区别在于submit的task不一定是按照加入自己维护的list顺序完成的。  
       ExecutorService中从list中遍历的每个Future对象并不一定处于完成状态，这时调用get()方法就会被阻塞住，如果系统是设计成每个线程完成后就能根据其结果继续做后面的事，这样对于处于list后面的但是先完成的线程就会增加了额外的等待时间。  
  
  
         而CompletionService的实现是维护一个保存Future对象的BlockingQueue。只有当这个Future对象状态是结束的时候，才会加入到这个Queue中，take()方法其实就是Producer-Consumer中的Consumer。它会从Queue中取出Future对象，如果Queue是空的，就会阻塞在那里，直到有完成的Future对象加入到Queue中。所以，先完成的必定先被取出。这样就减少了不必要的等待时间。

[Center]: /images/20210511/2340f25a1cea4a3599de4656db612572.png