从源码看线程池执行任务的流程及原理

Posted 2020-12-17 shen-qian

1. 线程池的必要性与核心参数

频繁创建、销毁线程的开销过大，所以创建线程频率高的场景一般都选择使用线程池，例如tomcat与客户端通信时处理发来的请求。创建线程池的参数如下

public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,
                            TimeUnit unit,BlockingQueue<Runnable> workQueue,
                            ThreadFactory threadFactory,RejectedExecutionHandler handler)

• corePoolSize：池中核心线程数量，线程数量不大于corePoolSize时线程不会因为空闲时间超过阈值而被销毁。
• maximumPoolSize：最大线程数，当线程数量大于corePoolSize而小于maximumPoolSize时，如果一个线程空闲时间超过阈值，则会被回收。
• keepAliveTime：非核心线程空闲的最大时间，超过这个值就会被回收。
• unit：keepAliveTime参数的时间单位。
• workQueue：用来暂存任务的队列。
• threadFactory：由于提交到线程池的是任务，需要包装成线程来执行，threadFactory就是用来生产线程的。
• handler：当线程池由于各种原因不接受一个任务时，使用这个对象的拒绝方法拒绝，不同的实现类的拒绝策略不同，直接抛异常/重试/无视

根据不同的参数搭配，创建的线程池适应不同的场景，Executors类可以生成几种典型的线程池：

• 固定线程数的线程池：corePoolSize、maximumPoolSize设置成一样，并使用无界的阻塞队列。这样线程池数量从0增加到corePoolSize后就一直保持这个数量，再有任务来时直接加入阻塞队列，反正阻塞队列是无界限的。这种线程池适合并发量较大且波动不大的场景，但是要注意队列过大占用资源的问题。
• 单线程线程池：固定线程数线程池的特殊场景，corePoolSize、maximumPoolSize都设为1即可。适合并发量稳定保持在低水平的场景，也要注意阻塞队列过大的问题。
• 直接提交线程池：corePoolSize设为0，maximumPoolSize设为最大值，阻塞队列是一个直接提交的队列，对于新来的任务每次都建立线程处理。适合并发量很难确定的场景。

2. 执行任务的流程

• 接收到任务后的主流程

public void execute(Runnable command) {
        if (command == null)
            throw new NullPointerException();
        //获取线程池状态及当前线程数记录，高3位记录状态，低29位记录线程数
        int c = ctl.get();   
        //如果线程数小于核心线程阈值
        if (workerCountOf(c) < corePoolSize) {
            //直接包装成Worker运行（true表示将任务以核心线程身份包装）
            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);
        }
        //如果前面加入工作队列失败，则以非核心线程身份包装任务（false标志）
        else if (!addWorker(command, false))
            reject(command);  //如果上一步也失败，则拒绝这个任务
    }

总结：（1）来了新任务，先查看核心线程数是否已经达到，未达到则直接以核心线程的方式新建线程执行任务；（2）如果线程数已经达到核心阈值则将任务加入阻塞队列；（3）如果加入队列失败则以非核心的方式建立线程执行任务；（4）还是失败则拒绝这个任务

• 新建线程执行任务：addWorker(Runnable firstTask, boolean core)的操作

private boolean addWorker(Runnable firstTask, boolean core) {
        /**
        此处省略一段代码，大致工作是检查当前线程数量是否小于阈值（否就返回false），
        CAS的检查线程池状态和CAS的将线程数+1。
        */
        boolean workerStarted = false;
        boolean workerAdded = false;
        Worker w = null;
        try {
 //根据任务包装Worker，Worker对象有一个线程对象也在此时根据Worker对象生成，Worker实现了Runnable接口
            w = new Worker(firstTask);   
            //获取Worker里生成的线程对象
            final Thread t = w.thread;   
            if (t != null) {
                final ReentrantLock mainLock = this.mainLock;
                mainLock.lock();
                try {
                    //获取线程状态
                    int rs = runStateOf(ctl.get());
                    //检查线程池状态的合法性
                    if (rs < SHUTDOWN ||(rs == SHUTDOWN && firstTask == null)) {
                        if (t.isAlive()) // 如果此刻线程就已经活跃，那么就是出错了
                            throw new IllegalThreadStateException();
                        workers.add(w);      //将生成的Worker对象放到Set集合
                        int s = workers.size();
                        if (s > largestPoolSize)
                            largestPoolSize = s;
                        workerAdded = true;  //记录Worker已经成功添加到集合
                    }
                } finally {
                    mainLock.unlock();
                }
                if (workerAdded) {
                    t.start();     //如果Worker对象已经成功添加，则启动这个对象的线程
                    workerStarted = true;  //记录任务已经开始执行
                }
            }
        } finally {
            if (! workerStarted)
                addWorkerFailed(w);   //如果最终发现任务没有标志为开始，则调用响应处理方法
        }
        return workerStarted;
    }

这里要注意Worker本身实现了Ruunable接口有一个run方法，Worker对象持有提交来的task和一个线程对象，这个线程对象根据Worker对象生成所以运行的是Worker的run方法，而Worker的run方法里又是调用了task的run方法。其实就是一个代理模式：Worker对象代理了提交的task。

所以现在知道了t.start()后会运行Worker的run方法，下面看Worker的run方法：

 public void run() {
            runWorker(this);  //调用另一个runWorker方法，并将自己传入
        }

• 真正运行任务的方法：runWorker(Worker w)

final void runWorker(Worker w) {
        Thread wt = Thread.currentThread();
        Runnable task = w.firstTask;  //首先拿到提交的那个task
        w.firstTask = null;
        w.unlock(); // allow interrupts
        boolean completedAbruptly = true;
        try {
            //核心操作就是循环的getTask()，从阻塞队列里拿task来执行
            while (task != null || (task = getTask()) != null) {
                w.lock();
                //检查线程池状态
                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();    //执行真实角色的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);
        }
    }

可以看到，使用Worker的run方法作为线程的运行方法代理task的run方法，主要操作是在运行完task的run方法后再getTask()去队列里获取新任务继续执行。