jdk源码解析五之Executor框架(ThreadPoolExecutor,Thread,ScheduledExecutorService)

一时失言乱红尘 2023-02-13 05:52 10阅读 0赞

### 文章目录 ###

*  Executor
 *   *  ExecutorService
     *   *  ScheduledExecutorService
 *  AbstractExecutorService
 *   *  总结
 *  ThreadPoolExecutor
 *   *  成员变量以及方法
     *  构造
     *  execute
     *   *  tryTerminate
         *  interruptIdleWorkers
         *  run
     *  shutdown
     *  shutdownNow
     *  总结
 *  ScheduledThreadPoolExecutor
 *   *  //todo
 *  Thread
 *   *  构造
     *  start
     *  interrupt
     *  join
     *  State
     *  run
     *  总结
 *  ThreadFactory
 *   *  DefaultThreadFactory
 *  Executors
 *  ExecutorCompletionService
 *   *  构造
     *  submit
     *  take
     *  poll
     *  总结

Executor基于生产者消费者模式,提交任务操作相当于生产者,执行任务的线程相当于消费者

线程不能不停创建,所以复用线程衍生出了线程池的概念,而不停的复用线程,假如有10个线程去跑100请求,还有90请求怎么办？引入了阻塞队列.而如果10个线程最多只能处理100线程,那101线程怎么办?这里就引入new ThreadPoolExecutor.CallerRunsPolicy()4种处理的概念

开启多少线程,最大多少线程,?  
线程的任务使FIFO,LIFO还是按照优先级执行?  
多余的任务,是否阻塞队列缓存?  
队列撑爆了,使用何种策略处理?  
取消的时候,平滑过渡,还是强制终止所有任务?  
出现异常,可定义异常处理器setDefaultUncaughtExceptionHandler/setUncaughtExceptionHandler

设置线程池大小  
线程池过大,造成大量线程在相对很小的cpu和内存资源上发生竞争，会导致内存过高  
线程池过小,许多空闲处理器无法工作,降低吞吐  
需要考虑,cpu，内存,任务是计算密集类型还是io密集还是2者,  
如果需要执行不同类别任务,并且之间行为相差很大,考虑使用多个线程池，从而使每个线程池可以根据各自的工作负载来调整  
计算密集型,线程大小为N+1(多一个备份)  
包含I/O或者其他阻塞任务  
N=cpu个数 Runtime.availableProcessors  
U=CPU利用率 0<=U<=1  
W/C=任务等待时间/计算时间  
线程池最优大小=N*U*(1+W/C)  
资源:该资源的可用总量/每个任务对该资源的需求量

N *(W+C)/C

[美团给出的解决方案][Link 1]

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L051YW5fRmVuZw_size_16_color_FFFFFF_t_70]

# Executor #

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L051YW5fRmVuZw_size_16_color_FFFFFF_t_70 1]

## ExecutorService ##

提供了管理终止的方法，以及可为跟踪一个或多个异步任务执行状况而生成 Future 的方法。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L051YW5fRmVuZw_size_16_color_FFFFFF_t_70 2]

### ScheduledExecutorService ###

对可调度的支持

![在这里插入图片描述][2020060210092042.png]

# AbstractExecutorService #

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L051YW5fRmVuZw_size_16_color_FFFFFF_t_70 3]

protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) {
        
            //底层封装FutureTask
            return new FutureTask<T>(runnable, value);
        }
    
      public Future<?> submit(Runnable task) {
        
            if (task == null) throw new NullPointerException();
            RunnableFuture<Void> ftask = newTaskFor(task, null);
            //子类实现
            execute(ftask);
            return ftask;
        }
    
      //返回最快的结果
        public <T> T invokeAny(Collection<? extends Callable<T>> tasks)
            throws InterruptedException, ExecutionException {
        
            try {
        
                return doInvokeAny(tasks, false, 0);
            } catch (TimeoutException cannotHappen) {
        
                assert false;
                return null;
            }
        }
    
     private <T> T doInvokeAny(Collection<? extends Callable<T>> tasks,
                                  boolean timed, long nanos)
            throws InterruptedException, ExecutionException, TimeoutException {
        
            //任务为空校验
            if (tasks == null)
                throw new NullPointerException();
            int ntasks = tasks.size();
            if (ntasks == 0)
                throw new IllegalArgumentException();
    
            ArrayList<Future<T>> futures = new ArrayList<Future<T>>(ntasks);
            //使用ExecutorCompletionService,获取结果前阻塞.内部维护一个阻塞队列,继承Future保存结果
            ExecutorCompletionService<T> ecs =
                new ExecutorCompletionService<T>(this);
    
    
            try {
        
    
                ExecutionException ee = null;
                final long deadline = timed ? System.nanoTime() + nanos : 0L;
                //获取回调方法的迭代器
                Iterator<? extends Callable<T>> it = tasks.iterator();
                //结果封装到集合
                futures.add(
                        //执行回调函数
                        ecs.submit(it.next())
                );
                --ntasks;
                //记录执行的回调函数个数
                int active = 1;
    
                for (;;) {
        
                    //获取结果,非阻塞
                    Future<T> f = ecs.poll();
                    //尚未获取结果
                    if (f == null) {
        
                        //回调任务>1,则继续执行下一个回调任务
                        if (ntasks > 0) {
        
                            --ntasks;
                            futures.add(ecs.submit(it.next()));
                            //记录正在执行个数
                            ++active;
                        }
                        else if (active == 0)
                            //正在执行个数为0，则说明获取结果阶段抛出异常,直接结束
                            break;
                        else if (timed) {
        
                            //如果阻塞,则设置超时获取结果
                            f = ecs.poll(nanos, TimeUnit.NANOSECONDS);
                            if (f == null)
                                throw new TimeoutException();
                            //超时时间减少
                            nanos = deadline - System.nanoTime();
                        }
                        else
                            //阻塞,获取并移除表示下一个已完成任务的 Future
                            f = ecs.take();
                    }
                    //阻塞队列如果有结果
                    if (f != null) {
        
                        --active;
                        try {
        
                            //返回结果
                            return f.get();
                        } catch (ExecutionException eex) {
        
                            ee = eex;
                        } catch (RuntimeException rex) {
        
                            ee = new ExecutionException(rex);
                        }
                    }
                }
    
                if (ee == null)
                    ee = new ExecutionException();
                throw ee;
    
            } finally {
        
                for (int i = 0, size = futures.size(); i < size; i++)
                    //取消所有结果
                    futures.get(i).cancel(true);
            }
        }
    
     public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
            throws InterruptedException {
        
            if (tasks == null)
                throw new NullPointerException();
            ArrayList<Future<T>> futures = new ArrayList<Future<T>>(tasks.size());
            boolean done = false;
            try {
        
                for (Callable<T> t : tasks) {
        
                    //FutureTask底层封装
                    RunnableFuture<T> f = newTaskFor(t);
                    //结果存储集合
                    futures.add(f);
                    //子类实现
                    execute(f);
                }
                for (int i = 0, size = futures.size(); i < size; i++) {
        
                    //获取结果
                    Future<T> f = futures.get(i);
                    //如果还在执行
                    if (!f.isDone()) {
        
                        try {
        
                            //阻塞获取结果
                            f.get();
                        } catch (CancellationException ignore) {
        
                        } catch (ExecutionException ignore) {
        
                        }
                    }
                }
                done = true;
                //返回结果集
                return futures;
            } finally {
        
                //如果中途抛出异常
                if (!done)
                    for (int i = 0, size = futures.size(); i < size; i++)
                        //取消所有任务
                        futures.get(i).cancel(true);
            }
        }
    
     public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,
                                             long timeout, TimeUnit unit)
            throws InterruptedException {
        
            if (tasks == null)
                throw new NullPointerException();
            long nanos = unit.toNanos(timeout);
            ArrayList<Future<T>> futures = new ArrayList<Future<T>>(tasks.size());
            boolean done = false;
            try {
        
                //所有回调任务封装成Task,且保存到集合
                for (Callable<T> t : tasks)
                    futures.add(newTaskFor(t));
    
                //记录超时时间
                final long deadline = System.nanoTime() + nanos;
                final int size = futures.size();
    
                // Interleave time checks and calls to execute in case
                // executor doesn't have any/much parallelism.
                for (int i = 0; i < size; i++) {
        
                    //执行任务
                    execute((Runnable)futures.get(i));
                    nanos = deadline - System.nanoTime();
                    //如果超时,则直接返回结果
                    if (nanos <= 0L)
                        return futures;
                }
    
                for (int i = 0; i < size; i++) {
        
                    Future<T> f = futures.get(i);
                    //如果任务还在执行
                    if (!f.isDone()) {
        
                        if (nanos <= 0L)
                            return futures;
                        try {
        
                            //超时获取结果
                            f.get(nanos, TimeUnit.NANOSECONDS);
                        } catch (CancellationException ignore) {
        
                        } catch (ExecutionException ignore) {
        
                        } catch (TimeoutException toe) {
        
                            return futures;
                        }
                        nanos = deadline - System.nanoTime();
                    }
                }
                done = true;
                return futures;
            } finally {
        
                if (!done)
                    for (int i = 0, size = futures.size(); i < size; i++)
                        futures.get(i).cancel(true);
            }
        }

## 总结 ##

底层使用FutureTask  
invokeAny使用ExecutorCompletionService阻塞队列获取结果

# ThreadPoolExecutor #

![在这里插入图片描述][20200602111052547.png]

## 成员变量以及方法 ##

//保存线程数量和线程池的状态,初始CTL=111 0(32-3个0)
        private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
        private static final int COUNT_BITS = Integer.SIZE - 3;
        //线程池最大1^29-1
        private static final int CAPACITY   = (1 << COUNT_BITS) - 1;
    
        // runState is stored in the high-order bits
        //5种运行状态,分别用高3位保存运行状态
        // 接收新任务,并执行队列中的任务
        private static final int RUNNING    = -1 << COUNT_BITS;
        // 不接收新任务,但是执行队列中的任务
        private static final int SHUTDOWN   =  0 << COUNT_BITS;
        // 不接收新任务,不执行队列中的任务,中断正在执行中的任务
        private static final int STOP       =  1 << COUNT_BITS;
        //所有的任务都已结束,线程数量为 0,处于该状态的线程池即将调用 terminated()方法
        private static final int TIDYING    =  2 << COUNT_BITS;
        // terminated()方法执行完成
        private static final int TERMINATED =  3 << COUNT_BITS;
    
        // Packing and unpacking ctl
        //获取高3位
        private static int runStateOf(int c)     {
         return c & ~CAPACITY; }
        //获取低29位
        private static int workerCountOf(int c)  {
         return c & CAPACITY; }
        private static int ctlOf(int rs, int wc) {
         return rs | wc; }

## 构造 ##

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.acc = System.getSecurityManager() == null ?
                    null :
                    AccessController.getContext();
            //设置核心大小
            this.corePoolSize = corePoolSize;
            //设置最大线程大小
            //当添加新的执行任务,如果运行的线程少于 corePoolSize，则创建新线程来处理请求
            this.maximumPoolSize = maximumPoolSize;
            //设置工作队列
            this.workQueue = workQueue;
            //空闲时间
            //如果池中当前有多于 corePoolSize 的线程，则这些多出的线程在空闲时间超过 keepAliveTime 时将会终止
            this.keepAliveTime = unit.toNanos(keepAliveTime);
            //创建线程的工厂,默认Executors.defaultThreadFactory()
            this.threadFactory = threadFactory;
            //饱和策略,默认AbortPolicy()
            /*
        AbortPolicy(终止策略):默认策略,抛出RejectExecutionException
        DisCardPolicy:新提交的任务无法保存到队列中等待执行时,抛弃该任务
        DisCardOldestPolicy:抛弃下一个将被执行的任务,尝试重新提交新的任务(如果工作队列是一个优先队列,此策略会导致抛弃优先级最高的任务)
        CallerRunsPolicy:将任务回退到调用者执行,主线程在一段时间内不能提交任务,使工作者线程有时间处理完正在执行的任务,在此期间主线程不会accept,到达的请求会保存在TCP层队列，持续过载,TCP会抛弃请求
                    路径为：线程池-工作队列-应用程序-TCP-客户端
             */
            this.handler = handler;
        }

## execute ##

public void execute(Runnable command) {
        
            //空校验
            if (command == null)
                throw new NullPointerException();
    
            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();
                //如果任务不处于Run状态,则删除任务,并拒绝任务.
                if (! isRunning(recheck) && remove(command))
                    //拒绝任务
                    reject(command);
                else if (workerCountOf(recheck) == 0)
                    //处于运行状态,但可使用线程为0,则试着创建线程最大线程数新开一个新线程
                    addWorker(null, false);
            }
            else if
                //核心池没处于运行状态或者队列已满，试着创建线程最大线程数新开一个新线程处理
            (!addWorker(command, false))
                //如果创建新线程失败了，说明线程池被关闭或者线程池完全满了，拒绝任务
                reject(command);
        }

private boolean addWorker(Runnable firstTask, boolean core) {
        
            retry://goto 语句,避免死循环
            for (;;) {
        
                //获取线程数量和线程池的状态
                int c = ctl.get();
                //获取运行状态
                int rs = runStateOf(c);
    
                // Check if queue empty only if necessary.
                /*
                1:线程池已经最少关闭
                2:状态没关闭,任务不为空,且工作线程为空
                同时满足以上2条件,则直接返回false
    
                2种情况下允许添加新线程
                线程池处于运行状态
                线程池处于关闭状态,且任务为null,且工作线程不为空
                 */
                //状态至少SHUTDOWN,
                if (rs >= SHUTDOWN &&
                        //状态不为shutDown
                    ! (rs == SHUTDOWN &&
                            //任务不为空
                       firstTask == null &&
                            //工作线程为空
                       ! workQueue.isEmpty()))
                    return false;
    
                //自旋
                for (;;) {
        
                    //获取当前工作线程个数
                    int wc = workerCountOf(c);
                    //当前工作线程>最大上限 或根据是否开启最大线程判断,则返回
                    if (wc >= CAPACITY ||
                        wc >= (core ? corePoolSize : maximumPoolSize))
                        return false;
                    //当前工作线程+1,成功则结束
                    if (compareAndIncrementWorkerCount(c))
                        break retry;
                    //cas操作失败
                    //获取最新ctl值
                    c = ctl.get();  // Re-read ctl
                    //说明更改了线程池的状态,继续重试
                    if (runStateOf(c) != rs)
                        continue retry;
                    //CAS由于更改workerCount而失败，继续内层循环
                    // else CAS failed due to workerCount change; retry inner loop
                }
            }
    
            boolean workerStarted = false;
            boolean workerAdded = false;
            Worker w = null;
            try {
        
                //创建工作线程
                w = new Worker(firstTask);
                //获取线程
                final Thread t = w.thread;
                //线程不为空
                if (t != null) {
        
                    //获取main锁
                    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());
    
                        //如果处于运行状态,或者处于关闭状态且任务为null
                        if (rs < SHUTDOWN ||
                            (rs == SHUTDOWN && firstTask == null)) {
        
                            //线程存活
                            if (t.isAlive()) // precheck that t is startable
                                //线程尚未启动,就处于存活状态,则异常
                                throw new IllegalThreadStateException();
                            //添加到workers集合
                            workers.add(w);
                            //获取当前工作线程大小
                            int s = workers.size();
                            //超过largestPoolSize
                            if (s > largestPoolSize)
                                //重新设置largestPoolSize
                                largestPoolSize = s;
                            //标记正常处理
                            workerAdded = true;
                        }
                    } finally {
        
                        //释放锁
                        mainLock.unlock();
                    }
                    //正常添加
                    if (workerAdded) {
        
                        //执行
                        t.start();
                        //标记正常启动
                        workerStarted = true;
                    }
                }
            } finally {
        
                //如果没正常启动
                if (! workerStarted)
                    //异常处理,包括删除添加的工作线程,工作线程个数-1等
                    addWorkerFailed(w);
            }
            return workerStarted;
        }

private void addWorkerFailed(Worker w) {
        
            //获取重入锁
            final ReentrantLock mainLock = this.mainLock;
            mainLock.lock();
            try {
        
                if (w != null)
                    //移除当前工作线程
                    workers.remove(w);
                //CAS当前工作线程数量-1
                decrementWorkerCount();
                //终止线程池
                tryTerminate();
            } finally {
        
                mainLock.unlock();
            }
        }
    
        private void decrementWorkerCount() {
        
            do {
        } while (! compareAndDecrementWorkerCount(ctl.get()));
        }

### tryTerminate ###

final void tryTerminate() {
        
            for (;;) {
        
                int c = ctl.get();
                /*处于运行状态,或者
                 *线程池状态>=TIDYING
                 *线程池=SHUTDOWN并且workQueue不为空
                 *直接return，不能终止
                 */
                if (isRunning(c) ||
                    runStateAtLeast(c, TIDYING) ||
                    (runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty()))
                    return;
                //工作线程个数>0
                if (workerCountOf(c) != 0) {
         // Eligible to terminate
                    //中断工作线程
                    interruptIdleWorkers(ONLY_ONE);
                    return;
                }
    
                final ReentrantLock mainLock = this.mainLock;
                mainLock.lock();
                try {
        
                    //CAS设置线程池状态为TIDYING
                    if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {
        
                        try {
        
                            terminated();
                        } finally {
        
                            //设置线程池的状态为TERMINATED
                            ctl.set(ctlOf(TERMINATED, 0));
                            //发送释放信号给在termination条件上等待的线程
                            termination.signalAll();
                        }
                        return;
                    }
                } finally {
        
                    //释放锁
                    mainLock.unlock();
                }
                // else retry on failed CAS
            }
        }

### interruptIdleWorkers ###

private void interruptIdleWorkers(boolean onlyOne) {
        
            final ReentrantLock mainLock = this.mainLock;
            mainLock.lock();
            try {
        
                //遍历工作线程
                for (Worker w : workers) {
        
                    Thread t = w.thread;
                    //当前线程没有中断,且尝试获取锁成功
                    if (!t.isInterrupted() && w.tryLock()) {
        
                        try {
        
                            //中断
                            t.interrupt();
                        } catch (SecurityException ignore) {
        
                        } finally {
        
                            w.unlock();
                        }
                    }
                    //最多中断一个线程
                    if (onlyOne)
                        break;
                }
            } finally {
        
                mainLock.unlock();
            }
        }

### run ###

private boolean addWorker(Runnable firstTask, boolean core) {
        
    //  ...
                w = new Worker(firstTask);
    //...
    //这里调用的Worker.run
                        t.start();
    //...
    }

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

final void runWorker(Worker w) {
        
            //获取当前线程
            Thread wt = Thread.currentThread();
            //获取任务
            Runnable task = w.firstTask;
            //提前释放
            w.firstTask = null;
            /*
             *unlock方法会调用AQS的release方法
             *release方法会调用具体实现类也就是Worker的tryRelease方法
             *也就是将AQS状态置为0，允许中断
             * interruptIfStarted()中只有 state>=0 才允许调用中断A
             */
            w.unlock(); // allow interrupts
            //标记是否正常执行完
            boolean completedAbruptly = true;
            try {
        
                //如果task不为null,或者获取任务不为null,则进入循环
                //线程复用
                while (task != null || (task = getTask()) != null) {
        
                    //不是为了防止并发执行任务，为了在 shutdown()时不终止正在运行的 worker
                    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或者
                    if ((runStateAtLeast(ctl.get(), STOP) ||
                    // 当前线程是否被中断（检查中断标志），返回一个boolean并清除中断状态，第二次再调用时中断状态已经被清除，将返回一个false。
                         (Thread.interrupted() &&
                                 //且当前线程池>=stop状态
                          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 {
        
                        //清除当前任务,下次则执行getTask获取最新任务
                        task = null;
                        //记录当前线程完成任务个数
                        w.completedTasks++;
                        w.unlock();
                    }
                }
                //标记正常完成
                completedAbruptly = false;
            } finally {
        
                //后置处理
                processWorkerExit(w, completedAbruptly);
            }
        }

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.
                //如果线程池状态>=SHUTDOWN,且workQueue为空
                // >=STOP,shutdownNow()会导致变成 STOP（此时不用考虑 workQueue的情况）
                if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
        
                    //CAS 当前线程数-1
                    decrementWorkerCount();
                    //当前线程会退出执行
                    return null;
                }
    
                int wc = workerCountOf(c);
    
                // Are workers subject to culling?
                //判断核心线程是否允许进行超时,默认不允许
                //当前线程数是否>核心线程数,对于超过核心线程的线程进行超时控制
                boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
    
                //超过最大线程数,只有2种设置最大线程数,构造+setMaximumPoolSize.
                // 当构造设置大小后,按照正常的顺序,不会超过线程数,也就是说,这里可能执行了setMaximumPoolSize,在CPU时钟周期时,获取的时之前的旧值
                // 或超时
                if ((wc > maximumPoolSize || (timed && timedOut))
                        //当前线程个数>1,且工作队列为空
                    && (wc > 1 || workQueue.isEmpty())) {
        
                    //当前线程数-1,成功返回null,结束当前线程执行任务
                    if (compareAndDecrementWorkerCount(c))
                        return null;
                    continue;
                }
    
                try {
        
                    //如果设置了超时,则阻塞获取,否则直接获取
                    Runnable r = timed ?
                        workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                        workQueue.take();
                    if (r != null)
                        return r;
                    //r为null,说明超时,下次自旋回收
                    timedOut = true;
                } catch (InterruptedException retry) {
        
                    //出现中断,设置超时为false,并循环重试
                    timedOut = false;
                }
            }
        }

private void processWorkerExit(Worker w, boolean completedAbruptly) {
        
            //没正常执行成功,正常执行成功,则不需要因为getTask执行了-1操作
            if (completedAbruptly) // If abrupt, then workerCount wasn't adjusted
                //CAS 当前线程数量-1
                decrementWorkerCount();
    
            final ReentrantLock mainLock = this.mainLock;
            mainLock.lock();
            try {
        
                //统计所有线程执行成功次数
                completedTaskCount += w.completedTasks;
                //删除当前工作线程
                workers.remove(w);
            } finally {
        
                mainLock.unlock();
            }
    
            //终止线程池
            tryTerminate();
    
            int c = ctl.get();
            //如果是否<stop,也就是SHUTDOWN和RUNNING状态
            if (runStateLessThan(c, STOP)) {
        
                //是否正常执行成功
                if (!completedAbruptly) {
        
                    //是否允许核心线程超时
                    int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
                    //允许核心线程超时,且队列不为空，设置最少存活一个线程处理任务
                    if (min == 0 && ! workQueue.isEmpty())
                        min = 1;
                    //当前线程个数>1,则直接返回
                    if (workerCountOf(c) >= min)
                        return; // replacement not needed
                }
                //试着创建线程最大线程数新开一个新线程
                addWorker(null, false);
            }
        }

## shutdown ##

public void shutdown() {
        
            final ReentrantLock mainLock = this.mainLock;
            mainLock.lock();
            try {
        
                checkShutdownAccess();
                //设置线程池状态
                advanceRunState(SHUTDOWN);
                //中断所有线程
                interruptIdleWorkers();
                //钩子方法,子类实现
                onShutdown(); // hook for ScheduledThreadPoolExecutor
            } finally {
        
                mainLock.unlock();
            }
            //终止线程池
            tryTerminate();
        }
        private void advanceRunState(int targetState) {
        
            for (;;) {
        
                int c = ctl.get();
                //c是否>=targetState状态
                if (runStateAtLeast(c, targetState) ||
                        //设置成目标状态
                    ctl.compareAndSet(c, ctlOf(targetState, workerCountOf(c))))
                    break;
            }
        }
    
        private void interruptIdleWorkers() {
        
            //中断所有线程
            //onlyOne
            //true表示只中断一个
            //false:中断所有
            interruptIdleWorkers(false);
        }

## shutdownNow ##

public List<Runnable> shutdownNow() {
        
            List<Runnable> tasks;
            final ReentrantLock mainLock = this.mainLock;
            mainLock.lock();
            try {
        
                checkShutdownAccess();
                //设置状态为stop
                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();
            }
        }

void interruptIfStarted() {
        
                Thread t;
                //允许中断且当前线程不为null,且测试是否已经中断
                if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {
        
                    try {
        
                        //中断
                        t.interrupt();
                    } catch (SecurityException ignore) {
        
                    }
                }
            }

private List<Runnable> drainQueue() {
        
            BlockingQueue<Runnable> q = workQueue;
            ArrayList<Runnable> taskList = new ArrayList<Runnable>();
            //移除此队列中所有可用的元素，并将它们添加到给定 collection 中。
            q.drainTo(taskList);
            if (!q.isEmpty()) {
        
                for (Runnable r : q.toArray(new Runnable[0])) {
        
                    if (q.remove(r))
                        //添加
                        taskList.add(r);
                }
            }
            return taskList;
        }

## 总结 ##

使用Worker\#lock和unlock分别控制是否能够中断.  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L051YW5fRmVuZw_size_16_color_FFFFFF_t_70 4]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L051YW5fRmVuZw_size_16_color_FFFFFF_t_70 5]  
执行shutdown的时候,会中断所有空闲线程,繁忙的线程则不处理  
shutdownNow中断所有可以标记了unlock的线程,且返回尚未执行的所有任务  
每个线程和任务绑定使用,当任务执行完成后,线程复用.  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L051YW5fRmVuZw_size_16_color_FFFFFF_t_70 6]

# ScheduledThreadPoolExecutor #

## //todo ##

# Thread #

## 构造 ##

private void init(ThreadGroup g, Runnable target, String name,
                          long stackSize, AccessControlContext acc,
                          boolean inheritThreadLocals) {
        
            if (name == null) {
        
                throw new NullPointerException("name cannot be null");
            }
    
            //线程名,默认"Thread-" + nextThreadNum()
            this.name = name;
    
            //获取当前线程
            Thread parent = currentThread();
            SecurityManager security = System.getSecurityManager();
            //线程组为空
            if (g == null) {
        
                /* Determine if it's an applet or not */
    
                /* If there is a security manager, ask the security manager
                   what to do. */
                if (security != null) {
        
                    g = security.getThreadGroup();
                }
    
                /* If the security doesn't have a strong opinion of the matter
                   use the parent thread group. */
                if (g == null) {
        
                    //默认继承父线程线程组
                    g = parent.getThreadGroup();
                }
            }
    
            /* checkAccess regardless of whether or not threadgroup is
               explicitly passed in. */
            g.checkAccess();
    
            /*
             * Do we have the required permissions?
             */
            if (security != null) {
        
                if (isCCLOverridden(getClass())) {
        
                    security.checkPermission(SUBCLASS_IMPLEMENTATION_PERMISSION);
                }
            }
    
            g.addUnstarted();
    
            this.group = g;
            //设置是否为守护线程,继承调用线程的主线程,main默认是false
            this.daemon = parent.isDaemon();
            //默认5
            this.priority = parent.getPriority();
            if (security == null || isCCLOverridden(parent.getClass()))
                this.contextClassLoader = parent.getContextClassLoader();
            else
                this.contextClassLoader = parent.contextClassLoader;
            this.inheritedAccessControlContext =
                    acc != null ? acc : AccessController.getContext();
            this.target = target;
            setPriority(priority);
            if (inheritThreadLocals && parent.inheritableThreadLocals != null)
                //创建线程共享变量副本
                this.inheritableThreadLocals =
                    ThreadLocal.createInheritedMap(parent.inheritableThreadLocals);
            /* Stash the specified stack size in case the VM cares */
            //设置栈大小，如果未指定大小，将在jvm 初始化参数中声明：Xss参数进行指定*/
            this.stackSize = stackSize;
    
            /* Set thread ID */
            //设置线程id
            tid = nextThreadID();
        }

## start ##

public synchronized void start() {
        
            /**
             * This method is not invoked for the main method thread or "system"
             * group threads created/set up by the VM. Any new functionality added
             * to this method in the future may have to also be added to the VM.
             *
             * A zero status value corresponds to state "NEW".
             */
            //当前线程状态
            if (threadStatus != 0)
                throw new IllegalThreadStateException();
    
            /* Notify the group that this thread is about to be started
             * so that it can be added to the group's list of threads
             * and the group's unstarted count can be decremented. */
            //当前线程加入线程组
            group.add(this);
    
            boolean started = false;
            try {
        
                //启动
                start0();
                //标记正常结束
                started = true;
            } finally {
        
                try {
        
                    if (!started) {
        
                       //线程启动失败，从线程组里面移除该线程
                        group.threadStartFailed(this);
                    }
                } catch (Throwable ignore) {
        
                    /* do nothing. If start0 threw a Throwable then
                      it will be passed up the call stack */
                }
            }
        }

## interrupt ##

public void interrupt() {
        
            //如果不是当前线程
            if (this != Thread.currentThread())
                //判断当前线程是否允许修改其他线程
                checkAccess();
    
            //中断
            synchronized (blockerLock) {
        
                Interruptible b = blocker;
                if (b != null) {
        
                    //设置中断标识位
                    interrupt0();           // Just to set the interrupt flag
                    b.interrupt(this);
                    return;
                }
            }
            interrupt0();
        }

## join ##

public final synchronized void join(long millis)
        throws InterruptedException {
        
            long base = System.currentTimeMillis();
            long now = 0;
    
            if (millis < 0) {
        
                throw new IllegalArgumentException("timeout value is negative");
            }
    
            //等待该线程终止的时间最长为 millis 毫秒。
            if (millis == 0) {
        
                //测试线程是否处于活动状态。如果线程已经启动且尚未终止，则为活动状态。
                while (isAlive()) {
        
                    wait(0);
                }
            } else {
        
                //指定了超时时间
                while (isAlive()) {
        
                    long delay = millis - now;
                    //超时,结束
                    if (delay <= 0) {
        
                        break;
                    }
                    //等待阻塞
                    wait(delay);
                    now = System.currentTimeMillis() - base;
                }
            }
        }

![在这里插入图片描述][format_png]

## State ##

/* NEW：初始状态，线程被构建，还未调用start()方法；
        RUNNABLE：运行状态，在java多线程模型中，就绪和运行都是运行状态；
        BLOCKED：阻塞状态；
        WAITING：等待状态，比如中断，需要其他的线程来唤醒；
        TIME_WAITING：超时等待，可以在指定的时间内自行返回；
        TERMINATED：终止状态，线程执行完毕。*/
        public enum State {
        }

## run ##

public void run() {
        
            if (target != null) {
        
                target.run();
            }
        }

## 总结 ##

thread很多方法算是底层调用的了,  
设置优先级调用,取决于操作系统

# ThreadFactory #

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L051YW5fRmVuZw_size_16_color_FFFFFF_t_70 7]  
基本上都是当做内部类使用,随便看一个实现即可  
java.util.concurrent.Executors.DefaultThreadFactory

## DefaultThreadFactory ##

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() {
        
                SecurityManager s = System.getSecurityManager();
                //默认继承父类线程组
                group = (s != null) ? s.getThreadGroup() :
                                      Thread.currentThread().getThreadGroup();
    
                //pool-自增1开始-thread-
                namePrefix = "pool-" +
                              poolNumber.getAndIncrement() +
                             "-thread-";
            }
    
            public Thread newThread(Runnable r) {
        
                //pool-1-thread-1
                Thread t = new Thread(group, r,
                                      namePrefix + threadNumber.getAndIncrement(),
                                      0);
                //是否是守护线程
                if (t.isDaemon())
                    //如果是守护线程,则设置成普通线程
                    t.setDaemon(false);
                //
                //设置优先级,默认5
                if (t.getPriority() != Thread.NORM_PRIORITY)
                    t.setPriority(Thread.NORM_PRIORITY);
                return t;
            }
        }

# Executors #

public static ExecutorService newCachedThreadPool(ThreadFactory threadFactory) {
        
            //线程最大设置Integer.MAX,基本大小=0,超时设置1分钟
            return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                          60L, TimeUnit.SECONDS,
                                          new SynchronousQueue<Runnable>(),
                                          threadFactory);
        }
    
        public static ExecutorService newFixedThreadPool(int nThreads, ThreadFactory threadFactory) {
        
      //创建一个可重用固定线程数的线程池，以共享的无界队列方式来运行这些线程
            return new ThreadPoolExecutor(nThreads, nThreads,
                                          0L, TimeUnit.MILLISECONDS,
                                          new LinkedBlockingQueue<Runnable>(),
                                          threadFactory);
        }
    
        public static ScheduledExecutorService newScheduledThreadPool(
                int corePoolSize, ThreadFactory threadFactory) {
        
            //创建一个线程池，它可安排在给定延迟后运行命令或者定期地执行。
            return new ScheduledThreadPoolExecutor(corePoolSize, threadFactory);
        }
    
       public static ExecutorService newSingleThreadExecutor() {
        
            //创建一个使用单个 worker 线程的 Executor，以无界队列方式来运行该线程。
            return new FinalizableDelegatedExecutorService
                (new ThreadPoolExecutor(1, 1,
                                        0L, TimeUnit.MILLISECONDS,
                                        new LinkedBlockingQueue<Runnable>()));
        }

# ExecutorCompletionService #

阻塞获取结果

## 构造 ##

public ExecutorCompletionService(Executor executor) {
        
            if (executor == null)
                throw new NullPointerException();
            this.executor = executor;
            this.aes = (executor instanceof AbstractExecutorService) ?
                (AbstractExecutorService) executor : null;
            //使用LinkedBlockingQueue保存结果
            this.completionQueue = new LinkedBlockingQueue<Future<V>>();
        }
    
    
        public ExecutorCompletionService(Executor executor,
                                         BlockingQueue<Future<V>> completionQueue) {
        
            if (executor == null || completionQueue == null)
                throw new NullPointerException();
            this.executor = executor;
            this.aes = (executor instanceof AbstractExecutorService) ?
                (AbstractExecutorService) executor : null;
            //自定义阻塞队列
            this.completionQueue = completionQueue;
        }

## submit ##

public Future<V> submit(Callable<V> task) {
        
            if (task == null) throw new NullPointerException();
            //封装成FutureTask
            RunnableFuture<V> f = newTaskFor(task);
            //封装成阻塞结果的FutureTask,执行
            executor.execute(new QueueingFuture(f));
            return f;
        }

private class QueueingFuture extends FutureTask<Void> {
        
            QueueingFuture(RunnableFuture<V> task) {
        
                super(task, null);
                this.task = task;
            }
            protected void done() {
        
                //实现父类,完成后添加到阻塞队列
                completionQueue.add(task); }
            private final Future<V> task;
        }

## take ##

public Future<V> take() throws InterruptedException {
        
            //阻塞获取结果
            return completionQueue.take();
        }

## poll ##

public Future<V> poll() {
        
            //直接获取结果,没有返回null
            return completionQueue.poll();
        }

## 总结 ##

底层封装阻塞队列,且封装了FutureTask,当执行结束,把Future存储到阻塞队列中.

[Link 1]: https://tech.meituan.com/2020/04/02/java-pooling-pratice-in-meituan.html
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L051YW5fRmVuZw_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20200722121809189.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L051YW5fRmVuZw==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L051YW5fRmVuZw_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20200528000143654.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L051YW5fRmVuZw==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L051YW5fRmVuZw_size_16_color_FFFFFF_t_70 2]: https://img-blog.csdnimg.cn/20200528000213932.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L051YW5fRmVuZw==,size_16,color_FFFFFF,t_70
[2020060210092042.png]: https://img-blog.csdnimg.cn/2020060210092042.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L051YW5fRmVuZw_size_16_color_FFFFFF_t_70 3]: https://img-blog.csdnimg.cn/20200602110509446.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L051YW5fRmVuZw==,size_16,color_FFFFFF,t_70
[20200602111052547.png]: https://img-blog.csdnimg.cn/20200602111052547.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L051YW5fRmVuZw_size_16_color_FFFFFF_t_70 4]: https://img-blog.csdnimg.cn/20200602184225371.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L051YW5fRmVuZw==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L051YW5fRmVuZw_size_16_color_FFFFFF_t_70 5]: https://img-blog.csdnimg.cn/20200602184244809.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L051YW5fRmVuZw==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L051YW5fRmVuZw_size_16_color_FFFFFF_t_70 6]: https://img-blog.csdnimg.cn/2020060218475787.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L051YW5fRmVuZw==,size_16,color_FFFFFF,t_70
[format_png]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9pbWFnZXMyMDE1LmNuYmxvZ3MuY29tL2Jsb2cvNjc3MDU0LzIwMTcwNC82NzcwNTQtMjAxNzA0MDExMzU5MjgwODYtMTc0NTE4OTYyNy5qcGc?x-oss-process=image/format,png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L051YW5fRmVuZw_size_16_color_FFFFFF_t_70 7]: https://img-blog.csdnimg.cn/20200602192109685.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L051YW5fRmVuZw==,size_16,color_FFFFFF,t_70