JAVA-ThreadPoolExecutor笔记

时间 2019-11-08

标签 java threadpoolexecutor 笔记栏目 Java 繁體版

原文原文链接

常量分析

// 线程池控制器，高3位表明状态，低29位表示线程池的大小
// 例如：11100000000000000000000000000001，高3位111表明线程池处于RUNNING状态，低29位00000000000000000000000000001表明当前有一个工做线程.
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));

// COUNT_BITS为29
private static final int COUNT_BITS = Integer.SIZE - 3;

// 线程池最大容量为2^30 - 1，也就是ctl的低29位所有为1.
// 二进制值:000111111111111111111111111111111
private static final int CAPACITY   = (1 << COUNT_BITS) - 1;

// runState is stored in the high-order bits
// RUNNING:11100000000000000000000000000000
private static final int RUNNING    = -1 << COUNT_BITS;

// SHUTDOWN:00000000000000000000000000000000
private static final int SHUTDOWN   =  0 << COUNT_BITS;

// STOP:00100000000000000000000000000000
private static final int STOP       =  1 << COUNT_BITS;

// TIDYING:01000000000000000000000000000000
private static final int TIDYING    =  2 << COUNT_BITS;

// TERMINATED:01100000000000000000000000000000
private static final int TERMINATED =  3 << COUNT_BITS;

// Packing and unpacking ctl
// 计算线程池当前的状态.
// 例如:当前ctl(也就是传入的参数c)为01100000000000000000000000000111, ~CAPACITY为CAPACITY取反.
// 那么c & ~CAPACITY也就是01100000000000000000000000000111 & 11100000000000000000000000000000，结果为:01100000000000000000000000000000
private static int runStateOf(int c)     { return c & ~CAPACITY; }

// 与runStateOf方法相似，不过这是求当前线程池中的线程数量
private static int workerCountOf(int c)  { return c & CAPACITY; }

// 求控制器，rs表明线程池的runState，wc表明线程池当前的线程数量，作或操做其实就是求出当前线程池的控制器的值.
private static int ctlOf(int rs, int wc) { return rs | wc; }
复制代码

主流程解析

int corePoolSize:核心线程数.
int maximumPoolSize:最大线程数.
long keepAliveTime:除核心线程以外的其余线程保持存活的时间.
TimeUnit unit:时间单位.
BlockingQueue workQueue:任务队列.当前核心线程没法知足任务需求时，先将任务放入队列中.
ThreadFactory threadFactory:线程工厂.
RejectedExecutionHandler handler:拒绝策略处理器.

public void execute(Runnable command) {
    // 任务为null，直接抛出异常
    if (command == null)
        throw new NullPointerException();
    
    // 获取控制器    
    int c = ctl.get();
    // 计算工做线程数是否小于corePoolSize
    if (workerCountOf(c) < corePoolSize) {
        // 添加工做线程
        if (addWorker(command, true))
            // 添加成功，返回
            return;
            
        // 从新获取控制器,可能有其余线程对其进行了修改.
        c = ctl.get();
    }
    
    // 线程池是RUNNING状态，而且任务放入队列成功.
    if (isRunning(c) && workQueue.offer(command)) {
        // 再次获取控制器
        int recheck = ctl.get();
        // 线程状态不是RUNNING，从队列中删除任务，并调用RejectedExecutionHandler进行处理.
        if (! isRunning(recheck) && remove(command))
            reject(command);
        else if (workerCountOf(recheck) == 0) // 判断工做线程数是否为0，为0则添加工做线程.
            addWorker(null, false);
    }
    // 线程池的任务数大于corePoolSize，而且队列也已经放满，则添加工做线程来对任务进行处理，若是添加失败，执行reject策略.
    else if (!addWorker(command, false))
        reject(command);
}
复制代码

添加工做线程流程解析

private boolean addWorker(Runnable firstTask, boolean core) {
    retry:
    for (;;) {
        int c = ctl.get();
        int rs = runStateOf(c);
        // Check if queue empty only if necessary.
        // 状态判断
        // rs >= SHUTDOWN表示状态为SHUTDOWN、STOP、TIDYING或者TERMINATED
        // 总体意思为，若是当前线程池的状态为非RUNNING状态，而且状态不为SHUTDOWN或者入参任务不为null或者任务队列为空的状况下，再也不建立工做线程.
        if (rs >= SHUTDOWN &&
            ! (rs == SHUTDOWN &&
               firstTask == null &&
               ! workQueue.isEmpty()))
            return false;
            
        // 该循环的意思相似于CAS操做，经过循环不停的去对线程池的工做线程数量+1.
        for (;;) {
            // 获取工做线程数量.
            int wc = workerCountOf(c);
            // 若是工做线程的数量大于CAPACITY(2^30 - 1)或者大于了设定的参数值(当core为true时是corePoolSize，不然为maximumPoolSize)，直接返回失败
            // 返回失败以后的处理有两种状况:若是core为true的话，会把任务放入任务队列中;若是core为false的话，会执行reject策略.
            if (wc >= CAPACITY ||
                wc >= (core ? corePoolSize : maximumPoolSize))
                return false;
            // 利用原子类的CAS操做对工做线程数量+1
            if (compareAndIncrementWorkerCount(c))
                // +1成功，退出retry循环.
                break retry;
            c = ctl.get();  // Re-read ctl
            // 状态发生了变化，则从新进行retry循环，从新进行状态检查，不然只需在当前循环中继续cas操做.
            if (runStateOf(c) != rs)
                continue retry;
        }
    }
    boolean workerStarted = false;
    boolean workerAdded = false;
    Worker w = null;
    try {
        w = new Worker(firstTask);
        final Thread t = w.thread;
        if (t != null) {
            // 获取锁
            final ReentrantLock mainLock = this.mainLock;
            mainLock.lock();
            try {
                int rs = runStateOf(ctl.get());
                // 若是状态为RUNNING或者状态为SHUTDOWN而且任务为null.
                if (rs < SHUTDOWN ||
                    (rs == SHUTDOWN && firstTask == null)) {
                    // 判断线程是否已启动.
                    if (t.isAlive()) // precheck that t is startable
                        throw new IllegalThreadStateException();
                    // 加入工做线程队列
                    workers.add(w);
                    
                    // 更新线程池的大小(largestPoolSize:线程池的工做线程的最大数量)，加锁主要是为了更新该字段.
                    int s = workers.size();
                    if (s > largestPoolSize)
                        largestPoolSize = s;
                    workerAdded = true;
                }
            } finally {
                mainLock.unlock();
            }
            
            // 若是添加到工做线程队列成功，则启动线程，将workerStarted参数设置为true.
            if (workerAdded) {
                t.start();
                workerStarted = true;
            }
        }
    } finally {
        if (! workerStarted)
            // 添加工做线程失败的处理逻辑.
            addWorkerFailed(w);
    }
    return workerStarted;
}
复制代码

addWorkerFailed(w)方法解析程序员

private void addWorkerFailed(Worker w) {
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        if (w != null)
            // worker不为null，从工做线程队列中移除.
            workers.remove(w);
        // worker数量-1.
        decrementWorkerCount();
        // 检查terminate.
        tryTerminate();
    } finally {
        mainLock.unlock();
    }
}


final void tryTerminate() {
    for (;;) {
        int c = ctl.get();
        
        // 若是状态时RUNNING，或者状态是TIDYING或TERMINATED，或者状态是SHUTDOWN而且任务队列不为空的状况，直接返回.
        // 状态时RUNNING的状况，不能执行terminate操做
        // 状态时TIDYING或者TERMINATED的状况，不须要再执行一次terminate操做.
        // 当是SHUTDOWN状态可是任务队列不为空的话，说明还有任务须要执行，也没法执行terminate操做.
        if (isRunning(c) ||
            runStateAtLeast(c, TIDYING) ||
            (runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty()))
            return;
            
        // 若是worker的数量不为0，则将其中一个worker的中断标志设置为true，并返回.
        if (workerCountOf(c) != 0) { // Eligible to terminate
            interruptIdleWorkers(ONLY_ONE);
            return;
        }
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            // 将状态设置为TIDYING
            if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {
                try {
                    // hook方法.空实现
                    terminated();
                } finally {
                    // 将状态设置为TERMINATED
                    ctl.set(ctlOf(TERMINATED, 0));
                    // 唤醒全部在maintain lock上等待的线程.
                    // 好比，若是主线程使用了executor的awaitTermination方法，那么就会在该语句以后被唤醒.
                    termination.signalAll();
                }
                return;
            }
        } finally {
            mainLock.unlock();
        }
        // else retry on failed CAS
    }
}


private void interruptIdleWorkers(boolean onlyOne) {
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        for (Worker w : workers) {
            Thread t = w.thread;
            // 线程处于非interrupted状态，而且获取到了worker对象的锁
            // 这里获取worker对象锁的主要缘由是防止线程正在执行任务而被中断.
            if (!t.isInterrupted() && w.tryLock()) {
                try {
                    t.interrupt();
                } catch (SecurityException ignore) {
                } finally {
                    w.unlock();
                }
            }
            if (onlyOne)
                break;
        }
    } finally {
        mainLock.unlock();
    }
}
复制代码

Worker解析

Worker继承了AQS，而且实现了Runnable接口.bash

Worker(Runnable firstTask) {
  // 这里state是AQS中一个volatile变量，Worker中的tryLock和unlock方法都是基于该变量实现的.
  // 将state的值设置为-1，禁止在执行runWorker以前被中断.
  setState(-1);
  this.firstTask = firstTask;
  this.thread = getThreadFactory().newThread(this);
}

public void run() {
    // 重点方法.
    runWorker(this);
}


final void runWorker(Worker w) {
    Thread wt = Thread.currentThread();
    Runnable task = w.firstTask;
    w.firstTask = null;
    // 释放锁(将state的值设置为0)，使调用interruptIdleWorkers()方法的线程能够获取锁来设置线程的中断标志.
    w.unlock(); // allow interrupts
    
    // 是否忽然完成，好比用户代码发生异常，致使直接走到finally代码块中.
    boolean completedAbruptly = true;
    try {
        // task不为null，或者从任务队列中获取task不为null.
        // 若设置了中断标志为true，那么在getTask方法上阻塞的线程能够直接抛出InterruptedException，从而结束线程.
        while (task != null || (task = getTask()) != null) {
            // 加锁，防止在执行任务的过程当中线程被中断.
            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
            // 上面的英文意思很清楚，这里想说的是为何代码要这么写
            // 查看Thread的源码能够看到Thread.interrupted()方法会清除掉当前线程的中断标志.
            // 所以当||操做符前面的条件为false时(也就是状态不为STOP、TIDYING、TERMINATED)时，就会调用Thread.interrupted()方法，将线程的中断标志清除掉.
            if ((runStateAtLeast(ctl.get(), STOP) ||
                 (Thread.interrupted() &&
                  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 {
                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.
        // 状态检查
        if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
            decrementWorkerCount();
            return null;
        }
        int wc = workerCountOf(c);
        // Are workers subject to culling?
        // 判断是否须要超时
        // wc > corePoolSize表示工做线程数大于corePoolSize，在获取任务的时候要加超时操做.
        boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
        
        // 就是检查上一次操做是否超时以及任务队列是否为空.
        if ((wc > maximumPoolSize || (timed && timedOut))
            && (wc > 1 || workQueue.isEmpty())) {
            // worker线程数-1
            if (compareAndDecrementWorkerCount(c))
                // -1成功，返回null，使线程退出while循环，正常结束.
                return null;
            // 继续for循环.
            continue;
        }
        
        try {
            // 从任务队列中拉取任务，若是timed为false，则线程会一直阻塞，直到任务队列中有值为止;若是timed为true，超时返回null，将timedOut设置为true，而后在上面的if判断中进行判断.
            Runnable r = timed ?
                workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                workQueue.take();
            if (r != null)
                return r;
            timedOut = true;
        } catch (InterruptedException retry) {
            timedOut = false;
        }
    }
}


private void processWorkerExit(Worker w, boolean completedAbruptly) {
    // completedAbruptly若是是true，也就是忽然完成，业务代码抛出了异常，worker的数量-1.
    // 若是是worker是正常退出的话，只有多是设置了allowCoreThreadTimeOut，该字段的意思是容许核心worker线程超时.
    // 意思是就算工做线程数小于或等于corePoolSize，当某个线程未在keepAliveTime内获取到任务时，也将退出循环，结束线程.
    if (completedAbruptly)
        decrementWorkerCount();
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        // 完成的任务数+1
        completedTaskCount += w.completedTasks;
        // 从队列中移除worker
        workers.remove(w);
    } finally {
        mainLock.unlock();
    }
    
    // 参考上面该方法的解析
    tryTerminate();
    int c = ctl.get();

    if (runStateLessThan(c, STOP)) {
        // 线程正常退出.
        if (!completedAbruptly) {
            // 若是容许核心线程数超时，则min为0，不然为corePoolSize
            int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
            // 若是min为0，而且任务队列不为空，则最小线程数为1
            if (min == 0 && ! workQueue.isEmpty())
                min = 1;
            
            // 若是工做线程的数量大于等于min，则退出方法.
            if (workerCountOf(c) >= min)
                return; // replacement not needed
        }
        // 添加工做线程，参考上面addWorker方法解析.
        addWorker(null, false);
    }
}
复制代码

拒绝策略

在ThreadPoolExecutor中实现了4种默认的拒绝策略ui

CallerRunsPolicy:在executor没有被shutdown的状况下，直接执行任务.

public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
    if (!e.isShutdown()) {
        r.run();
    }
}
复制代码

AbortPolicy:这个就厉害了，直接抛出RejectedExecutionException异常!

public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
    throw new RejectedExecutionException("Task " + r.toString() +
                                            " rejected from " +
                                            e.toString());
}
复制代码

DiscardPolicy:直译过来就是，悄咪咪的将任务丢弃掉，嘘！不能让程序员知道!

public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
}
复制代码

DiscardOldestPolicy:将任务队列中最老的任务丢弃掉(这么老了就别占着茅坑不拉屎啦，对于FIFO队列是丢弃掉下一个将被执行的任务，若是是优先级队列那么这种策略将会丢弃优先级最高的任务，所以最好不要将该策略与优先级队列一块儿使用)，而后将新任务放入execute方法中执行(能不能执行或者占个坑就看运气了)

public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
  if (!e.isShutdown()) {
    e.getQueue().poll();
    e.execute(r);
  }
}
复制代码

shutDown和shutDownNow

shutdown和shutDownNow的区别主要是，shutdown会等线程将当前任务执行完成才进行interrupt操做，而shutDownNow是无论线程是否正在执行任务都进行interrupt操做.shutdown是将线程池状态修改成SHUTDOWN，而shutDownNow是将线程池的状态修改成STOP.this

shutdown:

public void shutdown() {
  final ReentrantLock mainLock = this.mainLock;
  mainLock.lock();
  try {
      checkShutdownAccess();
      // 将状态设置为SHUTDOWN
      advanceRunState(SHUTDOWN);
      // 这个方法是和shutDownNow的区别.interruptIdleWorkers方法上面已经分析过
      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);
      // 和shutdown调用的方法也不同.
      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();
  }
}

// Worker对象的方法
void interruptIfStarted() {
  Thread t;
  // state的状态只有在Worker对象未调用runWorker以前才会是负数，所以只要worker调用了runWorker方法，不论是加锁仍是未加锁，getState方法的返回值都是大于等于0的.
  if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {
      try {
          t.interrupt();
      } catch (SecurityException ignore) {
      }
  }
}
复制代码