概述
在java多线程开发中，一般使用线程的时候都是创建一个Thread对象，然后调用start方法执行，这样做没有什么问题，但是如果有多任务并发执行的时候，你可能需要频繁地创建多个线程来执行任务，这样会造成性能方面的问题，体现如下：
1.大量的线程的创建和销毁，本身就是一个很大的性能开销
2.大量线程同时运作的时候，会造成资源紧张，我们知道线程的底层机制就是切分CPU的时间，在大量线程互相抢占资源的时候，可能会造成阻塞现象
基于上面的原因，JDK为我们提供了线程池。线程池可以帮我们管理一定数量的线程，通过重复使用线程来避免大量线程的创建和销毁，从而提高了线程的使用效率
线程池的使用：
关于线程池，首先要了解ThreadPoolExecutor 类，创建线程池就是通过这个类去创建的，它的构造函数如下：
public ThreadPoolExecutor(int corePoolSize, //核心线程数量
int maximumPoolSize, //最大线程数量
long keepAliveTime, //非核心线程的空余存活时间
TimeUnit unit, //存活时间的单位
BlockingQueue<Runnable> workQueue, //保存待执行任务的队列
ThreadFactory threadFactory, //创建新线程使用的工厂
RejectedExecutionHandler handler // 当任务无法执行时的处理器
) {
}
下面来详细解析一下各个参数的含义：
1.corePoolSize （核心线程数量）
默认情况下，核心线程会一直处于存活状态，即使它们是闲置的。如果将ThreadPoolExecutor的allowCoreThreadTimeOut属性设置为true，那么核心线程也会在超时之后结束，超时时间就是参数中设置的keepAlivTime的值
值得注意的是，当线程数量小于核心线程数时，有任务要执行时，即使当前有空闲线程也会创建一个新的核心线程
2.maximumPoolSize（最大线程数量）
包括核心线程数 + 非核心线程数量，如果任务队列满的情况下，并且线程总数小于最大线程数，此时会创建一个新的线程来执行任务
3.keepAliveTime（非核心线程的存活时间）
意思是非核心线程在没有任务执行的情况下的最大的存活时间，也就是说非核心线程在空闲了一定时间以后就会挂掉，当然如果线程池设置 allowCoreThreadTimeOut(true)，核心线程也一样会有这种存活期限
4.unit （存活时间单位）
5.workQueue（任务队列）
线程池中的任务队列，我们提交给线程池的Runnable任务就保存在这个队列中。这是一个BlockQueue类型的结构，属于阻塞队列， 类似于生产者消费者模式，队列中有任务的时候才能进行取出，当队列满的时候，添加也会被阻塞
6.threadFactory（创建线程的工厂）
通过这个工厂类，我们可以给创建线程取名字或者优先级之类的参数
7.handler （任务饱和策略）
主要有四种策略：
CallerRunsPolicy：只要线程池没有关闭，就直接用调用者所在线程来运行任务
AbortPolicy：直接抛出RejectedExecutionException异常
DiscardPolicy：不通知地把任务抛弃了，不干了
DiscardOldestPolicy：把队列中呆了最久的那个任务抛弃了，然后再调用execute方法重试
workQueue（任务队列）的分析：
线程池中使用的队列是BlockQueue接口，常用的实现有以下几种：
1.ArrayBlockingQueue：基于数组，有界，先进先出原则，一般不用
2.LinkedBlockingQueue：基于链表，按先进先出排序，Executors.newFixedThreadPool() 使用的就是个队列
3.SynchronousQueue：这个队列不保存任务，只负责传递，Executors.newCachedThreadPool就使用了这个队列，当有任务到来时如果当前没有空闲的线程，就会创建一个线程来执行新任务
4.PriorityBlockingQueue：具有优先级的任务队列
下面我们封装一个自己的线程池，代码如下：
public class MyThreadPool {
private final String TAG = this.getClass().getSimpleName();
// 核心线程数为 CPU数＊2
private static final int CORE_POOL_SIZE = Runtime.getRuntime().availableProcessors() * 2;
private static final int MAXIMUM_POOL_SIZE = 64; // 线程队列最大线程数
private static final int KEEP_ALIVE_TIME = 1; // 保持存活时间 1秒
//任务队列
private final BlockingQueue<Runnable> mWorkQueue = new LinkedBlockingQueue<>(128);
//创建线程的工厂
private final ThreadFactory DEFAULT_THREAD_FACTORY = new ThreadFactory() {
private final AtomicInteger mCount = new AtomicInteger(1);
public Thread newThread(Runnable r) {
Thread thread = new Thread(r, TAG + " #" + mCount.getAndIncrement());
thread.setPriority(Thread.NORM_PRIORITY);
return thread;
}
};
//创建一个线程池
private ThreadPoolExecutor mExecutor = new ThreadPoolExecutor(CORE_POOL_SIZE, MAXIMUM_POOL_SIZE, KEEP_ALIVE_TIME,
TimeUnit.SECONDS, mWorkQueue, DEFAULT_THREAD_FACTORY,
new ThreadPoolExecutor.DiscardOldestPolicy());
private static volatile MyThreadPool mInstance = new MyThreadPool();
public static MyThreadPool getInstance() {
return mInstance;
}
//执行任务
public void execute(Runnable runnable) {
mExecutor.execute(runnable);
}
@Deprecated
public void shutdownNow() {
mExecutor.shutdownNow();
}
}可以看到，其实创建一个线程池很简单
JDK提供的常见线程池
JDK为我们提供的四种常见线程池的实现，可以用Executors类去创建
1.newFixedThreadExecutor：固定线程数模式
public static ExecutorService newFixedThreadPool(int nThreads) {
return new ThreadPoolExecutor(nThreads, nThreads,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>());
}此线程池的核心线程数等于最大线程数，由构造方法传参决定，任务队列用LinkedBlockingQueue，队列默认容量为Integer.MAX_VALUE，相当于没有上限
这个线程池的执行规则为，当线程数小于核心线程数时，有新任务到来会创建新的线程去执行任务，当线程数达到核心线程数的时候，新任务到来如果暂时没有空闲的线程，就会先加入到阻塞队列中，后面再由线程去队列中取任务执行
此种线程模式由于任务队列的容量巨大，核心线程数可以自定义，所以适合于负载量大的并发执行，当然要根据实际情况设置核心线程数
2.newSingleThreadExecutor：单线程模式
public static ExecutorService newSingleThreadExecutor() {
return new FinalizableDelegatedExecutorService
(new ThreadPoolExecutor(1, 1,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>()));
}
这种模式只有一个核心线程并且最大线程数也是一，所有的任务都由一个线程去执行，这个模式适用于串行执行任务，每个任务需要按顺序执行并且不需要并发执行
3.newCachedThreadPool：线程缓存模式
public static ExecutorService newCachedThreadPool() {
return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
60L, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>());
}此种线程池的核心线程数量为0，但最大线程数是Integer.MAX_VALUE，所以可以创建非常多的非核心线程，最大空闲时间为60秒，队列用的SynchronousQueue，这种队列不保存任务，每当有任务到来的时候，如果没有空闲线程就创建一个新的线程，而线程被空闲60秒以后就会结束。由于可以创建非常多的非核心线程，而且本身的队列不缓存任务并且没有核心线程，所以这种模式首先是适合大量轻任务的并发执行，何谓轻任务，就是任务量体小，执行时间小的任务
4.newScheduledThreadPool：定时线程模式
public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
return new ScheduledThreadPoolExecutor(corePoolSize);
}
//核心线程数是固定的,非核心线程无限大,并且非核心线程数有10s的空闲存活时间
public ScheduledThreadPoolExecutor(int corePoolSize) {
super(corePoolSize, Integer.MAX_VALUE,
DEFAULT_KEEPALIVE_MILLIS, MILLISECONDS,
new DelayedWorkQueue());
}
//super执行这里
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue) {
this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
Executors.defaultThreadFactory(), defaultHandler);
}此线程池的核心线程数是固定的，而非核心线程没有数量上限，非核心线程允许的最大空闲时间为10s。这种线程池适用于执行定时或周期性的任务，常见的心跳包机制可以用这种线程池
其中DelayedWorkQueue是一个延时队列，存入任务的时候会附带一个延时值，代表这个任务要过了这个延时值才能被取出来
线程池提交任务的方式：
1.execute（参数）：提交了任务没有返回值
2.submit（参数）：提交了任务有一个返回值
来看看submit方法的源码：
public Future<?> submit(Runnable task) {
if (task == null) throw new NullPointerException();
RunnableFuture<Void> ftask = newTaskFor(task, null);
execute(ftask);
return ftask;
}其实submit里面也是调用了execute方法，只是将task包装成了RunnableFuture实例，然后返回这个Future对象，通过这个对象我们可以判断任务是否执行成功。获得执行结果调用Future.get()方法，这个方法被线程会被阻塞直到任务完成
关闭线程池
线程池即使不执行任务也会占用一些资源，所以不使用的时候最好关闭线程池
关闭线程池的方法有两种：
1.shutdown（）
public void shutdown() {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
checkShutdownAccess(); //获取权限
advanceRunState(SHUTDOWN); //修改运行状态为SHUTDOWN
interruptIdleWorkers(); //遍历停止未开启的线程
onShutdown(); // 目前空实现
} finally {
mainLock.unlock();
}
tryTerminate();
}
private void interruptIdleWorkers(boolean onlyOne) {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
for (Worker w : workers) { //遍历所有线程
Thread t = w.thread;
//多了一个条件w.tryLock()，表示拿到锁后就中断，因为只有未开启的线程才能拿到锁
//正在运行的线程本来就已经上锁了而且不是可重入锁，所以会获取锁失败
if (!t.isInterrupted() && w.tryLock()) {
try {
t.interrupt();
} catch (SecurityException ignore) {
} finally {
w.unlock();
}
}
if (onlyOne)
break;
}
} finally {
mainLock.unlock();
}
}
将线程设置为SHUTDOWN状态，然后中断尚未开始执行的线程，不能再添加线程了
2.shutdownNow（）
public List<Runnable> shutdownNow() {
List<Runnable> tasks;
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
checkShutdownAccess();
advanceRunState(STOP); //修改状态
interruptWorkers(); //中断所有线程
tasks = drainQueue();
} finally {
mainLock.unlock();
}
tryTerminate();
return tasks;
}
//中断所有线程
private void interruptWorkers() {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
for (Worker w : workers)
w.interruptIfStarted();
} finally {
mainLock.unlock();
}
}设置为STOP状态，尝试中断所有线程，不管有没有在执行
所以说，shutdown和shutdownNow的主要区别是前者中断未执行的线程，后者中断所有线程
线程池使用的总结：
根据前面的介绍，我们小结一下JDK提供的几种线程池的使用策略
1.CachedThreadPool 用于并发执行大量短期的小任务，或者是负载较轻的服务器
2.FixedThreadPool 用于负载比较重的服务器，为了资源的合理利用，需要限制当前线程数量
3.SingleThreadExecutor 用于串行执行任务的场景，每个任务必须按顺序执行，不需要并发执行
4.ScheduledThreadPoolExecutor 用于需要多个后台线程执行周期任务，同时需要限制线程数量的场景
当自定义线程池时，如果是CPU密集型的任务（需要进行大量计算等等），应该尽量配置少的线程数量，一般设置为CPU个数+1个线程，这样可以避免太多线程争抢资源的情况
如果是IO密集型的任务（主要时间在IO工作，因为IO速度远远低于CPU速度，所以CPU空闲多一点），则应该多配置线程数，一般为CPU数的两倍，这样可以充分利用CPU功的能
线程池的源码解析：
当我们用线程池执行任务的时候会调用execute方法或者submit方法，其中submit也是调用了execute方法，所以我们这里先从execute方法的源码开始分析：
public void execute(Runnable command) {
if (command == null)
throw new NullPointerException();
//获取当前线程池的生命周期状态码，其实里面包含了线程池的生命周期和工作线程个数
int c = ctl.get();
//如果当前线程的数量小于核心线程数，则通过addWorker方法创建新的核心线程并执行任务
if (workerCountOf(c) < corePoolSize) {
if (addWorker(command, true))
return;
//创建核心线程失败
c = ctl.get();
}
//如果线程池的线程数量超过了核心线程数量并且线程池处于为可工作状态
//则添加任务到队列
if (isRunning(c) && workQueue.offer(command)) {
int recheck = ctl.get();
//再次判断线程状态，如果非工作状态，则移除任务
if (! isRunning(recheck) && remove(command))
reject(command); //调用拒绝方法
//如果没有工作线程，则单独创建线程,而不指定任务
else if (workerCountOf(recheck) == 0)
addWorker(null, false);
}
//如果线程数量大于等于核心线程数，并且任务队列已经满的情况下
//则创建一个非核心线程来执行任务，创建失败的话调用拒绝方法
else if (!addWorker(command, false))
reject(command);
}执行任务的逻辑代码中已经注释得比较清楚了，其中addWorker方法是创建新线程的核心方法，reject方法是处理创建失败的回调
重点来看看addWorker方法，参数有两个：
Runnable firstTask
需要执行的任务,也可以设置为null(在SHUTDOWN情况下,单纯的创建线程来执行任务)
boolean core
创建的线程是否为核心线程
private boolean addWorker(Runnable firstTask, boolean core) {
//for(;;)是死循环，retry主要用来跳出循环或者跳到retry所在地方重新执行
retry:
for (;;) {
int c = ctl.get();
int rs = runStateOf(c); //线程池生命周期
// Check if queue empty only if necessary.
//线程池在关闭的时候，如果当前任务为null，并且任务队列不为空，则返回false
if (rs >= SHUTDOWN &&
! (rs == SHUTDOWN &&
firstTask == null &&
! workQueue.isEmpty()))
return false;
//for循环创建当前线程是否可行的
for (;;) {
int wc = workerCountOf(c); //工作线程数量
//工作线程数大于最大容量，或者创建核心线程但已经大于等于核心线程数，或者创建
//非核心线程但大于最大允许线程数的时候，都会返回false
if (wc >= CAPACITY ||
wc >= (core ? corePoolSize : maximumPoolSize))
return false;
//设置工作线程数量增加1，如果成功则跳出循环，失败的话重新循环
if (compareAndIncrementWorkerCount(c))
break retry;
c = ctl.get(); // Re-read ctl
if (runStateOf(c) != rs)
continue retry;
// else CAS failed due to workerCount change; retry inner loop
}
}
//下面是创建线程的过程，创建过程中需要加锁保证安全，Worker是线程的包装类
boolean workerStarted = false;
boolean workerAdded = false;
Worker w = null;
try {
w = new Worker(firstTask); //用Worker包装任务
final Thread t = w.thread;
if (t != null) {
//加锁
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
// Recheck while holding lock.
// Back out on ThreadFactory failure or if
// shut down before lock acquired.
int rs = runStateOf(ctl.get());
//有两种情况会创建线程，第一是当前线程池为可工作状态，第二是
//虽然为SHUTDOWN状态但是task为null，此时也还要创建线程
if (rs < SHUTDOWN ||
(rs == SHUTDOWN && firstTask == null)) {
if (t.isAlive()) // precheck that t is startable
throw new IllegalThreadStateException();
//添加worker对象
workers.add(w);
int s = workers.size();
if (s > largestPoolSize)
largestPoolSize = s;
workerAdded = true; //设置为添加成功，即创建线程成功
}
} finally {
//释放锁
mainLock.unlock();
}
if (workerAdded) {
//线程设置为就绪状态
t.start();
workerStarted = true;
}
}
} finally {
if (! workerStarted)
addWorkerFailed(w); //处理创建失败
}
return workerStarted;
}上面是线程池创建新线程的核心代码，其中有一个比较重要的类是Worker，来看看Worker的核心源码：
private final class Worker
extends AbstractQueuedSynchronizer
implements Runnable
{
*********省略代码************
//线程对象
final Thread thread;
//任务对象
Runnable firstTask;
Worker(Runnable firstTask) {
setState(-1); // inhibit interrupts until runWorker
this.firstTask = firstTask;
this.thread = getThreadFactory().newThread(this);
}
/** Delegates main run loop to outer runWorker. */
public void run() {
//线程执行任务
runWorker(this);
}
*******省略代码*******
}抽取一些关键代码，worker类里面包装了线程类thread和任务类runnable，而且worker本身也实现了Runnable接口，也就是说Worker本身也是作为一个任务而存在，worker在作为一个任务放入当前线程thread去执行，当worker任务被执行的时候，会调用一个runWorker方法，可以猜测的是此方法肯定又会执行那个真正的任务firstTask，来看看runWorker方法的源码：
final void runWorker(Worker w) {
Thread wt = Thread.currentThread();
//获取worker的任务，即真正任务
Runnable task = w.firstTask;
w.firstTask = null;
w.unlock(); // allow interrupts
boolean completedAbruptly = true;
try {
while (task != null || (task = getTask()) != null) {
w.lock();
// If pool is stopping, ensure thread is interrupted;
// if not, ensure thread is not interrupted. This
// requires a recheck in second case to deal with
// shutdownNow race while clearing interrupt
if ((runStateAtLeast(ctl.get(), STOP) ||
(Thread.interrupted() &&
runStateAtLeast(ctl.get(), STOP))) &&
!wt.isInterrupted())
wt.interrupt();
try {
beforeExecute(wt, task);
Throwable thrown = null;
try {
//执行真正任务
task.run();
} catch (RuntimeException x) {
thrown = x; throw x;
} catch (Error x) {
thrown = x; throw x;
} catch (Throwable x) {
thrown = x; throw new Error(x);
} finally {
afterExecute(task, thrown);
}
} finally {
task = null;
w.completedTasks++;
w.unlock();
}
}
completedAbruptly = false;
} finally {
processWorkerExit(w, completedAbruptly);
}
}这个方法逻辑很简单，首先是执行当前的真正任务task，执行完以后其实是进入到一个while循环，while循环的条件有一个getTask方法，就是不断从任务队列中取任务，如果不为null则执行，看看getTask方法源码：
private Runnable getTask() {
boolean timedOut = false; // Did the last poll() time out?
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);
// Check if queue empty only if necessary.
if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
decrementWorkerCount();
return null;
}
int wc = workerCountOf(c);
// Are workers subject to culling?
boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
if ((wc > maximumPoolSize || (timed && timedOut))
&& (wc > 1 || workQueue.isEmpty())) {
if (compareAndDecrementWorkerCount(c))
return null;
continue;
}
try {
Runnable r = timed ?
workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
workQueue.take();
if (r != null)
return r;
timedOut = true;
} catch (InterruptedException retry) {
timedOut = false;
}
}
}getTask方法如果返回null的话，则线程会执行完毕，下面是返回null的几种情况：
1）任务队列为空并且是线程池处理非工作状态
2）线程数量大于最大线程数
3）线程已经超时了
返回null的时候，runWorker方法就会跳出循环，结束任务，然后线程结束
去任务队列取任务的时候，因为队列是阻塞队列，所以当队列为空的时候会被阻塞，所以在未超时的情况下，线程池里面的线程也不会马上死亡
线程池的源码分析就到这里吧！！
最后
以上就是斯文音响为你收集整理的深入理解Java多线程-线程池（ThreadPool）线程池的源码解析：的全部内容，希望文章能够帮你解决深入理解Java多线程-线程池（ThreadPool）线程池的源码解析：所遇到的程序开发问题。
如果觉得靠谱客网站的内容还不错，欢迎将靠谱客网站推荐给程序员好友。
发表评论 取消回复