浅谈线程池的工作原理

喜欢ヅ旅行 2022-01-11 05:33 313阅读 0赞

jdk1.5引入了concurrent包，包中增加了线程池的概念。  
关于线程池的工作原理，作者在execute做了注释。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM5ODM5MDc1_size_16_color_FFFFFF_t_70]  
注释意思为:  
1 .当新任务来时，execute执行，如果池中的线程数低于核心线程，此时就会创建新的线程来处理当前任务。在添加工作线程时，会检查运行状态和工作线程的数量，在addWorker返回false时，不会添加工作线程.。  
2 .即使一个任务可以成功的进入队列，我们仍然需要一个双重检查以便判断是否要新增一个线程(因为在上次检查之后，可能存在线程的死亡)或者线程池已经关闭。所以我们双重检查状态并且在必要的时候进行回滚。  
3 .如果任务不能进入队列，我们就尝试增加一个新的线程。如果失败了，我们就知道线程池已经关闭或者是线程池已经饱和(线程池队列已满且工作线程数已达最大)

线程池相关类或接口有Executor、ExecutorService、AbstractExecutorService、ThreadPoolExecutor、ScheduledThreadPoolExecutor等等。  
在网上查阅相关资料时，看见一个用astah画的类图，使用Idea的Diagrams功能查看了这些类的继承关系，和下图描述的相同。原文地址：https://blog.csdn.net/he90227/article/details/52576452  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM5ODM5MDc1_size_16_color_FFFFFF_t_70 1]  
线程池的工作原理简图如下:  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM5ODM5MDc1_size_16_color_FFFFFF_t_70 2]  
ThreadPoolExecutor的构造函数有多个，参数最全的如下：  
![在这里插入图片描述][20190624224219889.png]  
corePoolSize：核心线程的数量  
maximumPoolSize：最大线程数  
keepAliveTime：非核心线程保活时间。如果设置allowCoreThreadTimeout=true。这样就会让核心线程池中的线程有了存活的时间  
unit：保活时间单位  
workQueue：工作队列  
threadFactory：线程工厂  
handler：拒绝策略  
下面我们一一解读这些参数  
corePoolSize：核心线程的数量。核心线程指，当任务到来时，如果线程池的线程数量低于这个数值，此时不论池子里有没有线程，都会创建一个线程来执行当前任务，这部分线程是核心线程。如果想要一次性创建所有核心线程，可以调用线程池的prestartAllCoreThreads( )方法  
maximumPoolSize：最大线程数。最大线程数是指，当我们的线程数量多于核心线程的设定值，此时会优先入队列。如果队列已满，此时会判断当前线程数量是否低于最大线程数，如果低于，此时会创建新的线程来执行当前任务。  
keepAliveTime：非核心线程保活时间。并不是说先创建的线程就是核心线程，这里的核心线程是指没有任务可运行，并且此时的线程数量高于核心线程数，如果过了保活时间。此时就会销毁线程。所以当前线程如果不想被销毁，就要努力的去队列里获取任务执行。  
workQueue：工作队列。BlockingQueue是一个接口，其实现类比较常用的有以下三种  
ArrayBlockingQueue、LinkedBlockingQueue、SynchronousQueue。

ArryayBlockingQueue是一个有界队列  
LinkedBlockingQueue是一个无界队列，当我们使用这个队列时，最大线程数的设置就没有了意义，因为永远不会填满这个队列。  
SynchronousQueue是一个没有任何容量的队列，来了任务就执行，不存在缓冲这一说法。每一个入队操作必须等到另一个线程调用移除操作，否则入队将一直阻塞  
threadFactory：线程工厂，我们可以自定义一个线程工厂，可以指定线程的名称。  
handler：拒绝策略。当我们的队列已满，并且线程数已达最大线程数，此时再来任务时，就会运行拒绝策略。拒绝策略一共有4种  
1 .AbortPolicy。抛异常RejectedExecutionException（默认方式）  
2 .CallerRunsPolicy。执行execute方法的线程来运行该任务。main线程，也就是主线程执行被队列拒绝的任务。主线程执行完后，队列可能就有位置继续接受任务了。主线程临时执行任务，可能导致主线程阻塞  
3 .DiscardOldestPolicy。丢弃工作队列的最近的一个任务，然后执行当前任务  
4 .DiscardPolicy。直接丢弃任务，不抛异常

关于线程池对线程的复用  
我们首先看一下，线程池到底是怎么执行我们传递进去的任务的。  
线程池执行任务的api是execute方法，源码如下：

public void execute(Runnable command){
            //获取线程池中线程的数量
            int c = ctl.get();
            //判断线程数量是否小于核心线程
            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方法的源码,去掉一些无关代码

private boolean addWorker(Runnable firstTask, boolean core) {
            Worker w = null;
            try {
    	            w = new Worker(firstTask);
    	            final Thread t = w.thread;
    	            t.start();
                }
            } finally {
               
            }
    }

我们可以看到，addWorker方法将当前线程任务当做参数传进Worker对象的构造器中。  
我们看一下Worker的源码，

private final class Worker
            extends AbstractQueuedSynchronizer
            implements Runnable
    Worker(Runnable firstTask) {
                setState(-1); // inhibit interrupts until runWorker
                this.firstTask = firstTask;
                this.thread = getThreadFactory().newThread(this);
    }

我们可以看到，Worker对象就是一个线程。构造器中将当前任务传递给了Worker对象中的firstTask对象  
然后将当前Worker对象传入线程对象，构造成线程  
然后启动Worker线程。此时,Worker对象中的run方法就会启动。  
run方法也挺简单

public void run() {
                runWorker(this);
            }

直接执行了 runWorker方法。  
我们看一下runWorker方法的源码。同样去掉无关代码

final void runWorker(Worker w) {
            Thread wt = Thread.currentThread();
            //取出Worker对象中的任务
            Runnable task = w.firstTask;
            //清空firstTask任务
            w.firstTask = null;
            try {
                //如果task不为空，或者从队列中获取到任务
                while (task != null || (task = getTask()) != null) {
                        //运行当前线程任务
                        task.run();
                }
            } finally {
                //杀死线程
                processWorkerExit(w, completedAbruptly);
            }
        }

首先判断任务如果不为空，就执行任务的run方法。线程的主要逻辑就是run方法，run方法执行完，该条线程的使命就完成了。至此，我们传进去的当前任务就执行完成了。  
但是，线程池的线程复用是咋做的呢？  
答案就在于while循环中的getTask方法。  
我们看一下getTask的源码。去掉无关代码

private Runnable getTask() {
            boolean timedOut = false; // Did the last poll() time out?
                //判断allowCoreThreadTimeout的值
                //判断线程的数量是否大于核心线程的数量
                boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
                try {
                    //如果timed为true，就调用poll方法，否则就调用take方法
                    Runnable r = timed ?
                            workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                            workQueue.take();
                    if (r != null)
                        //从队列中取出任务就返回该任务
                        return r;
                    timedOut = true;
                } catch (InterruptedException retry) {
                    timedOut = false;
                }
        }

当线程数量大于核心线程的数量或者我们设置了allowCoreThreadTimeOut变量为true，此时就会调用poll方法，在keepAliveTime时间内获取线程，即：有超时时间。如果timed为false，就会使用take方法获取线程，获取不到，就在该位置阻塞，直到有任务进入队列。  
至此，我们就明白了。线程池的作者设计了一个循环，不停的从队列中获取任务，只要能从队列中获取到任务，就会一直获取。然后执行线程的run方法。一直使用当前的worker线程执行任务，从而达到了线程的复用。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM5ODM5MDc1_size_16_color_FFFFFF_t_70]: /images/20220111/33131bbc4feb4e4abc2539447a3f257b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM5ODM5MDc1_size_16_color_FFFFFF_t_70 1]: /images/20220111/1b4ab2a0137d425bb8a0a9868fb91bd0.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM5ODM5MDc1_size_16_color_FFFFFF_t_70 2]: /images/20220111/44d0ea0f0301406fb7f9e624394ac2d1.png
[20190624224219889.png]: /images/20220111/b2592fe37fc141f490f896fc061a5cb7.png