Java 线程池分析

Posted 2020-10-01 qiezijiajia

在项目中经常会用到java线程池，但是别人问起线程池的原理，线程池的策略怎么实现的？ 答得不太好，所以按照源码分析一番，首先看下最常用的线程池代码：

public class ThreadPoolTest {

    private static Executor executor= Executors.newFixedThreadPool(10);  //一般通过fixThreadPool起线程池

    public static void main(String[] args){

        for(int i=0;i<50;i++){

            executor.execute(new Task());  //通过线程池执行task
        }
    }

    static class Task implements Runnable{  //Task任务，打印当前执行该任务的线程名称


        @Override
        public void run() {

            System.out.println(Thread.currentThread().getName());
        }
    }
}

这个输出结果为，可以看到，没有线程池的超过10的，所以线程池执行只用了10个线程：

pool-1-thread-1
pool-1-thread-2
pool-1-thread-3
pool-1-thread-4
pool-1-thread-5
pool-1-thread-6
pool-1-thread-7
pool-1-thread-8
pool-1-thread-9
pool-1-thread-10
pool-1-thread-10
pool-1-thread-2
pool-1-thread-3
pool-1-thread-5
pool-1-thread-2
pool-1-thread-4
pool-1-thread-2
pool-1-thread-3
pool-1-thread-2

 

接下来分析一下源码，在newFixedThreadPool(10)的时候，源码是这样的，返回一个ThreadPoolExecutor

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

再看下ThreadPoolExecutor的构造方法，如下，参数说明我标在注释中了，默认参数见上面传的。所以在newFixedThreadPool（10）时，其实只传了一个参数，其他的都为默认。

public ThreadPoolExecutor(int corePoolSize,  //线程池核心池大小，当提交一个任务时，线程池会新创建一个线程池来执行任务，直到线程池大小等于该值。等于后继续提交的任务被保存在阻塞队列，等待执行。
                              int maximumPoolSize, //最大线程池大小，当阻塞队列满后，如果继续提交任务，则创建新的线程执行任务，前提是当前线程数小于maximumPoolSize；
                              long keepAliveTime, //线程池keepalive时间，即空闲线程池的存活时间。
                              TimeUnit unit,  //keepAliveTime的单位
                              BlockingQueue<Runnable> workQueue) //阻塞队列，比如实现Runnable接口{
        this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
             Executors.defaultThreadFactory(), defaultHandler);
    }

说明：在JDK中提供了如下阻塞队列：

1、ArrayBlockingQueue：基于数组结构的有界阻塞队列，按FIFO排序任务；
2、LinkedBlockingQuene：基于链表结构的阻塞队列，按FIFO排序任务，吞吐量通常要高于ArrayBlockingQuene；
3、SynchronousQuene：一个不存储元素的阻塞队列，每个插入操作必须等到另一个线程调用移除操作，否则插入操作一直处于阻塞状态，吞吐量通常要高于LinkedBlockingQuene；
4、priorityBlockingQuene：具有优先级的无界阻塞队列；

 

接下来执行this构造方法，查看this构造方法源码，多了两个参数，一个为ThreadFactory（线程池工厂方法），另一个为RejectedExecutionHandler（饱和策略），上面传入的参数为Executors.defaultThreadFactory()和defaultHandler，可以看到defaultThreadFactory方法如下，defaultHandler的策略为abort。

    static class DefaultThreadFactory implements ThreadFactory {  //默认线程池工厂方法
        private static final AtomicInteger poolNumber = new AtomicInteger(1);
        private final ThreadGroup group;
        private final AtomicInteger threadNumber = new AtomicInteger(1);
        private final String namePrefix;

        DefaultThreadFactory() {  //设置线程名，group等
            SecurityManager s = System.getSecurityManager();
            group = (s != null) ? s.getThreadGroup() :
                                  Thread.currentThread().getThreadGroup();
            namePrefix = "pool-" +
                          poolNumber.getAndIncrement() +
                         "-thread-";
        }

 

private static final RejectedExecutionHandler defaultHandler =  //默认饱和策略
        new AbortPolicy();

 

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;
    }

 

了解源码后，接下来要验证一个事情，即提交任务后线程池的执行策略及饱和策略

我们将最开始部分的代码的线程池，自己通过ThreadPoolExecutor的实现，只修改了线程池的实例部门，将maximumPoolSize设置为20，同时将阻塞队列的大小改为长度为3的ArrayBlockingQueue，饱和策略默认不传。这时候还是去提交50个任务。理论上，我们应该会创建20个线程。

public class ThreadPoolTest {
    private static Executor executor1= new ThreadPoolExecutor(10,20,0, TimeUnit.MILLISECONDS,new ArrayBlockingQueue<Runnable>(3));

    public static void main(String[] args){

        for(int i=0;i<50;i++){

            executor.execute(new Task());
        }
    }

    static class Task implements Runnable{


        @Override
        public void run() {

            System.out.println(Thread.currentThread().getName());
        }
    }
}

执行结果如下，确实看到了20个线程的名称：

pool-1-thread-1
pool-1-thread-3
pool-1-thread-2
pool-1-thread-5
pool-1-thread-4
pool-1-thread-6
pool-1-thread-7
pool-1-thread-8
pool-1-thread-9
pool-1-thread-10
pool-1-thread-1
pool-1-thread-1
pool-1-thread-1
pool-1-thread-1
pool-1-thread-11
pool-1-thread-12
pool-1-thread-12
pool-1-thread-12
pool-1-thread-12
pool-1-thread-12
pool-1-thread-13
pool-1-thread-14
pool-1-thread-14
pool-1-thread-14
pool-1-thread-14
pool-1-thread-15
pool-1-thread-16
pool-1-thread-16
pool-1-thread-16
pool-1-thread-16
pool-1-thread-17
pool-1-thread-17
pool-1-thread-17
pool-1-thread-17
pool-1-thread-18
pool-1-thread-18
pool-1-thread-18
pool-1-thread-18
pool-1-thread-19
pool-1-thread-19
pool-1-thread-19
pool-1-thread-19
pool-1-thread-20
pool-1-thread-20
pool-1-thread-20
pool-1-thread-20

 

再接下来我们将maximumPoolSize也改为10，再执行代码，这个时候应该会出现任务提交失败的情况；因为阻塞队列也满了，线程池最多只能创建10个，提交的任务有50个，处理不过来。

结果如下，可以看到完成的task为13个，即10个线程池大小+长度为3的有界阻塞队列，最后使用的策略为abort策略：

pool-1-thread-1
pool-1-thread-3
pool-1-thread-2
pool-1-thread-4
pool-1-thread-5
pool-1-thread-6
pool-1-thread-7
pool-1-thread-8
pool-1-thread-9
pool-1-thread-9
pool-1-thread-9
pool-1-thread-9
pool-1-thread-10
Exception in thread "main" java.util.concurrent.RejectedExecutionException: Task [email protected] rejected from [email protected][Running, pool size = 10, active threads = 0, queued tasks = 0, completed tasks = 13]
at java.util.concurrent.ThreadPoolExecutor$AbortPolicy.rejectedExecution(ThreadPoolExecutor.java:2047)

 

 

 

线程池的源码及调度流程分析完成了，接下来看下线程池的executor怎么去执行的
先看下线程的状态

其中AtomicInteger变量ctl的功能非常强大：利用低29位表示线程池中线程数，通过高3位表示线程池的运行状态：
1、RUNNING：-1 << COUNT_BITS，即高3位为111，该状态的线程池会接收新任务，并处理阻塞队列中的任务；
2、SHUTDOWN： 0 << COUNT_BITS，即高3位为000，该状态的线程池不会接收新任务，但会处理阻塞队列中的任务；
3、STOP ： 1 << COUNT_BITS，即高3位为001，该状态的线程不会接收新任务，也不会处理阻塞队列中的任务，而且会中断正在运行的任务；
4、TIDYING ： 2 << COUNT_BITS，即高3位为010；
5、TERMINATED： 3 << COUNT_BITS，即高3位为011；

public void execute(Runnable command) {
        if (command == null)
            throw new NullPointerException();
        /*
         * Proceed in 3 steps:
         *
         * 1. If fewer than corePoolSize threads are running, try to
         * start a new thread with the given command as its first
         * task.  The call to addWorker atomically checks runState and
         * workerCount, and so prevents false alarms that would add
         * threads when it shouldn‘t, by returning false.
         *
         * 2. If a task can be successfully queued, then we still need
         * to double-check whether we should have added a thread
         * (because existing ones died since last checking) or that
         * the pool shut down since entry into this method. So we
         * recheck state and if necessary roll back the enqueuing if
         * stopped, or start a new thread if there are none.
         *
         * 3. If we cannot queue task, then we try to add a new
         * thread.  If it fails, we know we are shut down or saturated
         * and so reject the task.
         */
        int c = ctl.get();  //获取ctl的低29位
        if (workerCountOf(c) < corePoolSize) { //workerCountOf方法根据ctl的低29位，得到线程池的当前线程数，如果当前的线程数小于核心池大小，则直接addWorker
            if (addWorker(command, true))
                return;
            c = ctl.get();
        }
        if (isRunning(c) && workQueue.offer(command)) { //如果当前线程池处于Running状态（即能够接受任务，处理阻塞队列中的任务），且把任务提交给了阻塞队列
            int recheck = ctl.get();
            if (! isRunning(recheck) && remove(command))  //再次检查线程池的状态，如果线程池没有RUNNING（即无法放入队列），并且将任务从队列移除，交给reject处理
                reject(command);
            else if (workerCountOf(recheck) == 0)  //否则，如果线程池数为0，则添加worker，启动一个null任务的线程
                addWorker(null, false);
        }
        else if (!addWorker(command, false))   //否则如果添加addWork失败，则reject。
            reject(command);
    }

 

addWorker主要负责创建新的线程并执行任务，主代码如下：

boolean workerStarted = false;
        boolean workerAdded = false;
        Worker w = null;
        try {
            w = new Worker(firstTask);
            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());

                    if (rs < SHUTDOWN ||
                        (rs == SHUTDOWN && firstTask == null)) {
                        if (t.isAlive()) // precheck that t is startable
                            throw new IllegalThreadStateException();
                        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;

 

线程池的工作线程通过Woker类实现，在ReentrantLock锁的保证下，把Woker实例插入到HashSet后，并启动Woker中的线程，其中Worker类设计如下：
1、继承了AQS类，可以方便的实现工作线程的中止操作；
2、实现了Runnable接口，可以将自身作为一个任务在工作线程中执行；
3、当前提交的任务firstTask作为参数传入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);
        }

        /** Delegates main run loop to outer runWorker  */
        public void run() {
            runWorker(this);
        }

线程工厂在创建线程thread时，将Woker实例本身this作为参数传入，当执行start方法启动线程thread时，本质是执行了Worker的runWorker方法。

 

 

在线程池执行任务的时候，除了executor还有一种方法叫做submit，通过ExecutorService.submit()方法提交的任务，可以获取任务执行完的返回值。

在实际业务场景中，Future和Callable基本是成对出现的，Callable负责产生结果，Future负责获取结果。
1、Callable接口类似于Runnable，只是Runnable没有返回值。
2、Callable任务除了返回正常结果之外，如果发生异常，该异常也会被返回，即Future可以拿到异步执行任务各种结果；
3、Future.get方法会导致主线程阻塞，直到Callable任务执行完成；