JDK1.8源码分析：Executor和ThreadPoolExecutor线程池的设计和源码实现

小鱼儿 2022-03-19 08:29 123阅读 0赞

### 概述 ###

*  Executor线程执行器框架是在jDK1.5提供的，设计的主要目的是实现Runnable任务的提交和执行分离：任务提交主要是指在应用代码中实现Runnable接口定义业务相关的一个任务，然后交给一个Thread线程对象来执行；任务的执行是指Thread线程执行该任务。有了Executor线程执行器之后，我们只需要实现Runnalbe接口定义自己的任务即可，不需要显示创建Thread对象来执行，而是交给Executor，Executor会负责调度线程来执行该任务，具体为：
 *  在Executor内部的线程池调度一个空闲的线程来执行该任务，或者如果没有空闲线程，则将该任务放到内部的一个队列排队等待，或者如果没有空闲线程，队列也没有空闲空间存放了，则使用相应的拒绝策略来拒绝执行这个任务。
 *  由于Executor是异步执行任务的，对每次的任务提交可以返回一个Future对象，可以通过该Future对象来阻塞等待该任务的执行结果，或者通过该Future对象来取消该任务的执行。
 *  除了实现任务提交和任务执行分离来实现自动化线程调度执行，简化编程复杂度，Executor线程执行器还包括以下优点：
    
    1.  通过线程池机制，实现线程复用，避免频繁的线程创建和销毁，提高了在执行大量任务时的性能，避免之前那种每个任务显式创建一个Thread对象来执行；
    2.  通过设置线程池和内部队列的大小，可以根据当前系统资源情况，灵活控制系统的资源消耗，避免创建过多线程或者队列存放过多任务，造成系统资源耗尽，如内存，CPU资源，导致系统宕机；
    3.  提供了任务执行情况的统计功能，如该Executor实例当前一共执行了多少任务。

### ThreadPoolExecutor ###

*  Executor接口的核心实现类为ThreadPoolExecutor，在ThredPoolExector中主要维护了一个线程池，一个线程安全的阻塞队列来进行任务的管理和执行，同时在类设计层面，ThreadPoolExecutor提供以下几个参数来对内部的线程池，队列等进行控制：
    
    1.  corePoolSize：线程池核心大小，即任务需要排队之前的线程池的最大大小，如果corePoolSize个线程全部繁忙，则新来的任务需要排队了；
    2.  maximumPoolSize：线程池最大大小，即当corePoolSize个线程全部繁忙，等待队列workQueue也满了，则会继续新建线程来执行任务，直到总的线程数量达到maximumPoolSize。所以如果workQueue使用的是无界队列，则该参数无效，因为队列永远不会满；
    3.  keepAliveTime：超过corePoolSize的线程，如果空闲超过keepAliveTime时间，则需要被销毁回收，默认corePoolSize范围内的线程一旦创建，即使空闲也不会回收的。在内部实现中，由于每个线程在空闲时是在工作队列workQueue中阻塞等待任务到来的，故该keepAliveTime是设置在 workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS)，即阻塞keepAliveTime时间之后，如果超时，则需要销毁该线程了，具体后面分析；
    4.  workQueue：当前没有空闲线程时，任务的等待队列；
    5.  threadFactory：线程池的线程创建工厂类；
    6.  RejectedExecutionHandler：任务拒绝策略，即线程池没有空闲线程且无法继续创建线程，队列也排满等待执行的任务时，新来的任务的拒绝策略。
 *  以下主要基于ThreadPoolExecutor来对Executor线程执行器进行分析。

#### 构造函数 ####

*  ThreadPoolExecutor的构造函数主要是对以上参数进行赋值，具体的线程池中的线程是延迟初始化的，即在有任务提交的时候才开始创建线程：
    
        // 任务等待队列
        private final BlockingQueue<Runnable> workQueue;
        
        private int largestPoolSize;
        
        private volatile ThreadFactory threadFactory;
        
        /** * Handler called when saturated or shutdown in execute. */
        private volatile RejectedExecutionHandler handler;
        
        
        // 超过corePoolSize的线程，如果没有处理任务，则空闲多久被remove
        private volatile long keepAliveTime;
        
        // corePoolSize范围的线程在空闲时，是否运行remove，默认为false，即不能
        // 当设置为true时，注意keepAliveTime需要大于0，否则会抛异常
        private volatile boolean allowCoreThreadTimeOut;
        
        private volatile int corePoolSize;
        
        // 不过用户配置线程池最大多少，都不能超过CAPACITY，即(1 << 29) - 1
        private volatile int maximumPoolSize;
        
        /** * The default rejected execution handler */
        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;
        }

#### 工作线程和线程池 ####

*  工作线程池主要通过一个HashSet来实现，使用HashSet而不会存在线程安全问题是因为在ThreadPoolExecutor内部对HashSet中节点的增删，即线程节点的增删，都是需要通过一个ReentrantLock来加锁，定义如下：
    
        /** * Set containing all worker threads in pool. Accessed only when * holding mainLock. */
        // workers线程的增删是在mainLock的加锁保护下进行的，这样保证了线程安全，故可以使用HashSet
        private final HashSet<Worker> workers = new HashSet<Worker>();
        
        /** * Wait condition to support awaitTermination */
        private final Condition termination = mainLock.newCondition();
 *  工作线程主要是通过定义一个Worker类来实现：继承于AQS，实现了Runnable接口，继承AQS的主要目的是使用AQS提供的互斥锁，即state=1，来表明当前worker是否在处理任务还是空闲，实现Runnable接口的主要目的是：将该worker对象自身作为一个task放到Worker内部的线程thread执行，在run方法中定义该工作线程的工作逻辑，主要是调用runWorker(this)方法，具体在任务执行部分分析runWorker。
 *  内部包含了一个Thread对象，这个是实际执行任务的线程，同时包含一个Runable对象firstTask，由于是延迟创建工作线程的，所以这个是第一个触发创建这个工作线程的任务。
    
        // 继承AQS主要是使用AQS提供的互斥锁，即state=1，来表明当前worker是否在处理任务还是空闲，
        // 即加锁表示正在处理任务，不加锁表示空闲
        private final class Worker
            extends AbstractQueuedSynchronizer
            implements Runnable
        { 
            /** * This class will never be serialized, but we provide a * serialVersionUID to suppress a javac warning. */
            private static final long serialVersionUID = 6138294804551838833L;
        
            /** Thread this worker is running in. Null if factory fails. */
            final Thread thread;
            /** Initial task to run. Possibly null. */
            Runnable firstTask;
            /** Per-thread task counter */
            volatile long completedTasks;
            
            /** * Creates with given first task and thread from ThreadFactory. * @param firstTask the first task (null if none) */
            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);
            }
            
            ...
            
        }

#### 任务的提交 ####

*  任务的提交主要在ThreadPoolExecutor的execute方法定义，其中execute为void方法，没有返回值；如果提交任务，需要获取一个Future对象来跟踪这个任务的执行，获取执行结果，则需要使用submit方法。在内部实现中submit也是调用了execute来提交任务到线程池。
 *  execute定义：具体看代码注释，其中完成了对Executor框架设计中对corePoolSize，等待队列，maximumPoolSize，拒绝策略相关语义的实现，即对一个新任务的提交：
    
    1.  当线程池线程数量少于corePoolSize时，则创建一个新的线程来执行这个任务，不管当前线程池是否存在空闲线程；
    2.  当线程池线程数量达到corePoolSize时，则将该任务放到等待队列中，由于线程池中的空闲线程从这个队列取出然后执行；其中workQueue的offer方法为非阻塞，成功则返回true，否则返回false；
    3.  当等待队列workQueue也满了，当线程池线程数量少于maximumPoolSize时，则继续创建新线程执行这个任务，如果线程数量超过了maximumPoolSize，则此时说明存在太大任务等待执行了，则调用reject方法，根据具体的拒绝策略处理这个任务，默认为AbortPolicy，即在execute方法抛异常，所以如果可能存在这种情况，则一般需要在应用代码中捕获该异常。
    
        public void execute(Runnable command) { 
            if (command == null)
                throw new NullPointerException();
            
            int c = ctl.get();
            // corePoolSize不满，则新建线程来处理task
            if (workerCountOf(c) < corePoolSize) { 
                if (addWorker(command, true))
                    return;
                c = ctl.get();
            }
            // corePoolSize满了，放入队列
            // isRunning通过判断状态是否是SHUTDOWN之前的，是则说明还在运行，
            // 否则说明调用了shutdown或者shutdownNow
            if (isRunning(c) && workQueue.offer(command)) { 
                int recheck = ctl.get();
                if (! isRunning(recheck) && remove(command))
                    // 根据reject策略，拒绝该任务的执行，默认为AbortPolicy
                    reject(command);
                else if (workerCountOf(recheck) == 0)
                    addWorker(null, false);
            }
            // 队列满了，core为false，表示使用maximumPoolSize作为边界，
            // 小于maximumPoolSize则，进行创建线程
            else if (!addWorker(command, false))
                reject(command);
        }
 *  addWorker延迟创建工作线程实现如下：首先在第一个自旋中通过比较当前线程池线程数量，corePoolSize，maximumPoolSize来确定是否可以继续创建线程；如果可以则创建Worker工作线程，并在加锁mainLock的包含下，将该工作线程添加到线程池workers中。最后调用该工作线程内部的线程thread的start的方法，开始这个工作线程的执行。
    
        // core为true表示将当前worker的数量与corePoolSize对比，大于则不再新建worker线程，否则与maximumPoolSize对比
        private boolean addWorker(Runnable firstTask, boolean core) { 
            retry:
            // 自旋检查当前的worker线程数量
            for (;;) { 
                int c = ctl.get();
                int rs = runStateOf(c);
        
                // 被shutdown或者shutdownNow了则不能进行添加任务
                // Check if queue empty only if necessary.
                if (rs >= SHUTDOWN &&
                    ! (rs == SHUTDOWN &&
                       firstTask == null &&
                       ! workQueue.isEmpty()))
                    return false;
        
                for (;;) { 
                    int wc = workerCountOf(c);
                    // 判断是否可以新建worker线程了，最大不能超过CAPACITY
                    if (wc >= CAPACITY ||
                        wc >= (core ? corePoolSize : maximumPoolSize))
                        return false;
                    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
                }
            }
        
            boolean workerStarted = false;
            boolean workerAdded = false;
            Worker w = null;
            // 创建worker线程
            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) { 
                        // 调用thread的start方法，开始这个工作线程的执行
                        t.start();
                        workerStarted = true;
                    }
                }
            } finally { 
                if (! workerStarted)
                    addWorkerFailed(w);
            }
            return workerStarted;
        }
        
        
        /** * Creates with given first task and thread from ThreadFactory. * @param firstTask the first task (null if none) */
        Worker(Runnable firstTask) { 
            setState(-1); // inhibit interrupts until runWorker
            this.firstTask = firstTask;
            this.thread = getThreadFactory().newThread(this);
        }

#### 任务的执行 ####

*  每个工作线程Worker主要是在run方法中调用runWorker来定义工作线程的具体工作逻辑，即执行创建该worker时提交的任务和从任务等待队列workQueue获取提交的任务并执行：主要是在while循环中调用getTask方法从任务等待队列获取任务，其中getTask为阻塞执行的，即如果等待队列没有任务，则阻塞等待：
    
        // worker线程的run方法调用该方法，定义worker线程的任务处理逻辑
        final void runWorker(Worker w) { 
            Thread wt = Thread.currentThread();
            // 初始化任务，即worker是lazy创建的，所以这个是触发创建worker的task
            Runnable task = w.firstTask;
            w.firstTask = null;
            w.unlock(); // allow interrupts
            boolean completedAbruptly = true;
            try { 
                // 从队列获取任务
                // getTask在worker线程数量少于等于corePoolSize为阻塞获取，故无限阻塞在while这里；
                // 在大于corePoolSize，为阻塞指定时间keepAliveTime获取，如果没有任务，则退出while，执行到finally了
                while (task != null || (task = getTask()) != null) { 
                    // 运行任务时，需要先加锁，加锁状态表示正在运行，否则是idle状态
                    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
        
                    // 状态为STOP，表示调用过了shutdownNow，
                    // 中断worker线程，尝试停止当前正在执行的任务
                    if ((runStateAtLeast(ctl.get(), STOP) ||
                         (Thread.interrupted() &&
                          runStateAtLeast(ctl.get(), STOP))) &&
                        !wt.isInterrupted())
                        wt.interrupt();
                    try { 
                        beforeExecute(wt, task);
                        Throwable thrown = null;
                        try { 
                            // 执行Runnable task的run
                            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++;
                        // 是否锁，回归空闲idle状态
                        w.unlock();
                    }
                }
                completedAbruptly = false;
            } finally { 
                // 大于corePoolSize的worker线程，则空闲的时候需要关闭
                processWorkerExit(w, completedAbruptly);
            }
        }
 *  getTask：从任务等待队列workQueue阻塞等待获取任务。同时也在这里实现了超过corePoolSize的线程的超时清理控制，即如果超过了corePoolSize的线程，则调用了workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS)法，即阻塞keepAliveTime时间，如果还是没有任务，则返回null，则在runWorker中退出while循环，然后关闭这个工作线程。
    
        private Runnable getTask() { 
            boolean timedOut = false; // Did the last poll() time out?
            // 自旋
            for (;;) { 
                int c = ctl.get();
                int rs = runStateOf(c);
        
                // 状态为STOP了，说明调用过了shutdownNow，则不再从队列取任务来执行了
                // 或者状态为SHUTDOWN，说明调用过了shutdown，如果任务队列为空，也不再从队列取任务来执行了
                // 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?
                // 当前workers线程数量是否大于corePoolSize
                boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
        
                if ((wc > maximumPoolSize || (timed && timedOut))
                    && (wc > 1 || workQueue.isEmpty())) { 
                    if (compareAndDecrementWorkerCount(c))
                        return null;
                    continue;
                }
        
                try { 
                    // 如果当前worker线程数量大于corePoolSize了，则使用poll阻塞指定时间keepAliveTime，从任务队列取任务；
                    // 否则使用take无限阻塞从任务队列取任务，因为这个时候新来的任务是新建worker线程来处理的，
                    // 已经存在的worker线程只需阻塞在这里等待corePoolSize满了，开始往队列填充任务时才执行
                    Runnable r = timed ?
                        workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                        workQueue.take();
                    if (r != null)
                        return r;
                    timedOut = true;
                } catch (InterruptedException retry) { 
                    timedOut = false;
                }
            }
        }

#### 任务的执行结果 ####

*  如果需要获取任务的执行结果，则需要通过ThreadPoolExecutor的submit方法来提交任务，这个方法会返回一个Future接口的实现类对象，具体为FutureTask。submit方法在ThreadPoolExecutor的基类，即抽象类AbstractExecutorService中定义：内部是将task包装成了FutureTask，然后通过execute方法来提交到线程池。
    
        public Future<?> submit(Runnable task) { 
            if (task == null) throw new NullPointerException();
            // 将应用代码中的task包装成一个FutureTask，然后交给线程池执行
            // 线程池调度一个线程来执行FutureTask的run，在FutureTask的run中调用task的run
            RunnableFuture<Void> ftask = newTaskFor(task, null);
            execute(ftask);
            // 返回这个包装了task的FutureTask，在应用代码中可以通过get来获取执行结果
            return ftask;
        }
        
        protected <T> RunnableFuture<T>     newTaskFor(Runnable runnable, T value) { 
            return new FutureTask<T>(runnable, value);
        }
 *  关于Future和FutureTask的设计，如何通过get获取执行结果，通过cannel取消任务执行的实现原理，详见：[JDK1.8源码分析：Future和FutureTask-任务异步执行结果  
    、  
    ][JDK1.8_Future_FutureTask-_ _]

#### 任务的拒绝 ####

*  当线程池无法继续创建线程，任务等待队列都满了时，则对于新提交的任务，则通过reject方法来调用对应的拒绝策略来处理。
 *  ThreadPoolExecutor提供了四种拒绝策略，分别为抛Abort异常，在线程池主线程中直接执行该任务，默默丢弃不做任务操作，从任务等待队列移除等待最久的任务，即队列头对应的任务，默认为抛Abort异常：
    
        public static class AbortPolicy implements RejectedExecutionHandler { 
            
            ...
            
            public void rejectedExecution(Runnable r, ThreadPoolExecutor e) { 
                throw new RejectedExecutionException("Task " + r.toString() +
                                                     " rejected from " +
                                                     e.toString());
            }
        }
        
        public static class CallerRunsPolicy implements RejectedExecutionHandler { 
            
            ...
            
            public void rejectedExecution(Runnable r, ThreadPoolExecutor e) { 
                if (!e.isShutdown()) { 
                    r.run();
                }
            }
        }
        
        public static class DiscardPolicy implements RejectedExecutionHandler { 
        
            ...
            
            public void rejectedExecution(Runnable r, ThreadPoolExecutor e) { 
            }
        }
        
        public static class DiscardOldestPolicy implements RejectedExecutionHandler { 
            ...
            
            public void rejectedExecution(Runnable r, ThreadPoolExecutor e) { 
                if (!e.isShutdown()) { 
                    e.getQueue().poll();
                    e.execute(r);
                }
            }
        }

#### 线程池的关闭 ####

*  线程池的关闭主要包括平滑关闭和暴力关闭两种，内部主要是通过runState状态变量来做控制。
 *  shutdown：平滑关闭，停止新任务的提交，等待正在排队的任务和正在执行的任务执行完成：
    
        public void shutdown() { 
            final ReentrantLock mainLock = this.mainLock;
            mainLock.lock();
            try { 
                checkShutdownAccess();
                // SHUTDOWN表示不能新添加任务了，之前添加和正在运行的任务可以运行完
                advanceRunState(SHUTDOWN);
                // 中断空闲的worker线程
                interruptIdleWorkers();
                onShutdown(); // hook for ScheduledThreadPoolExecutor
            } finally { 
                mainLock.unlock();
            }
            tryTerminate();
        }
 *  shutdownNow：暴力关闭，停止新任务的提交，中断正在执行任务的线程，停止排队任务的执行和尝试停止正在执行任务的执行。
    
        public List<Runnable> shutdownNow() { 
            List<Runnable> tasks;
            final ReentrantLock mainLock = this.mainLock;
            mainLock.lock();
            try { 
                checkShutdownAccess();
                // STOP表示不能添加新任务，之前添加的任务不再安排执行，中断正在执行的任务
                advanceRunState(STOP);
                // 中断所有worker线程，不管是空闲还是正在执行任务的worker线程
                interruptWorkers();
                tasks = drainQueue();
            } finally { 
                mainLock.unlock();
            }
            tryTerminate();
            return tasks;
        }

[JDK1.8_Future_FutureTask-_ _]: https://blog.csdn.net/u010013573/article/details/87435850