线程池Executor

1.为何需要线程池

在java中，使用线程来执行异步任务时，线程的创建和销毁需要一定的开销，如果我们为每一个任务创建一个新的线程来执行的话，那么这些线程的创建与销毁将消耗大量的计算资源。同时为每一个任务创建一个新线程来执行，这样的方式可能会使处于高负荷状态的应用最终崩溃。所以线程池的出现为解决这个问题带来曙光。我们将在线程池中创建若干条线程，当有任务需要执行时就从该线程池中获取一条线程来执行任务，如果一时间任务过多，超出线程池的线程数量，那么后面的线程任务就进入一个等待队列进行等待，直到线程池有线程处于空闲时才从等待队列获取要执行的任务进行处理，以此循环…这样就大大减少了线程创建和销毁的开销，也会缓解我们的应用处于超负荷时的情况。

1.1Executor框架的两级调度模型

在java线程启动时会创建一个本地操作系统线程，当该java线程终止时，这个操作系统线程也会被回收。而每一个java线程都会被一对一映射为本地操作系统的线程，操作系统会调度所有的线程并将它们分别给可用的CPU。而所谓的映射方式是这样实现的，在上层，java多线程程序通过把应用分为若干个任务，然后使用用户级的调度器（Executor框架）将这些任务映射为固定数量的线程；在底层，操作系统内核将这些线程映射到硬件处理器上。这样种两级调度模型如下图所示：
线程池Executor
从图中我们可以看出，应用程序通过Executor框架控制上层的调度，而下层的调度由操作系统内核控制，下层的调度不受应用程序的控制。

1.2 Executor框架的结构

Executor框架的结构主要包括3个部分
1.任务：包括被执行任务需要实现的接口：Runnable接口或Callable接口
2.任务的执行：包括任务执行机制的核心接口Executor，以及继承自Executor的EexcutorService接口。Exrcutor有两个关键类实现了ExecutorService接口（ThreadPoolExecutor和ScheduledThreadPoolExecutor）。
3.异步计算的结果：包括接口Future和实现Future接口的FutureTask类
下面我们通过一个UML图来认识一下这些类间的关系：
线程池Executor
Extecutor是一个接口，它是Executor框架的基础，它将任务的提交与任务的执行分离开来。
ThreadPoolExecutor是线程池的核心实现类，用来执行被提交的任务。
ScheduledThreadPoolExecutor是一个实现类，可以在给定的延迟后运行命令，或者定期执行命令。ScheduledThreadPoolExecutor比Timer更灵活，功能更强大。
Future接口和实现Future接口的FutureTask类，代表异步计算的结果。
Runnable接口和Callable接口的实现类，都可以被ThreadPoolExecutor或者ScheduledThreadPoolExecutor执行。区别就是Runnable无法返回执行结果，而Callable可以返回执行结果。
线程池Executor
分析说明：
主线程首先创建实现Runnable或Callable接口的任务对象，工具类Executors可以把一个Runnable对象封装为一个Callable对象,使用如下两种方式：
Executors.callable(Runnable task)或者Executors.callable(Runnable task,Object resule)。
然后可以把Runnable对象直接提交给ExecutorService执行，方法为ExecutorService.execute(Runnable command)；或者也可以把Runnable对象或者Callable对象提交给ExecutorService执行，方法为ExecutorService.submit(Runnable task)或ExecutorService.submit(Callable task)。这里需要注意的是如果执行ExecutorService.submit(…),ExecutorService将返回一个实现Future接口的对象（其实就是FutureTask）。当然由于FutureTask实现了Runnable接口，我们也可以直接创建FutureTask，然后提交给ExecutorService执行。到此Executor框架的主要体系结构我们都介绍完了，我们对此有了大概了解后，下面我们就重点聊聊两个主要的线程池实现类。

2.ThreadPoolExecutor浅析

ThreadPoolExecutor是线程的真正实现，通常使用工厂类Executors来创建，但它的构造方法提供了一系列参数来配置线程池，下面我们就先介绍ThreadPoolExecutor的构造方法中各个参数的含义。


public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue,
                              ThreadFactory threadFactory) {
        this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
             threadFactory, defaultHandler);
    }

corePoolSize：线程池的核心线程数，默认情况下，核心线程数会一直在线程池中存活，即使它们处理闲置状态。如果将ThreadPoolExecutor的allowCoreThreadTimeOut属性设置为true，那么闲置的核心线程在等待新任务到来时会执行超时策略，这个时间间隔由keepAliveTime所指定，当等待时间超过keepAliveTime所指定的时长后，核心线程就会被终止。
maximumPoolSize：线程池所能容纳的最大线程数量，当活动线程数到达这个数值后，后续的新任务将会被阻塞。
keepAliveTime：非核心线程闲置时的超时时长，超过这个时长，非核心线程就会被回收。当ThreadPoolExecutor的allowCoreThreadTimeOut属性设置为true时，keepAliveTime同样会作用于核心线程。
unit：用于指定keepAliveTime参数的时间单位，这是一个枚举，常用的有TimeUnit.MILLISECONDS(毫秒)，TimeUnit.SECONDS(秒)以及TimeUnit.MINUTES(分钟)等。
workQueue：线程池中的任务队列，通过线程池的execute方法提交Runnable对象会存储在这个队列中。
threadFactory：线程工厂，为线程池提供创建新线程的功能。ThreadFactory是一个接口，它只有一个方法：Thread newThread（Runnable r）。
除了上面的参数外还有个不常用的参数，RejectExecutionHandler，这个参数表示当ThreadPoolExecutor已经关闭或者ThreadPoolExecutor已经饱和时（达到了最大线程池大小而且工作队列已经满），execute方法将会调用Handler的rejectExecution方法来通知调用者，默认情况下是抛出一个RejectExecutionException异常。了解完相关构造函数的参数，我们再来看看ThreadPoolExecutor执行任务时的大致规则：
（1）如果线程池的数量还未达到核心线程的数量，那么会直接启动一个核心线程来执行任务
（2）如果线程池中的线程数量已经达到或者超出核心线程的数量，那么任务会被插入到任务队列中排队等待执行。
（3）如果在步骤（2）中无法将任务插入到任务队列中，这往往是由于任务队列已满，这个时候如果线程数量未达到线程池规定的最大值，那么会立刻启动一个非核心线程来执行任务。
（4）如果在步骤（3）中线程数量已经达到线程池规定的最大值，那么就会拒绝执行此任务，ThreadPoolExecutor会调用RejectExecutionHandler的rejectExecution方法来通知调用者。
到此ThreadPoolExecutor的详细配置了解完了，ThreadPoolExecutor的执行规则也了解完了，那么接下来我们就来介绍3种常见的线程池，它们都直接或者间接地通过配置ThreadPoolExecutor来实现自己的功能特性，这个3种线程池分别是FixedThreadPool，CachedThreadPool，以及SingleThreadExecutor。

2.1FixedThreadPool

FixedThreadPool模式会使用一个优先固定数目的线程来处理若干数目的任务。规定数目的线程处理所有任务

ExecutorService fixedThreadPool=Executors.newFixedThreadPool(5);

FixedThreadPool创建方法源码：

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

FixedThreadPool的execute()方法的运行流程：
线程池Executor
线程在执行完图中的1后，会在循环中反复从LinkedBlockingQueue获取任务来执行。
这里还有点要说明的是FixedThreadPool使用的是无界队列LinkedBlockingQueue作为线程池的工作队列

2.2CachedThreadPool

CachedThreadPool首先会按照需要创建足够多的线程来执行任务(Task)。随着程序执行的过程，有的线程执行完了任务，可以被重新循环使用时，才不再创建新的线程来执行任务。创建方式：

ExecutorService cachedThreadPool=Executors.newCachedThreadPool();

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

从该静态方法，我们可以看到CachedThreadPool的corePoolSize被设置为0，而maximumPoolSize被设置Integer.MAX_VALUE，即maximumPoolSize是无界的，而keepAliveTime被设置为60L，单位为妙。也就是空闲线程等待时间最长为60秒，超过该时间将会被终止。而且在这里CachedThreadPool使用的是没有容量的SynchronousQueue作为线程池的工作队列，但其maximumPoolSize是无界的，也就是意味着如果主线程提交任务的速度高于maximumPoolSize中线程处理任务的速度时CachedThreadPool将会不断的创建新的线程，在极端情况下，CachedThreadPool会因为创建过多线程而耗尽CPU和内存资源。CachedThreadPool的execute()方法的运行流程：
线程池Executor

2.3SingleThreadExecutor

SingleThreadExecutor模式只会创建一个线程。它和FixedThreadPool比较类似，不过线程数是一个。如果多个任务被提交给SingleThreadExecutor的话，那么这些任务会被保存在一个队列中，并且会按照任务提交的顺序，一个先执行完成再执行另外一个线程。SingleThreadExecutor模式可以保证只有一个任务会被执行。这种特点可以被用来处理共享资源的问题而不需要考虑同步的问题。
创建方式：

ExecutorService singleThreadExecutor=Executors.newSingleThreadExecutor();

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

线程池Executor

2.4 各自的适用场景

FixedThreadPool：适用于为了满足资源管理需求，而需要限制当前线程的数量的应用场景，它适用于负载比较重的服务器。
SingleThreadExecutor：适用于需要保证执行顺序地执行各个任务；并且在任意时间点，不会有多个线程是活动的场景。
CachedThreadPool：大小无界的线程池，适用于执行很多的短期异步任务的小程序，或者负载较轻的服务器。

3.ScheduledThreadPoolExecutor

ScheduledThreadPoolExecutor继承自ThreadPoolExecutor。它主要用来在给定的延迟之后执行任务，或者定期执行任务。ScheduledThreadPoolExecutor的功能与Timer类似，但比Timer更强大，更灵活，Timer对应的是单个后台线程，而ScheduledThreadPoolExecutor可以在构造函数中指定多个对应的后台线程数。接下来我们先来了解一下ScheduledThreadPoolExecutor的运行机制：
线程池Executor

3.3 ScheduledThreadPoolExecutor和SingleThreadScheduledExecutor的适用场景

ScheduledThreadPoolExecutor：适用于多个后台线程执行周期性任务，同时为了满足资源管理的需求而需要限制后台线程数量的应用场景。
SingleThreadScheduledExecutor：适用于需要单个后台线程执行周期任务，同时需要保证任务顺序执行的应用场景。

3.4 ScheduledThreadPoolExecutor使用案例

我们创建一个Runnable的对象，然后使用ScheduledThreadPoolExecutor的Scheduled()来执行延迟任务，输出执行时间即可:
我们先来介绍一下该类延迟执行的方法：

public ScheduledFuture<?> schedule(Runnable command,long delay, TimeUnit unit);

参数解析：
command：就是一个实现Runnable接口的类
delay：延迟多久后执行。
unit：用于指定keepAliveTime参数的时间单位，这是一个枚举，常用的有TimeUnit.MILLISECONDS(毫秒)，TimeUnit.SECONDS(秒)以及TimeUnit.MINUTES(分钟)等。
这里要注意这个方法会返回ScheduledFuture实例，可以用于获取线程状态信息和延迟时间

例子:


public class WorkerThread implements Runnable{
	@Override
	public void run() {
		 System.out.println(Thread.currentThread().getName()+" Start. Time = "+getNowDate());
	     threadSleep();
	     System.out.println(Thread.currentThread().getName()+" End. Time = "+getNowDate());
		
	}
	/**
	 * 睡3秒
	 */
	public void threadSleep(){
		try {
			Thread.sleep(3000);
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}
	 /**
	  * 获取现在时间
	  * 
	  * @return 返回时间类型 yyyy-MM-dd HH:mm:ss
	  */
	public static String getNowDate() {
		  Date currentTime = new Date();
		  SimpleDateFormat formatter; 
		    formatter = new SimpleDateFormat ("yyyy-MM-dd HH:mm:ss"); 
		    String ctime = formatter.format(currentTime); 
		  return ctime;
		 }
}

执行类

public class ScheduledThreadPoolTest {
	
	public static void main(String[] args) {
		ScheduledExecutorService scheduledThreadPool = Executors.newScheduledThreadPool(5);
		 try {
	    //schedule to run after sometime
	    System.out.println("Current Time = "+getNowDate());
	    for(int i=0; i<3; i++){
	        Thread.sleep(1000);
	        WorkerThread worker = new WorkerThread();
	        //延迟10秒后执行
	        scheduledThreadPool.schedule(worker, 10, TimeUnit.SECONDS);
	    }
			Thread.sleep(3000);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	    scheduledThreadPool.shutdown();
	    while(!scheduledThreadPool.isTerminated()){
	        //wait for all tasks to finish
	    }
	    System.out.println("Finished all threads");
	}
	
	 /**
	  * 获取现在时间
	  * 
	  * @return 返回时间类型 yyyy-MM-dd HH:mm:ss
	  */
	 public static String getNowDate() {
	  Date currentTime = new Date();
	  SimpleDateFormat formatter; 
	    formatter = new SimpleDateFormat ("yyyy-MM-dd HH:mm:ss"); 
	    String ctime = formatter.format(currentTime); 
	  return ctime;
	 }
}

运行结果：
线程池Executor

3.5 scheduleWithFixedDelay实例

 public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command,long initialDelay,long delay,TimeUnit unit);

scheduleWithFixedDelay方法的作用是预定在初始的延迟结束后周期性地执行给定任务，在一次调用完成和下一次调用开始之间有长度为delay的延迟，其中initialDelay为初始延迟


public class ScheduledTask {
	public ScheduledThreadPoolExecutor se = new ScheduledThreadPoolExecutor(5);
	public static void main(String[] args) {
		new ScheduledTask();
	}
	public void fixedPeriodSchedule() {
		// 设定可以循环执行的runnable,初始延迟为0，这里设置的任务的间隔为5秒
		for(int i=0;i<5;i++){
			se.scheduleAtFixedRate(new FixedSchedule(), 0, 5, TimeUnit.SECONDS);
		}
	}
	public ScheduledTask() {
		fixedPeriodSchedule();
	}
	class FixedSchedule implements Runnable {
		public void run() {
			System.out.println("当前线程："+Thread.currentThread().getName()+"  当前时间："+new Date(System.currentTimeMillis()));
		}
	}
}

运行结果:
线程池Executor

线程池Executor

1.为何需要线程池

1.1Executor框架的两级调度模型

1.2 Executor框架的结构

2.ThreadPoolExecutor浅析

2.1FixedThreadPool

2.2CachedThreadPool

2.3SingleThreadExecutor

2.4 各自的适用场景

3.ScheduledThreadPoolExecutor

3.3 ScheduledThreadPoolExecutor和SingleThreadScheduledExecutor的适用场景

3.4 ScheduledThreadPoolExecutor使用案例

3.5 scheduleWithFixedDelay实例

相关推荐