一、Executor框架

1.1 介绍

Executor框架包括3大部分:

  • 任务。也就是工作单元,包括被执行任务需要实现的接口:Runnable接口或者Callable接口;
  • 任务的执行。也就是把任务分派给多个线程的执行机制,包括Executor接口及继承自Executor接口的ExecutorService接口。
  • 异步计算的结果。包括Future接口及实现了Future接口的FutureTask类。

Executor框架的成员及其关系可以用一下的关系图表示:

 

1.2 使用流程

Executor框架的使用示意图:

 

使用步骤:

  • 创建Runnable并重写run()方法或者Callable对象并重写call()方法,得到一个任务对象
class callableImp implements Callable<String >{
     
       
	@Override
	public String call() {
     
       
		try{
     
       
			String a = "return String";
			return a;
		}
		catch(Exception e){
     
       
			e.printStackTrace();
			return "exception";
		}
	}
}

  • 创建ExecutorService接口的实现类ThreadPoolExecutor类或者ScheduledThreadPoolExecutor类的对象,然后调用其execute()方法或者submit()方法,提交任务对象执行。
  • 主线程调用Future对象的get()方法获取返回值,或者调用Future对象的cancel()方法取消当前线程的执行。

1.3 成员介绍

Executor框架成员:ThreadPoolExecutor实现类、ScheduledThreadPoolExecutor实现类、Future接口、Runnable和Callable接口、Executors工厂类

Executor:执行器接口,也是最顶层的抽象核心接口, 分离了任务和任务的执行。ExecutorService在Executor的基础上提供了执行器生命周期管理,任务异步执行等功能。

Executors:生产具体的执行器的静态工厂,工具类。

ThreadPoolExecutor:线程池Executor,也是最常用的Executor,通常使用Executors来创建,可以创建三种类型的ThreadPoolExecutor:SingleThreadPoolExecutor,FixedThreadPool和 CachedThreadPool,以线程池的方式管理线程。

ScheduledThreadPoolExecutor:在ThreadPoolExecutor基础上,增加了对周期任务调度的支持。

Runnable和Callable接口:Runnable和Callable接口的实现类,可以被ThreadPoolExecutor和ScheduledThreadPoolExecutor执行,区别是,前没有返回结果,后者可以返回结果。

二、Callable和FutureTask

2.1 介绍

 

Callable与Runnable的区别

1、 Callable带返回值;
2、 会抛出异常;
3、 覆写call()⽅法,⽽不是run()⽅法;

FutureTask的构造需要一个Callable,Callable里有一个call()方法和Runnable的run()方法作用一样,只不过更强,可以返回结果

2.2 Callable接⼝的使⽤

//实现Callable接口
class MyThread implements Callable<Integer> {
   
     
    @Override
    public Integer call() throws Exception {
   
     
        System.out.println("callable come in ...");
        try {
   
      
	        TimeUnit.SECONDS.sleep(3); 
        } catch (InterruptedException e) {
   
     
	        e.printStackTrace(); 
        }
        return 1024;
    }
}

public class CallableDemo {
   
     
    public static void main(String[] args) throws ExecutionException,InterruptedException {
   
     
        //创建FutureTask类,接受MyThread。
        FutureTask<Integer> futureTask = new FutureTask<>(new MyThread());
        //将FutureTask对象放到Thread类的构造器里面。
        new Thread(futureTask, "AA").start();
        int result01 = 100;
        //用FutureTask的get方法得到返回值。
        int result02 = futureTask.get();

        System.out.println("result=" + (result01 + result02));
    }
}

 

如果异步任务还没有完成,futureTask.get()会让当前线程阻塞,直到获得结果。

三、ThreadPoolExecutor

3.1 线程池基本概念

概念

线程池主要是控制运⾏线程的数量,将待处理任务放到等待队列,然后创建线程执⾏这些任务。如果线程数量超过了最⼤数量,超出数量的任务排队等候,等其他线程执⾏完毕,再从队列中取出任务来执⾏。

线程池的主要特点

线程复⽤;控制最⼤并发数;管理线程。

  • 线程复⽤:不⽤⼀直new新线程,重复利⽤已经创建的线程来降低线程的创建和销毁开销,节省系统资源。
  • 提⾼响应速度:当任务达到时,不⽤创建新的线程,直接利⽤线程池的线程。
  • 管理线程:可以控制最⼤并发数,控制线程的创建等。

例如tomcat,web服务器最大并发量就是里面的线程池的线程数量
和火车站卖票窗口类似。

3.2 API介绍

 

线程池解决两个问题:

  • 如果有大量任务需要异步处理,这些任务交给线程处理势必就会创建大量的线程、再去销毁,如果有线程池就可以避免大量线程的创建和销毁切换,节省资源
  • 线程池能解决线程的管理问题。只需要关注业务模块不需要考虑线程的管理问题。
  • 维护一些基本的统计信息。多少任务成功、多少任务异常

3.3 ThreadPoolExecutor构造的七个参数

ThreadPoolExecutor一共提供了4种构造器,但其它三种内部其实都调用了下面的构造器。

/**
* 使用给定的参数创建ThreadPoolExecutor.
*
* @param corePoolSize 核心线程池中的核心线程数
* @param maximumPoolSize 总线程池中的最大线程数
* @param keepAliveTime 空闲线程的存活时间
* @param unit keepAliveTime的单位
* @param workQueue 任务队列, 保存已经提交但尚未被执行的线程
* @param threadFactory 线程创建工厂
* @param handler 拒绝策略 (当任务太多导致工作队列满时的处理策略)
*/
public ThreadPoolExecutor(int corePoolSize,
							int maximumPoolSize,
							long keepAliveTime,
							TimeUnit unit,
							BlockingQueue<Runnable> workQueue,
							ThreadFactory threadFactory,
							RejectedExecutionHandler handler) {
   
     
	if (corePoolSize < 0 ||
		maximumPoolSize <= 0 ||
		maximumPoolSize < corePoolSize ||
		keepAliveTime < 0)
		throw new IllegalArgumentException();
	if (workQueue == null || threadFactory == null || handler == null)
		throw new NullPointerException();
	this.corePoolSize = corePoolSize;
	this.maximumPoolSize = maximumPoolSize;
	this.workQueue = workQueue;
	this.keepAliveTime = unit.toNanos(keepAliveTime);
	this.threadFactory = threadFactory;
	this.handler = handler;
}

参数 意义
corePoolSize 线程池中的常驻核⼼线程数
maximumPoolSize 线程池中能够容纳同时指向的最⼤线程数,此值必须⼤于等于1
keepAliveTime 多余的空闲线程的存活时间,当前池中线程数量超过corePoolSize时,当空闲时间达到keepAliveTime时,多余线程会被销毁直到只剩下corePoolSize个线程为⽌
unit keepAliveTime存活时间的单位
workQueue 任务队列,存放已提交但尚未执⾏的任务
threadFactory 表示⽣成线程池中⼯作线程的线程⼯⼚,⽤于创建线程,⼀般默认的即可
handler 拒绝策略,表示当队列满了,并且⼯作线程⼤于等于线程池的最⼤线程数(maximumPoolSize)时,如何来拒绝请求执⾏的runnable的策略

3.4 线程池的状态流转

线程池状态定义;

/*
* RUNNING -> SHUTDOWN
*     On invocation of shutdown(), perhaps implicitly in finalize()
* (RUNNING or SHUTDOWN) -> STOP
*     On invocation of shutdownNow()
* SHUTDOWN -> TIDYING
*     When both queue and pool are empty
* STOP -> TIDYING
*     When pool is empty
* TIDYING -> TERMINATED
*     When the terminated() hook method has completed
*
*/
private static final int RUNNING = -1 << COUNT_BITS;
private static final int SHUTDOWN = 0 << COUNT_BITS;
private static final int STOP = 1 << COUNT_BITS;
private static final int TIDYING = 2 << COUNT_BITS;
private static final int TERMINATED = 3 << COUNT_BITS;

ThreadPoolExecutor一共定义了5种线程池状态:

  • RUNNING : 接受新任务, 且处理已经进入阻塞队列的任务
  • SHUTDOWN : 不接受新任务, 但处理已经进入阻塞队列的任务
  • STOP : 不接受新任务, 且不处理已经进入阻塞队列的任务, 同时中断正在运行的任务
  • TIDYING : 所有任务都已终止, 工作线程数为0, 线程转化为TIDYING状态并准备调用terminated方法
  • TERMINATED : terminated方法已经执行完成

各个状态之间的流转图:
 

3.5 线程池底层原理

理解线程池七个参数

线程池的创建参数,就像⼀个银⾏。

corePoolSize 就像银⾏的“当值窗⼝“,⽐如今天有2位柜员在受理客户请求(任务)。

如果超过2个客户,那么新的客户就会在等候区(等待队列 workQueue )等待。

等候区也满了,这个时候就要开启“加班窗⼝”,让其它3位柜员来加班,此时达到最⼤窗⼝maximumPoolSize ,为5个。

如果开启了所有窗⼝,等候区依然满员,此时就应该启动”拒绝策略“ handler ,告诉不断涌⼊的客户,叫他们不要进⼊,已经爆满了。

由于不再涌⼊新客户,办完事的客户增多,窗⼝开始空闲,这个时候就通过 keepAlivetTime 将多余的3个”加班窗⼝“取消,恢复到2个”当值窗⼝“。

案例图:

 

这个图有点小问题,应该是3,4,5在候客区,6,7,8在窗口。

线程池流程图解

原理图:上⾯银⾏的例⼦,实际上就是线程池的⼯作原理。

 

流程:

1、 在创建了线程池后,开始等待请求;
2、 当调⽤execute()⽅法添加⼀个请求任务时,线程池会做出如下判断:;
2、 1如果正在运⾏的线程数量⼩于corePoolSize,那么⻢上创建核⼼线程运⾏执⾏这个任务;
2、 2如果正在运⾏的线程数量⼤于或等于corePoolSize,那么将这个任务放⼊队列;
2、 3如果这个时候等待队列已满,且正在运⾏的线程数量⼩于maximumPoolSize,那么还是要创建⾮核⼼线程⽴刻运⾏这个任务;
2、 4如果这个时候等待队列已满,且正在运⾏的线程数量⼤于或等于maximumPoolSize,那么线程池会启动饱和拒绝策略来执⾏;
3、 当⼀个线程完成任务时,它会从等待队列中取出下⼀个任务来执⾏;
4、 当⼀个线程⽆事可做超过⼀定的时间(keepAliveTime)后,线程会判断:;

如果当前运⾏的线程数⼤于 corePoolSize ,那么这个⾮核⼼线程就被停掉。当线程池的所有任务完成后,它最终会收缩到 corePoolSize 的⼤⼩。

执行流程源码分析

执行execute

public void execute(Runnable command) {
   
     
	if (command == null)
		throw new NullPointerException();
	int c = ctl.get();//c是线程池状态
	if (workerCountOf(c) < corePoolSize) {
   
      // CASE1: 工作线程数 < 核心线程池上限
		if (addWorker(command, true)) // 添加工作线程并执行,true代表线程会被放到核心线程池中
			return;
		c = ctl.get();
	}
	
	// 执行到此处, 说明工作线程创建失败 或 工作线程数≥核心线程池上限
	if (isRunning(c) && workQueue.offer(command)) {
   
      // CASE2: 插入任务至队列
		// 成功插入再次检查线程池状态
		int recheck = ctl.get();
		if (!isRunning(recheck) && remove(command))//判断线程池状态,没有运行则移除该任务并执行拒绝策略
			reject(command);
		else if (workerCountOf(recheck) == 0)//检查工作线程是否为0,防止没有线程执行任务
			addWorker(null, false);//添加工作线程
	} else if (!addWorker(command, false)) // CASE3: 插入队列失败,则开启一个工作线程去执行该任务
		reject(command); // 执行拒绝策略
}

execute的整个执行流程关键是下面两点:

  • 如果工作线程数小于核心线程池上限(CorePoolSize),则直接新建一个工作线程并执行任务;
  • 如果工作线程数大于等于CorePoolSize,则尝试将任务加入到队列等待以后执行。如果加入队列失败了(比如队列已满的情况),则在总线程池未满的情况下( CorePoolSize ≤ 工作线程数 <maximumPoolSize )新建一个工作线程立即执行任务,否则执行拒绝策略。

3.6 线程池常用三种创建方式

类似Arrays 、 Collections ⼯具类, Executor 也有⾃⼰的⼯具类 Executors 。通过Executor框架的工具类Executors,可以快速创建三种类型的ThreadPoolExecutor:

注意: Executors中快速创建线程池的方式不只下面三种

FixedThreadPool

可重用固定线程数的线程池:

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

newFixedThreadPool:创建一个固定长度的线程池,每次提交一个任务的时候就会创建一个新的线程,直到达到线程池的最大数量限制。

  • 定长,可以控制线程最大并发数, corePoolSize 和 maximumPoolSize 的数值都是nThreads。
  • 超出线程数的任务会在队列中等待。
  • 工作队列为LinkedBlockingQueue。

创建方法

ExecutorService fixedThreadPool = Executors.newFixedThreadPool(int nThreads);

SingleThreadExecutor

使用单个线程的Executor

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

newSingleThreadExecutor:只创建一个工作线程执行任务,若这个唯一的线程异常故障了,会新建另一个线程来替代,newSingleThreadExecutor可以保证任务依照在工作队列的排队顺序来串行执行。

  • 有且仅有一个工作线程执行任务;
  • 所有任务按照工作队列的排队顺序执行,先进先出的顺序。
  • 工作队列LinkedBlockingQueue。

创建方法

ExecutorService singleThreadPool = Executors.newSingleThreadPool();

CachedThreadPool

会根据需要创建新线程的线程池

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

newCachedThreadPool将创建一个可缓存的线程池,如果当前线程数超过处理任务时,回收空闲线程;当需求增加时,可以添加新线程去处理任务。

特点:

  • 线程数无限制,corePoolSize数值为0, maximumPoolSize 的数值都是为Integer.MAX_VALUE。
  • 若线程未回收,任务到达时,会复用空闲线程;若无空闲线程,则新建线程执行任务。
  • 因为复用性,一定程序减少频繁创建/销毁线程,减少系统开销。
  • 工作队列选用SynchronousQueue。

创建方法

ExecutorService cachedThreadPool = Executors.newCachedThreadPool();

案例演示

/**
 * 线程池代码演示
 */
public class ThreadPoolDemo {
   
     
    public static void main(String[] args) {
   
     
        //System.out.println("=======Fixed Thread Pool========");
        //一个池子有5个工作线程,类似银行有5个受理窗口
        //threadPoolTask( Executors.newFixedThreadPool(5) );

        //System.out.println("======Single Thread Pool=========");
        // //一个池子有1个工作线程,类似银行有1个受理窗口
        //threadPoolTask( Executors.newSingleThreadExecutor() );

        //System.out.println("=====Cached Thread Pool=======");
        // //不定量线程,一个池子有N个工作线程,类似银行有N个受理窗口
        //threadPoolTask( Executors.newCachedThreadPool() );

		//自定义参数线程池
        System.out.println("=====Custom Thread Pool=======");
        threadPoolTask( new ThreadPoolExecutor(
                2,
                5,
                1L,
                TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(3),
                Executors.defaultThreadFactory(),
                new ThreadPoolExecutor.DiscardPolicy()
        ));
    }

    private static void threadPoolTask(ExecutorService threadPool) {
   
     
        //模拟有10个顾客来办理业务
        try {
   
     
            for (int i = 1; i <= 10; i++) {
   
     
                threadPool.execute(() -> {
   
     
                    System.out.println(Thread.currentThread().getName()+"\t办理业务");
                    try {
   
      
	                    TimeUnit.SECONDS.sleep(1); 
                    } catch (InterruptedException e) {
   
     
	                    e.printStackTrace(); 
                    }
                });

            }

        } catch (Exception e) {
   
     
            e.printStackTrace();
        } finally {
   
     
            threadPool.shutdown();
        }
    }
}

newFixedThreadPool:看到最多并且一直只有5个线程在处理

 

newSingleThreadExecutor:看到只有一个线程一直在处理

 

newCachedThreadPool:随着任务压力动态调整线程数量