JDK8 JUC阅读(5)

ThreadPoolExecutor

Posted by Jqy on January 11, 2020

ThreadPoolExecutor

内部的原子成员变量ctl记录了线程池的状态

//高3位表示线程池状态,其他位表示线程个数
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
private static final int COUNT_BITS = Integer.SIZE - 3;
private static final int CAPACITY   = (1 << COUNT_BITS) - 1;

// runState is stored in the high-order bits
private static final int RUNNING    = -1 << COUNT_BITS; //接受新任务并处理阻塞队列
private static final int SHUTDOWN   =  0 << COUNT_BITS; //拒绝新任务,处理阻塞队列
private static final int STOP       =  1 << COUNT_BITS; //拒绝新任务和阻塞队列,中断正在处理的任务
private static final int TIDYING    =  2 << COUNT_BITS; //将要调用terminated
private static final int TERMINATED =  3 << COUNT_BITS; //终止状态

// Packing and unpacking ctl
private static int runStateOf(int c)     { return c & ~CAPACITY; }
private static int workerCountOf(int c)  { return c & CAPACITY; }
//计算ctl的值
private static int ctlOf(int rs, int wc) { return rs | wc; }

线程池类型

  • newFixedThreadPool:核心线程,最大线程个数都为n
  • newSingleThreadExecutor:核心线程和最大线程为1
  • newCachedThreadPool:初始线程个数为0,最多为MAX_VALUE,加入同步队列的任务会马上执行,但队列中最多只有一个任务。 ThreadPoolExecutor的一些字段解释:
    1. mainLock:独占锁,控制Worker的增减
    2. termination:mainLock对应的条件队列,在线程调用awaitTermination时存放阻塞的线程
    3. Worker继承AQS和Runnable,为具体承载任务的对象
    4. DefaultThreadFactory:线程工厂

      1.execute

      该方法用于提交任务command到线程池中执行

      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();
             if (! isRunning(recheck) && remove(command))
                 reject(command);
             else if (workerCountOf(recheck) == 0)
                 addWorker(null, false);
         }
         else if (!addWorker(command, false))
             reject(command);
        }