• Java线程池源码解析

    1. 线程池的继承关系

    从上图可以看出来最顶层的接口为 Executor ,下面看一下这个接口中的方法

    1. public interface Executor {
    2.     //只有一个方法execute
    3.     void execute(Runnable command);
    4. }

    从代码中可以看出来只有一个 execute 方法。这也是我常用的一个来运行 Runable 的一种方式。然后看一下继承了 Executor 接口的 ExecutorService 接口中有哪些我们熟悉的而常用的方法

    1. public interface ExecutorService extends Executor {
    2.     
    3.     // 关闭线程池，已提交的任务继续执行，不接受继续提交新任务
    4.     //写例子的时候用到(PS在实际的项目组基本上没有用到，反正我是没有)
    5.     void shutdown();
    6.     
    7.     //关闭线程池，尝试停止正在执行的所有任务，不接受继续提交新任务
    8.      //这个也是基本上没用到
    9.     List<Runnable> shutdownNow();
    10. ​
    11.     // 线程池是否已关闭
    12.     // 还是没有用到
    13.     boolean isShutdown();
    14. ​
    15.     // 这个方法必须在调用shutdown或shutdownNow方法之后调用才会返回true
    16.     //尴尬没用过
    17.     boolean isTerminated();
    18. ​
    19.     //一脸懵逼没用过
    20.     boolean awaitTermination(long timeout, TimeUnit unit)
    21.         throws InterruptedException;
    22. ​
    23.    //带返回值的
    24.     <T> Future<T> submit(Callable<T> task);
    25. ​
    26.     //带返回值的 -- 这个很少用
    27.     <T> Future<T> submit(Runnable task, T result);
    28. ​
    29.     //带返回值---(成功返回值为null 有兴趣的可以去尝试一下，源码的英文注释上面有说明)
    30.     Future<?> submit(Runnable task);
    31. ​
    32.     //批量全部执行
    33.     <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
    34.         throws InterruptedException;
    35. ​
    36.     //批量全部执行--在规定的时间内
    37.     <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,
    38.                                   long timeout, TimeUnit unit)
    39.         throws InterruptedException;
    40. ​
    41.    //任意一个先执行完就返回
    42.     <T> T invokeAny(Collection<? extends Callable<T>> tasks)
    43.         throws InterruptedException, ExecutionException;
    44. ​
    45.     //任意一个先执行完就返回
    46.     <T> T invokeAny(Collection<? extends Callable<T>> tasks,
    47.                     long timeout, TimeUnit unit)
    48.         throws InterruptedException, ExecutionException, TimeoutException;
    49. }

    演示代码：

    1. public class InvokeAllTest {
    2. ​
    3.     public static void main(String[] args) throws  Exception{
    4.         ExecutorService service = Executors.newFixedThreadPool(10);
    5.         Collection<Test> a = new ArrayList<>();
    6.         for(int i = 0; i < 10; ++i){
    7.             a.add(new Test());
    8.         }
    9.         //System.out.println( service.invokeAny(a));
    10.         System.out.println( service.invokeAll(a));
    11.     }
    12. ​
    13. }
    14. ​
    15. class Test implements Callable<String>{
    16. ​
    17.     /**
    18.      * Computes a result, or throws an exception if unable to do so.
    19.      *
    20.      * @return computed result
    21.      * @throws Exception if unable to compute a result
    22.      */
    23.     @Override
    24.     public String call() throws Exception {
    25. ​
    26.         TimeUnit.SECONDS.sleep((int)(Math.random()*10));
    27. ​
    28.         return Thread.currentThread().getName();
    29.     }
    30. }

    看一下最后一个接口 ScheduledExecutorService 计划执行接口，从命名上就不难看出来这个用于执行任务的。

    1. public interface ScheduledExecutorService extends ExecutorService {
    2. ​
    3.     /**
    4.      * 创建并执行在给定延迟之后启用的一次性操作。
    5.      */
    6.     public ScheduledFuture<?> schedule(Runnable command,
    7.                                        long delay, TimeUnit unit);
    8. ​
    9.     /**
    10.      * 创建并执行在给定延迟之后启用的一次性操作。返回ScheduledFuture<V>
    11.      */
    12.     public <V> ScheduledFuture<V> schedule(Callable<V> callable,
    13.                                            long delay, TimeUnit unit);
    14. ​
    15.     /**
    16.      * 按指定频率周期执行某个任务。
    17.      */
    18.     public ScheduledFuture<?> scheduleAtFixedRate(Runnable command,
    19.                                                   long initialDelay,
    20.                                                   long period,
    21.                                                   TimeUnit unit);
    22. ​
    23.     /**
    24.      * 按指定频率间隔执行某个任务。
    25.      */
    26.     public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command,
    27.                                                      long initialDelay,
    28.                                                      long delay,
    29.                                                      TimeUnit unit);
    30. ​
    31. }

    另外，由于线程池支持获取线程执行的结果，所以，引入了 Future 接口，RunnableFuture 继承自此接口，然后我们最需要关心的就是它的实现类 FutureTask。到这里，记住这个概念，在线程池的使用过程中，我们是往线程池提交任务（task），使用过线程池的都知道，我们提交的每个任务是实现了 Runnable 接口的，其实就是先将 Runnable 的任务包装成 FutureTask，然后再提交到线程池。这样，读者才能比较容易记住 FutureTask 这个类名：它首先是一个任务（Task），然后具有 Future 接口的语义，即可以在将来（Future）得到执行的结果。

    2. AbstractExecutorService

    接着来看一下在抽象类 AbstractExecutorService 实现了哪些方法

    1. public abstract class AbstractExecutorService implements ExecutorService {
    2. ​
    3.     /**
    4.      * Runnable 转换为 Callable 的方法带指定返回值
    5.      */
    6.     protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) {
    7.         return new FutureTask<T>(runnable, value);
    8.     }
    9. ​
    10.     /**
    11.      * Runnable 转换为 Callable 的方法，不带指定返回值
    12.      */
    13.     protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
    14.         return new FutureTask<T>(callable);
    15.     }
    16. ​
    17.     /**
    18.      * 
    19.      */
    20.     public Future<?> submit(Runnable task) {
    21.         if (task == null) throw new NullPointerException();
    22.         //这里看一看出来在Runnable submit方法返回值为Future get的值为null
    23.         RunnableFuture<Void> ftask = newTaskFor(task, null);
    24.         execute(ftask);
    25.         return ftask;
    26.     }
    27. ​
    28.     public <T> Future<T> submit(Runnable task, T result) {
    29.         if (task == null) throw new NullPointerException();
    30.          //这里看一看出来在Runnable submit方法返回值为Future get的值为result
    31.         RunnableFuture<T> ftask = newTaskFor(task, result);
    32.         execute(ftask);
    33.         return ftask;
    34.     }
    35. ​
    36.     /**
    37.      * Callable类型
    38.      */
    39.     public <T> Future<T> submit(Callable<T> task) {
    40.         if (task == null) throw new NullPointerException();
    41.         RunnableFuture<T> ftask = newTaskFor(task);
    42.         execute(ftask);
    43.         return ftask;
    44.     }
    45. ​
    46.     /**
    47.      * 返回任意一个执行完成的结果
    48.      */
    49.     private <T> T doInvokeAny(Collection<? extends Callable<T>> tasks,
    50.                               boolean timed, long nanos)
    51.         throws InterruptedException, ExecutionException, TimeoutException {
    52.         if (tasks == null)
    53.             throw new NullPointerException();
    54.         int ntasks = tasks.size();
    55.         if (ntasks == 0)
    56.             throw new IllegalArgumentException();
    57.         //Future列表
    58.         ArrayList<Future<T>> futures = new ArrayList<Future<T>>(ntasks);
    59.         ExecutorCompletionService<T> ecs =
    60.             new ExecutorCompletionService<T>(this);
    61.         try {
    62.             
    63.             ExecutionException ee = null;
    64.             //截止时间---0就是没有截止时间
    65.             final long deadline = timed ? System.nanoTime() + nanos : 0L;
    66.             Iterator<? extends Callable<T>> it = tasks.iterator();
    67. ​
    68.            
    69.             futures.add(ecs.submit(it.next()));
    70.             --ntasks;
    71.             int active = 1;
    72. ​
    73.             for (;;) {
    74.                 //返回已经完成的任务Future<T> 没有就返回null -- 不停的循环轮询
    75.                 Future<T> f = ecs.poll();
    76.                 if (f == null) {
    77.                     if (ntasks > 0) {
    78.                         --ntasks;
    79.                         futures.add(ecs.submit(it.next()));
    80.                         ++active;
    81.                     }
    82.                     else if (active == 0)
    83.                         break;
    84.                     else if (timed) {
    85.                         f = ecs.poll(nanos, TimeUnit.NANOSECONDS);
    86.                         if (f == null)
    87.                             throw new TimeoutException();
    88.                         nanos = deadline - System.nanoTime();
    89.                     }
    90.                     else
    91.                         f = ecs.take();
    92.                 }
    93.                 if (f != null) {
    94.                     --active;
    95.                     try {
    96.                         return f.get();
    97.                     } catch (ExecutionException eex) {
    98.                         ee = eex;
    99.                     } catch (RuntimeException rex) {
    100.                         ee = new ExecutionException(rex);
    101.                     }
    102.                 }
    103.             }
    104. ​
    105.             if (ee == null)
    106.                 ee = new ExecutionException();
    107.             throw ee;
    108. ​
    109.         } finally {
    110.             for (int i = 0, size = futures.size(); i < size; i++)
    111.                 futures.get(i).cancel(true);
    112.         }
    113.     }
    114. ​
    115.     public <T> T invokeAny(Collection<? extends Callable<T>> tasks)
    116.         throws InterruptedException, ExecutionException {
    117.         try {
    118.             return doInvokeAny(tasks, false, 0);
    119.         } catch (TimeoutException cannotHappen) {
    120.             assert false;
    121.             return null;
    122.         }
    123.     }
    124. ​
    125.     public <T> T invokeAny(Collection<? extends Callable<T>> tasks,
    126.                            long timeout, TimeUnit unit)
    127.         throws InterruptedException, ExecutionException, TimeoutException {
    128.         return doInvokeAny(tasks, true, unit.toNanos(timeout));
    129.     }
    130. ​
    131.     public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
    132.         throws InterruptedException {
    133.         if (tasks == null)
    134.             throw new NullPointerException();
    135.         //创建返回值Future 的列表
    136.         ArrayList<Future<T>> futures = new ArrayList<Future<T>>(tasks.size());
    137.         boolean done = false;
    138.         try {
    139.             //放入线程池运行
    140.             for (Callable<T> t : tasks) {
    141.                 RunnableFuture<T> f = newTaskFor(t);
    142.                 futures.add(f);
    143.                 execute(f);
    144.             }
    145.             //等待运行完成
    146.             for (int i = 0, size = futures.size(); i < size; i++) {
    147.                 Future<T> f = futures.get(i);
    148.                 if (!f.isDone()) {
    149.                     try {
    150.                         f.get();
    151.                     } catch (CancellationException ignore) {
    152.                     } catch (ExecutionException ignore) {
    153.                     }
    154.                 }
    155.             }
    156.             done = true;
    157.             return futures;
    158.         } finally {
    159.             if (!done)
    160.                 //将没有运行完成的线程直接取消掉
    161.                 for (int i = 0, size = futures.size(); i < size; i++)
    162.                     futures.get(i).cancel(true);
    163.         }
    164.     }
    165. ​
    166.     public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,
    167.                                          long timeout, TimeUnit unit)
    168.         throws InterruptedException {
    169.         if (tasks == null)
    170.             throw new NullPointerException();
    171.         long nanos = unit.toNanos(timeout);
    172.         ArrayList<Future<T>> futures = new ArrayList<Future<T>>(tasks.size());
    173.         boolean done = false;
    174.         try {
    175.             //创建任务数组
    176.             for (Callable<T> t : tasks)
    177.                 futures.add(newTaskFor(t));
    178.             //截止时间
    179.             final long deadline = System.nanoTime() + nanos;
    180.             final int size = futures.size();
    181.             
    182.             //减去提交的时间
    183.             for (int i = 0; i < size; i++) {
    184.                 execute((Runnable)futures.get(i));
    185.                 nanos = deadline - System.nanoTime();
    186.                 //小于0直接返回现有的
    187.                 if (nanos <= 0L)
    188.                     return futures;
    189.             }
    190.             
    191.             //处理每个获取的时间
    192.             for (int i = 0; i < size; i++) {
    193.                 Future<T> f = futures.get(i);
    194.                 if (!f.isDone()) {
    195.                     if (nanos <= 0L)
    196.                         return futures;
    197.                     try {
    198.                         f.get(nanos, TimeUnit.NANOSECONDS);
    199.                     } catch (CancellationException ignore) {
    200.                     } catch (ExecutionException ignore) {
    201.                     } catch (TimeoutException toe) {
    202.                         //发现timeout直接返回
    203.                         return futures;
    204.                     }
    205.                     nanos = deadline - System.nanoTime();
    206.                 }
    207.             }
    208.             done = true;
    209.             return futures;
    210.         } finally {
    211.             if (!done)
    212.                 //返回后发现还有在运行的直接cacel掉
    213.                 for (int i = 0, size = futures.size(); i < size; i++)
    214.                     futures.get(i).cancel(true);
    215.         }
    216.     }
    217. ​
    218. }

    从上面可以看出来上面方法实现主要是通过调用 execute 和 ExecutorCompletionService 这个类。来实现了 submit , doInvokeAny ,invokeAll 这些方法。

    3. 看看最常用的实现 ThreadPoolExecutor

    首先我们来看一下 ThreadPoolExecutor 类中包含的成员变量进行逐一的分析

    1. //从开始的继承图可以看出来 ThreadPoolExecutor继承了AbstractExecutorService
    2. public class ThreadPoolExecutor extends AbstractExecutorService {
    3.     /**
    4.      * 主线程池控制状态ctl是一个atomic整型封装了两个概念字段
    5.      * 
    6.      *   线程数量, 定义了有效的线程数量
    7.      *   运行状态,    定义了：运行状态，关闭状态等等。
    8.      *
    9.      * 为了封装成一个整数, 我们限制线程的数量为
    10.      * (2^29)-1 (about 500 million) 而不是 (2^31)-1
    11.      * 如果将来出现这种情况，可以将变量更改为AtomicLong，并调整下面的shift/mask常量。
    12.      * 但是在需要之前，这段代码使用int会更快、更简单。
    13.      * 工作线程是允许启动和停止的，工作线程可能会有和存活的线程有短暂的数量不同
    14.      *
    15.      *   RUNNING:  接受新任务并处理排队的任务
    16.      *   SHUTDOWN: 不接受新任务但是处理排队任务
    17.      *   STOP:     不接受新任务不接受排队任务，并且中断在处理中的任务
    18.      *             
    19.      *   TIDYING(整理):  所有的任务中止, 工作线程为0，转换到状态清理的线程将运行terminate()钩子方法
    20.      *             
    21.      *   TERMINATED: terminated() 已经完成
    22.      *
    23.      * The numerical order among these values matters, to allow
    24.      * ordered comparisons. The runState monotonically increases over
    25.      * time, but need not hit each state. The transitions are:
    26.      *
    27.      * RUNNING -> SHUTDOWN
    28.      *     调用shutdown(),或者在finalize()中调用shutdown()
    29.      * (RUNNING or SHUTDOWN) -> STOP
    30.      *    调用shutdownNow()
    31.      * SHUTDOWN -> TIDYING
    32.      *    当队列和线程池都为空的时候
    33.      * STOP -> TIDYING
    34.      *    当线程池为空
    35.      * TIDYING -> TERMINATED
    36.      *    terminated() 方法执行完成
    37.      *
    38.      */
    39.     private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
    40.     //线程的数量的表示位--低29位表示线程数量
    41.     private static final int COUNT_BITS = Integer.SIZE - 3;
    42.     //最大的线程的容量(2^29)-1
    43.     private static final int CAPACITY   = (1 << COUNT_BITS) - 1;
    44. ​
    45.     // runState 用int的高三位表示
    46.     //11100000000000000000000000000000
    47.     private static final int RUNNING    = -1 << COUNT_BITS;
    48.     
    49.     //00000000000000000000000000000000
    50.     private static final int SHUTDOWN   =  0 << COUNT_BITS;
    51.     
    52.     //00100000000000000000000000000000
    53.     private static final int STOP       =  1 << COUNT_BITS;
    54.     
    55.     //01000000000000000000000000000000
    56.     private static final int TIDYING    =  2 << COUNT_BITS;
    57.     
    58.     //01100000000000000000000000000000
    59.     private static final int TERMINATED =  3 << COUNT_BITS;
    60. ​
    61.     // 拆解出运行状态
    62.     private static int runStateOf(int c)     { return c & ~CAPACITY; }
    63.     
    64.     //拆解出来线程数量
    65.     private static int workerCountOf(int c)  { return c & CAPACITY; }
    66.     
    67.     //把运行状态和线程数量打包成一个整数
    68.     private static int ctlOf(int rs, int wc) { return rs | wc; }
    69. ​
    70.     /**
    71.      * 线程的增加和减少都是通过CAS来进行的
    72.      */
    73.     private boolean compareAndIncrementWorkerCount(int expect) {
    74.         return ctl.compareAndSet(expect, expect + 1);
    75.     }
    76. ​
    77.     private boolean compareAndDecrementWorkerCount(int expect) {
    78.         return ctl.compareAndSet(expect, expect - 1);
    79.     }
    80. ​
    81. ​
    82.     private void decrementWorkerCount() {
    83.         do {} while (! compareAndDecrementWorkerCount(ctl.get()));
    84.     }
    85. ​
    86.     /**
    87.      * 阻塞队列
    88.      */
    89.     private final BlockingQueue<Runnable> workQueue;
    90. ​
    91.     /**
    92.      * 非公平的重入锁
    93.      */
    94.     private final ReentrantLock mainLock = new ReentrantLock();
    95. ​
    96.     /**
    97.      * 仅在持有主锁mainLock时访问
    98.      * .
    99.      */
    100.     private final HashSet<Worker> workers = new HashSet<Worker>();
    101. ​
    102. ​
    103.     private final Condition termination = mainLock.newCondition();
    104. ​
    105. ​
    106.     private int largestPoolSize;
    107. ​
    108. ​
    109.     private long completedTaskCount;
    110. ​
    111. ​
    112.     private volatile ThreadFactory threadFactory;
    113. ​
    114.     /**
    115.      * 当执行饱和或关闭时调用处理Handler
    116.      */
    117.     private volatile RejectedExecutionHandler handler;
    118. ​
    119.     /**
    120.      * 闲置的线程等待超时时间
    121.      */
    122.     private volatile long keepAliveTime;
    123. ​
    124.     /**
    125.      * 是否允许核心线程超时
    126.      */
    127.     private volatile boolean allowCoreThreadTimeOut;
    128. ​
    129.     /**
    130.      * 核心线程池的大小
    131.      */
    132.     private volatile int corePoolSize;
    133. ​
    134.     /**
    135.      * 线程的极大值
    136.      * 
    137.      */
    138.     private volatile int maximumPoolSize;
    139. ​
    140.     /**
    141.      * 默认的被拒执行的Handler
    142.      */
    143.     private static final RejectedExecutionHandler defaultHandler =
    144.         new AbortPolicy();
    145. ​
    146.     //Worker实现了AQS和Runnable的接口
    147.     private final class Worker
    148.         extends AbstractQueuedSynchronizer
    149.         implements Runnable
    150.     {
    151.         
    152.         private static final long serialVersionUID = 6138294804551838833L;
    153. ​
    154.        
    155.         final Thread thread;
    156.        
    157.         Runnable firstTask;
    158.        
    159.         volatile long completedTasks;
    160. ​
    161.         Worker(Runnable firstTask) {
    162.             setState(-1); // inhibit interrupts until runWorker
    163.             this.firstTask = firstTask;
    164.             this.thread = getThreadFactory().newThread(this);
    165.         }
    166. ​
    167.         /** Delegates main run loop to outer runWorker  */
    168.         public void run() {
    169.             runWorker(this);
    170.         }
    171. ​
    172.         protected boolean isHeldExclusively() {
    173.             return getState() != 0;
    174.         }
    175. ​
    176.         protected boolean tryAcquire(int unused) {
    177.             if (compareAndSetState(0, 1)) {
    178.                 setExclusiveOwnerThread(Thread.currentThread());
    179.                 return true;
    180.             }
    181.             return false;
    182.         }
    183. ​
    184.         protected boolean tryRelease(int unused) {
    185.             setExclusiveOwnerThread(null);
    186.             setState(0);
    187.             return true;
    188.         }
    189. ​
    190.         public void lock()        { acquire(1); }
    191.         public boolean tryLock()  { return tryAcquire(1); }
    192.         public void unlock()      { release(1); }
    193.         public boolean isLocked() { return isHeldExclusively(); }
    194. ​
    195.         void interruptIfStarted() {
    196.             Thread t;
    197.             if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {
    198.                 try {
    199.                     t.interrupt();
    200.                 } catch (SecurityException ignore) {
    201.                 }
    202.             }
    203.         }
    204.     }

    4. execute 的实现

    从上面可以看出来上面方法实现主要是通过调用 execute 和 ExecutorCompletionService 这个类。来实现了 submit , doInvokeAny ,invokeAll 这些方法。下面就来看一下 execute 这方法在 ThreadPoolExecutor 中是如何实现的。

    1. public void execute(Runnable command) {
    2.         if (command == null)
    3.             throw new NullPointerException();
    4.         /*
    5.          * Proceed in 3 steps:
    6.          *
    7.          * 1 如果运行的线程小于corePoolSize，则尝试以给定命令作为其第一个任务启动新线程。对
    8.          * addWorker的调用以原子方式检查runState和workerCount，从而通过返回false防止在不
    9.          * 应该添加线程时添加错误警报。
    10.          * 
    11.          * 2. 如果一个任务可以成功地排队，那么我们仍然需要再次检查是否应该添加线程(因为上次检
    12.          * 查后已有线程死亡)，或者是否应该在进入这个方法后关闭线程池。因此，我们重新检查状
    13.          * 态，如果有必要的话，在停止时回滚队列，或者在没有线程时启动新线程。
    14.          *
    15.          * 3. 如果无法对任务排队，则尝试添加新线程。如果它失败了，我们知道我们被关闭或饱和，所以拒绝任务。
    16.          * 所以拒绝任务。
    17.          * 
    18.          */
    19.      
    20.         //获取线程池中线程数量---默认是0
    21.         int c = ctl.get();
    22.         //如果设置了核心线程数先判断核心线程数是不是已经满了
    23.         if (workerCountOf(c) < corePoolSize) {
    24.             if (addWorker(command, true))
    25.                 return;
    26.             c = ctl.get();
    27.         }
    28.         //判断线程池是否处于运行状态并且还能往队列添加任务
    29.         if (isRunning(c) && workQueue.offer(command)) {
    30.             int recheck = ctl.get();
    31.             //双重检查---如果不是运行状态从队列中删除任务
    32.             if (! isRunning(recheck) && remove(command))
    33.                 //根据传入的不同策略处理器处理问题
    34.                 reject(command);
    35.             else if (workerCountOf(recheck) == 0)
    36.                 addWorker(null, false);
    37.         }
    38.         //添加非核心线程任务
    39.         else if (!addWorker(command, false))
    40.             //添加失败根据传入的不同的策略处理器处理问题
    41.             reject(command);
    42.     }

    execute 方法主要做了三件事情：

    1. 添加核心处理线程
    2. 线程池在运行状态，添加任务到任务阻塞队列中
    3. 新增非核心线程处理

    在线程池构造函数中有设置keepAliveTime，这个设置的就是非coreThread的存活时间。

    通过上面的源码发现主要是 addWorker 方法：

    1.  private boolean addWorker(Runnable firstTask, boolean core) {
    2.         //增加Worker的数量
    3.         retry:
    4.         for (;;) {
    5.             int c = ctl.get();
    6.             int rs = runStateOf(c);
    7. ​
    8.             // Check if queue empty only if necessary.
    9.             if (rs >= SHUTDOWN &&
    10.                 ! (rs == SHUTDOWN &&
    11.                    firstTask == null &&
    12.                    ! workQueue.isEmpty()))
    13.                 return false;
    14. ​
    15.             for (;;) {
    16.                 int wc = workerCountOf(c);
    17.                 if (wc >= CAPACITY ||
    18.                     wc >= (core ? corePoolSize : maximumPoolSize))
    19.                     return false;
    20.                 if (compareAndIncrementWorkerCount(c))
    21.                     break retry;
    22.                 c = ctl.get();  // Re-read ctl
    23.                 if (runStateOf(c) != rs)
    24.                     continue retry;
    25.             }
    26.         }
    27. ​
    28.         //创建Worker并且启动
    29.         boolean workerStarted = false;
    30.         boolean workerAdded = false;
    31.         Worker w = null;
    32.         try {
    33.             w = new Worker(firstTask);
    34.             final Thread t = w.thread;
    35.             if (t != null) {
    36.                 final ReentrantLock mainLock = this.mainLock;
    37.                 mainLock.lock();
    38.                 try {
    39.                     // Recheck while holding lock.
    40.                     // Back out on ThreadFactory failure or if
    41.                     // shut down before lock acquired.
    42.                     int rs = runStateOf(ctl.get());
    43. ​
    44.                     if (rs < SHUTDOWN ||
    45.                         (rs == SHUTDOWN && firstTask == null)) {
    46.                         if (t.isAlive()) // precheck that t is startable
    47.                             throw new IllegalThreadStateException();
    48.                         workers.add(w);
    49.                         int s = workers.size();
    50.                         if (s > largestPoolSize)
    51.                             largestPoolSize = s;
    52.                         workerAdded = true;
    53.                     }
    54.                 } finally {
    55.                     mainLock.unlock();
    56.                 }
    57.                 if (workerAdded) {
    58.                     t.start();
    59.                     workerStarted = true;
    60.                 }
    61.             }
    62.         } finally {
    63.             if (! workerStarted)
    64.                 addWorkerFailed(w);
    65.         }
    66.         return workerStarted;
    67.     }

    上面的代码也是做了两件事情：

    1. 增加worker数量的统计
    2. 创建新的Worker并且启动

    在线程池中的任务处理主要是靠一个 Worker 的内部类进行处理。下面来看一下这个内部类：

    1. private final class Worker
    2.         extends AbstractQueuedSynchronizer
    3.         implements Runnable
    4.     {
    5.         /**
    6.          * This class will never be serialized, but we provide a
    7.          * serialVersionUID to suppress a javac warning.
    8.          */
    9.         private static final long serialVersionUID = 6138294804551838833L;
    10. ​
    11.         /** Thread this worker is running in.  Null if factory fails. */
    12.         final Thread thread;
    13.         /** Initial task to run.  Possibly null. */
    14.         Runnable firstTask;
    15.         /** Per-thread task counter */
    16.         volatile long completedTasks;
    17. ​
    18.         /**
    19.          * Creates with given first task and thread from ThreadFactory.
    20.          * @param firstTask the first task (null if none)
    21.          */
    22.         Worker(Runnable firstTask) {
    23.             setState(-1); // inhibit interrupts until runWorker
    24.             this.firstTask = firstTask;
    25.             this.thread = getThreadFactory().newThread(this);
    26.         }
    27. ​
    28.         /** Delegates main run loop to outer runWorker  */
    29.         public void run() {
    30.             runWorker(this);
    31.         }
    32.         protected boolean isHeldExclusively() {
    33.             return getState() != 0;
    34.         }
    35. ​
    36.         protected boolean tryAcquire(int unused) {
    37.             if (compareAndSetState(0, 1)) {
    38.                 setExclusiveOwnerThread(Thread.currentThread());
    39.                 return true;
    40.             }
    41.             return false;
    42.         }
    43. ​
    44.         protected boolean tryRelease(int unused) {
    45.             setExclusiveOwnerThread(null);
    46.             setState(0);
    47.             return true;
    48.         }
    49. ​
    50.         public void lock()        { acquire(1); }
    51.         public boolean tryLock()  { return tryAcquire(1); }
    52.         public void unlock()      { release(1); }
    53.         public boolean isLocked() { return isHeldExclusively(); }
    54. ​
    55.         void interruptIfStarted() {
    56.             Thread t;
    57.             if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {
    58.                 try {
    59.                     t.interrupt();
    60.                 } catch (SecurityException ignore) {
    61.                 }
    62.             }
    63.         }
    64.     }

    Worker 继承了 AbstractQueuedSynchronizer 实现了 Runnable 。Worker 中有两个变量：

    • Thread变量
    • Runnable变量

    第一个是在创建Worker的时候，把Worker变成线程保存起来，也就是通过这样的方式来处理任务，Runnable保存的是创建Worker的时候执行的任务。那么这个Worker的run方法什么时候执行。在前面执行 addWorker 方法的时候，会有一个创建Worker的过程，然后调用了Thread.start()方法。这样就会执行到Worker的run方法，而在run方法中调用的是 ThreadPoolExecutor.runWorker 参数是当前Worker的实例:

    1. final void runWorker(Worker w) {
    2.         Thread wt = Thread.currentThread();
    3.         Runnable task = w.firstTask;
    4.         w.firstTask = null;
    5.         w.unlock(); // allow interrupts
    6.         boolean completedAbruptly = true;
    7.         try {
    8.             while (task != null || (task = getTask()) != null) {
    9.                 w.lock();
    10.                 if ((runStateAtLeast(ctl.get(), STOP) ||
    11.                      (Thread.interrupted() &&
    12.                       runStateAtLeast(ctl.get(), STOP))) &&
    13.                     !wt.isInterrupted())
    14.                     wt.interrupt();
    15.                 try {
    16.                     beforeExecute(wt, task);
    17.                     Throwable thrown = null;
    18.                     try {
    19.                         task.run();
    20.                     } catch (RuntimeException x) {
    21.                         thrown = x; throw x;
    22.                     } catch (Error x) {
    23.                         thrown = x; throw x;
    24.                     } catch (Throwable x) {
    25.                         thrown = x; throw new Error(x);
    26.                     } finally {
    27.                         afterExecute(task, thrown);
    28.                     }
    29.                 } finally {
    30.                     task = null;
    31.                     w.completedTasks++;
    32.                     w.unlock();
    33.                 }
    34.             }
    35.             completedAbruptly = false;
    36.         } finally {
    37.             processWorkerExit(w, completedAbruptly);
    38.         }
    39.     }

    首先获取Worker中的需要处理的任务去处理，当处理完成Worker中的通过获取getTask任务列表中的任务进行处理。根据是否有核心处理线程(Worker)来是否要退出当前Worker:

    1. private Runnable getTask() {
    2.         boolean timedOut = false; // Did the last poll() time out?
    3. ​
    4.         for (;;) {
    5.             int c = ctl.get();
    6.             int rs = runStateOf(c);
    7. ​
    8.             // Check if queue empty only if necessary.
    9.             if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
    10.                 decrementWorkerCount();
    11.                 return null;
    12.             }
    13. ​
    14.             int wc = workerCountOf(c);
    15. ​
    16.             // 默认情况下core线程不会失效
    17.             boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
    18. ​
    19.             if ((wc > maximumPoolSize || (timed && timedOut))
    20.                 && (wc > 1 || workQueue.isEmpty())) {
    21.                 if (compareAndDecrementWorkerCount(c))
    22.                     return null;
    23.                 continue;
    24.             }
    25. ​
    26.             try {
    27.                 //根据是否失效调用任务列表的不同方法
    28.                 Runnable r = timed ?
    29.                     //调用poll，在规定时间内还没有就返回null
    30.                     workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
    31.                     //没有就阻塞当前线程
    32.                     workQueue.take();
    33.                 if (r != null)
    34.                     return r;
    35.                 timedOut = true;
    36.             } catch (InterruptedException retry) {
    37.                 timedOut = false;
    38.             }
    39.         }
    40.     }

    5. 线程池化的模型图

    • 从池的空闲线程列表中选择一个 Thread，并且指派它去运行一个已提交的任务(一个 Runnable，Callable 的实现)

    • 当任务完成时，将该 Thread 返回给该列表，使其可被重用。

    6. 线程池拒绝策略

    • CallerRunsPolicy：在任务被拒绝添加后，会调用当前线程池的所在的线程去执行被拒绝的任务

    1. public static class CallerRunsPolicy implements RejectedExecutionHandler {
    2.        
    3.         public CallerRunsPolicy() { }
    4. ​
    5.       
    6.         public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
    7.             if (!e.isShutdown()) {
    8.                 r.run();
    9.             }
    10.         }
    11.     }

    AbortPolicy：直接抛出异常

    1. public static class AbortPolicy implements RejectedExecutionHandler {
    2.         
    3.         public AbortPolicy() { }
    4. ​
    5.         public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
    6.             throw new RejectedExecutionException("Task " + r.toString() +
    7.                                                  " rejected from " +
    8.                                                  e.toString());
    9.         }
    10.     }

    DiscardPolicy：会让被线程池拒绝的任务直接抛弃，不会抛异常也不会执行。

    1. public static class DiscardPolicy implements RejectedExecutionHandler {
    2. ​
    3.         public DiscardPolicy() { }
    4. ​
    5.         public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
    6.         }
    7.     }

    DiscardOldestPolicy：DiscardOldestPolicy策略的作用是，当任务呗拒绝添加时，会抛弃任务队列中最旧的任务也就是最先加入队列的，再把这个新任务添加进去。

    1. public static class DiscardOldestPolicy implements RejectedExecutionHandler {
    2. ​
    3.         public DiscardOldestPolicy() { }
    4. ​
    5.         public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
    6.             if (!e.isShutdown()) {
    7.                 e.getQueue().poll();
    8.                 e.execute(r);
    9.             }
    10.         }
    11.     }

    自定义策略，只要实现RejectedExecutionHandler接口

     

  • 相关阅读:
    LeetCode(力扣)78. 子集Python
    研发过程中的文档管理与工具
    String解析及其方法
    JS选择框全选效果（从入门到优化）
    参考文献批量添加上标
    Logback 日志格式参数说明
    CISP考试有哪些备考技巧
    使用git发布（删除/更改）GitHub仓库中的内容
    前缀和和差分和dijkstra算法和二分算法和floyd算法
    Google Earth Engine（GEE）——利用行列号进行影像去云，使用研究区大于单景影像的研究区范围
  • 原文地址：https://blog.csdn.net/m0_67698950/article/details/125426244