透线程池

原文地址：https://mp.weixin.qq.com/s/n2cZxqHqaMJM2mh9Tk13Bg

目录

  • TreadPoolexecutor源码解析
  • 类关系图
  • Executor接口
  • ExecutorService接口
  • AbstractExecutorService
  • 成员变量
  • 构造函数
  • Worker类解析
  • Worker简介
  • 成员变量
  • 构造方法
  • execute()解析
  • addWorker()解析
  • runWorker()解析
  • getTask()解析
  • processWorkerExit()解析
  • tryTerminate()解析
  • interruptIdleWorker()解析
  • 监控线程池

TreadPoolexecutor源码解析

1.类关系图

image

Executor接口

Executor提供execute()用来启动任务

 public interface Executor {
      //用来启动任务
      void execute(Runnable command);
   }

2.ExecutorService接口

ExecutorService提供了一些管理线程池方法和任务执行的方法

 public interface ExecutorService extends Executor {
        //关闭线程池，队列已经存在的任务可以继续执行
        void shutdown();
        //关闭线程池，中断未执行的任务
        List<Runnable> shutdownNow();
        //判断是否关闭
        boolean isShutdown();
        //判断是否终止
        boolean isTerminated();
        //设置超时终止
        boolean awaitTermination(long timeout, TimeUnit unit)
            throws InterruptedException;
        //提交Callable任务
        <T> Future<T> submit(Callable<T> task);
        //提交Runable任务，带返回值
        <T> Future<T> submit(Runnable task, T result);
        //提交Runnable任务不带返回值
        Future<?> submit(Runnable task);
        //invokeAll()是同步的，其需要等待任务的完成，才能返回。submit()是异步的
        <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
            throws InterruptedException;

        <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,
                                      long timeout, TimeUnit unit)
            throws InterruptedException;
        //invokeAny()取第一个任务的返回值，然后调用interrupt方法中断其它任务。
        <T> T invokeAny(Collection<? extends Callable<T>> tasks)
            throws InterruptedException, ExecutionException;

        <T> T invokeAny(Collection<? extends Callable<T>> tasks,
                        long timeout, TimeUnit unit)
            throws InterruptedException, ExecutionException, TimeoutException;
    }

AbstractExecutorService

AbstractExecutorService类重写某些方法。自定义了newTaskFor()用于构建RunnableFuture。

成员变量

/**
  * 线程池使用一个int变量存储线程池状态和工作线程数
  * int4个字节，32位，用高三位存储线程池状态，低29位存储工作线程数
  * 为什么使用一个变量来同时表示线程状态和线程数？就是节省空间。咨询了一下写c的朋友，他们经常这么写
  **/
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
//COUNT_BITS=29
private static final int COUNT_BITS = Integer.SIZE - 3;
//理论上线程池最大线程数量CAPACITY=536870911
private static final int CAPACITY   = (1 << COUNT_BITS) - 1;
/**
  * 线程池状态转换
  * RUNNING -> SHUTDOWN
  * RUNNING or SHUTDOWN -> STOP
  * SHUTDOWN or STOP -> TIDYING
  * TIDYING -> TERMINATED  terminated()执行完后变为该TERMINATED
  */
//接受新任务，可以处理阻塞队列里的任务
private static final int RUNNING    = -1 << COUNT_BITS;
//不接受新任务，可以处理阻塞队列里的任务。执行shutdown()会变为SHUTDOWN
private static final int SHUTDOWN   =  0 << COUNT_BITS;
//不接受新的任务，不处理阻塞队列里的任务，中断正在处理的任务。执行shutdownNow()会变为STOP
private static final int STOP       =  1 << COUNT_BITS;
//临时过渡状态，所有的任务都执行完了，当前线程池有效的线程数量为0，这个时候线程池的状态是TIDYING，执行terminated()变为TERMINATED
private static final int TIDYING    =  2 << COUNT_BITS;
//终止状态，terminated()调用完成后的状态
private static final int TERMINATED =  3 << COUNT_BITS;

//获取线程池状态
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; }
//用于保存等待执行的任务的阻塞队列。比如LinkedBlockQueue，SynchronousQueue等
private final BlockingQueue<Runnable> workQueue;
//重入锁，更新线程池核心大小、线程池最大大小等都有用到
private final ReentrantLock mainLock = new ReentrantLock();
//用于存储woker
private final HashSet<Worker> workers = new HashSet<Worker>();
//用于终止线程池
private final Condition termination = mainLock.newCondition();
//记录线程池中曾经出现过的最大线程数
private int largestPoolSize;
//完成任务数量
private long completedTaskCount;   
//线程工厂
private volatile ThreadFactory threadFactory;
/**
 *  rejectedExecutionHandler：任务拒绝策略
 *  DiscardOldestPolicy：丢弃队列里最近的一个任务，并执行当前任务
 *  AbortPolicy：抛出异常。这也是默认的策略
 *  CallerRunsPolicy：用调用者所在线程来运行任务
 *  DiscardPolicy：不处理，丢弃掉
 */
private volatile RejectedExecutionHandler handler;
/**
  * 线程空闲时间
  * 当线程空闲时间达到keepAliveTime，该线程会退出，直到线程数量等于corePoolSize。
  * 如果allowCoreThreadTimeout设置为true，则所有线程均会退出。
  */
private volatile long keepAliveTime;
//是否允许核心线程空闲超时退出，默认值为false。
private volatile boolean allowCoreThreadTimeOut;
/**
 * 核心线程数
 * 核心线程会一直存活，即使没有任务需要处理，当线程数小于核心线程数时。
 * 即使现有的线程空闲，线程池也会优先创建新线程来处理任务，而不是直接交给现有的线程处理。
 * 核心线程数在初始化时不会创建，只有提交任务的时候才会创建。核心线程在allowCoreThreadTimeout为true的时候超时会退出。
 */
private volatile int corePoolSize;
 /** 最大线程数
   * 当线程数大于或者等于核心线程，且任务队列已满时，线程池会创建新的线程，直到线程数量达到maxPoolSize。
   * 如果线程数已等于maxPoolSize，且任务队列已满，则已超出线程池的处理能力，线程池会采取拒绝操作。
   */
private volatile int maximumPoolSize;
//默认的拒绝策略：抛出异常
private static final RejectedExecutionHandler defaultHandler =
    new AbortPolicy();
private static final RuntimePermission shutdownPerm =
    new RuntimePermission("modifyThread");

构造函数

//直接提供了一个最终调用的构造函数
//大致逻辑就是给线程池核心参数赋值
public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue<Runnable> workQueue,
                          ThreadFactory threadFactory,
                          RejectedExecutionHandler handler) {
    if (corePoolSize < 0 ||
        maximumPoolSize <= 0 ||
        maximumPoolSize < corePoolSize ||
        keepAliveTime < 0)
        throw new IllegalArgumentException();
    if (workQueue == null || threadFactory == null || handler == null)
        throw new NullPointerException();
    this.corePoolSize = corePoolSize;
    this.maximumPoolSize = maximumPoolSize;
    this.workQueue = workQueue;
    //转为纳秒
    this.keepAliveTime = unit.toNanos(keepAliveTime);
    this.threadFactory = threadFactory;
    this.handler = handler;
}

Worker类解析

Worker简介

Woeker类是ThreadPoolExecutor一个内部类。此类继AbstractQueuedSynchronizer，
目的在于判断线程是否空闲以及是否可以被中断。实现Runnable，在run()中调用了runWorker()。
Worker类中firstTask用来保存传入的任务，thread是在调用构造方法时通过ThreadFactory来创建的线程，是用来处理任务的线程。注意：这个线程并不是task线程。

成员变量

  /** 工作线程空间，由线程池中所设置的线程工厂创建*/
    final Thread thread;
    /** 在构造方法中传入的任务*/
    Runnable firstTask;
    /** 执行完任务的总数*/
    volatile long completedTasks;

构造方法

 Worker(Runnable firstTask) {
        //state设置为-1是为了禁止在执行任务前对任务进行中断。
        setState(-1); 
        //提交的task
        this.firstTask = firstTask;
        //从线程工厂获取的线程，注意这个thread并不是用户线程
        this.thread = getThreadFactory().newThread(this);
    }

execute()解析

public void execute(Runnable command) {
     //判断提交的任务是否为空
    if (command == null)
        throw new NullPointerException();
    //获取线程池状态和工作线程数量结合体(下文统称为ctl)
    int c = ctl.get();
    //判断工作线程数量是否小于核心线程数
    if (workerCountOf(c) < corePoolSize) {
        //把任务添加到worker，添加成功则返回
        if (addWorker(command, true))
            return;
        //再次获取ctl
        c = ctl.get();
    }
    //如果线程池状态是正在运行并且添加一个任务到队列成功
    if (isRunning(c) && workQueue.offer(command)) {
        int recheck = ctl.get();
        //再次校验线程池状态，如果状态不是RUNNING则需要从队列中移除任务
        if (! isRunning(recheck) && remove(command))
             //执行拒绝策略
            reject(command);
        else if (workerCountOf(recheck) == 0)
            //进入这里说明空闲核心线程数都超时退出啦
            //因为任务已经放入队列了，所以此处不需要传入任务
            //注意事项，网上很多说这里创建一个线程不启动这是错误的。博主亲测这里创建了一个线程并且需要启动
            addWorker(null, false);
    }
   /*
    * 如果执行到这里有两种情况：
    * 线程池已经不是RUNNING状态
    * 线程池是RUNNING状态，workerCount>=corePoolSize并且workQueue已满
    */
    //调用addWorker(),传入false代表把线程池线程数量设置maximumPoolSize，如果添加失败则执行拒绝策略。
    else if (!addWorker(command, false))
        reject(command);
}

addWorker()解析

/**
  *@param firstTask 表示执行的任务；
  *@param core 表示限制添加线程的数量是根据corePoolSize来判断还是maximumPoolSize来判断； 
  * 如果为true，根据corePoolSize来判断；如果为false，则根据maximumPoolSize来判断
  */
private boolean addWorker(Runnable firstTask, boolean core) {
    retry:
    for (;;) {
        int c = ctl.get();
        //获取线程池状态
        int rs = runStateOf(c);
        /**
          * 只要满足下面任一条就直接返回false
          * 线程池状态为STOP, TYDING 或 TERMINATD 状态
          * 线程池状态为SHUTDOWN，并且firstTask != null 或者workQueue为空
          */
         //这里为什么不直接判断线程池状态呢？是因为有可能在线程池状态为RUNNING时已经把任务放入队列中，放入完成以后状态变为SHUTDOWN
        if (rs >= SHUTDOWN &&! (rs == SHUTDOWN &&
               firstTask == null &&! workQueue.isEmpty()))
            return false;
        for (;;) {
            //得到工作的线程数量
            int wc = workerCountOf(c);
            //如果工作线程数量大于理论上线程池容量；或者工作线程数大于(corePoolSize or maximumPoolSize) 直接返回false，添加失败
            if (wc >= CAPACITY || wc >= (core ? corePoolSize : maximumPoolSize))
                return false;
             //cas修改工作线程数，工作线程数+1。如果修改失败需要重新执行；成功退出循环
            if (compareAndIncrementWorkerCount(c))
                break retry;
            c = ctl.get();  
             //如果线程池状态变化则需要重新执行
            if (runStateOf(c) != rs)
                continue retry;
        }
    }
    //worker是否已经启动
    boolean workerStarted = false;
    //worker是否添加成功
    boolean workerAdded = false;
    Worker w = null;
    try {
        //构建worker
        w = new Worker(firstTask);
        //注意，这个thread不是firstTask，是从线程工厂造出来的
        final Thread t = w.thread;
        if (t != null) {
            //获取锁
            final ReentrantLock mainLock = this.mainLock;
            mainLock.lock();
            try {
                //获取线程池状态
                int rs = runStateOf(ctl.get());
                //如果线程池状态是RUNNING或者状态是SHUTDOWN但是队列里面还有任务
                if (rs < SHUTDOWN ||
                    (rs == SHUTDOWN && firstTask == null)) {
                    //如果t.isAlive()=true，说明是有问题的，都没有启动，t怎么会是活的呢。所以抛出异常。
                    if (t.isAlive()) 
                        throw new IllegalThreadStateException();
                    //把worker添加到set集合
                    workers.add(w);
                    //记录线程池中出现的最大线程数
                    int s = workers.size();
                    if (s > largestPoolSize)
                        largestPoolSize = s;
                    //添加成功标识
                    workerAdded = true;
                }
            } finally {
                //释放锁
                mainLock.unlock();
            }
            //添加成功启动线程，启动线程是调用了runWorker()
            if (workerAdded) {
                t.start();
                workerStarted = true;
            }
        }
    } finally {
         //启动失败
        if (! workerStarted)
            //启动失败需要从workers中移除当前构造的woker；工作线程数减1了；执行tryTerminate()判断是否终止线程池。
            addWorkerFailed(w);
    }
    return workerStarted;
}

runWorker()解析

final void runWorker(Worker w) {
    //获取当前线程
    Thread wt = Thread.currentThread();
    //需要执行的任务
    Runnable task = w.firstTask;
    //置为null
    w.firstTask = null;
    //这里是为了把之前的state=-1设置为state=0，此时允许中断
    w.unlock(); 
    //是否异常退出循环
    boolean completedAbruptly = true;
    try {
        //如果任务不为null或者getTask()不为null
        while (task != null || (task = getTask()) != null) {
            //获取锁。这里使用锁的目的在于标识正在处理任务
            w.lock();
             //线程池=SHUWDOWN，要保证当前线程是中断状态
             //线程池!=SHUWDOWN，要保证当前线程不是中断状态
            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 {
         //一定要注意。执行到这里说明getTask()返回null。说明当前线程池中不需要那么多线程来执行任务了，可以把多于corePoolSize数量的工作线程干掉
        processWorkerExit(w, completedAbruptly);
    }
}

getTask()解析

//什么情况才会执行getTask呢？说明工作线程数已经大于核心线程数才会执行getTask()。一定要记住这一点
private Runnable getTask() {
     //表示上次从阻塞队列中取任务时是否超时
    boolean timedOut = false; 
    for (;;) {
        int c = ctl.get();
        //获取线程池状态
        int rs = runStateOf(c);
       //如果线程池状态rs >= SHUTDOWN并且(rs >= STOP，或者阻塞队列为空)。则将workerCount减1并返回null。
        if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
            decrementWorkerCount();
            return null;
        }
         //获取工作线程数
        int wc = workerCountOf(c);
        //表示是否需要超时控制。allowCoreThreadTimeOut默认false；如果线程池数量超过核心线程数也是需要超时控制的
        boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
        //校验工作线程数量和任务队列是否为空
        if ((wc > maximumPoolSize || (timed && timedOut))
            && (wc > 1 || workQueue.isEmpty())) {
             //工作线程数量-1
            if (compareAndDecrementWorkerCount(c))
                return null;
            continue;
        }

        try {
            //根据是否超时从队列中获取任务
            Runnable r = timed ?workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) : workQueue.take();
            if (r != null)
                return r;
            timedOut = true;
        } catch (InterruptedException retry) {
            //说明发生了中断
            timedOut = false;
        }
    }
}

processWorkerExit()解析

private void processWorkerExit(Worker w, boolean completedAbruptly) {
       //是否异常，如果异常工作线程数量-1
    if (completedAbruptly) 
        decrementWorkerCount();

    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try { 
        //增加线程池完成任务数量
        completedTaskCount += w.completedTasks;
        //从worker中移除完成任务
        workers.remove(w);
    } finally {
        mainLock.unlock();
    }
    //根据线程池状态进行判断是否结束线程池
    tryTerminate();

    int c = ctl.get();
    if (runStateLessThan(c, STOP)) {
        if (!completedAbruptly) {
            int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
            if (min == 0 && ! workQueue.isEmpty())
                min = 1;
            if (workerCountOf(c) >= min)
                return; // replacement not needed
        }
        addWorker(null, false);
    }
}

tryTerminate()解析

 //根据线程池状态判断是否结束线程池
final void tryTerminate() {
    for (;;) {
        int c = ctl.get();
         //RUNNING不能结束线程池
         //线程池状态是TIDYING或TERMINATED说明线程池已经处于正在终止的路上
         //状态为SHUTDOWN，但是任务队列不为空不能结束线程池
        if (isRunning(c) ||
            runStateAtLeast(c, TIDYING) ||
            (runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty()))
            return;
         //工作线程数量不等于0，中断一个空闲的工作线程，并返回
        if (workerCountOf(c) != 0) { 
            interruptIdleWorkers(ONLY_ONE);
            return;
        }

        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
         // 设置线程池状态为TIDYING，如果设置成功，则调用terminated方法
            if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {
                try { 
                    //子类实现
                    terminated();
                } finally {
                   // 设置状态为TERMINATED
                    ctl.set(ctlOf(TERMINATED, 0));
                    termination.signalAll();
                }
                return;
            }
        } finally {
            mainLock.unlock();
        }
    }
}

interruptIdleWorker()解析

//该方法用于给空闲工作线程一个中断标识
private void interruptIdleWorkers(boolean onlyOne) {
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        //遍历worker，根据onlyOne判断，如果为ture只中断一个空闲线程
        for (Worker w : workers) {
            Thread t = w.thread;
            //线程没有被中断并且线程是空闲状态tryLock()判断是否空闲
            if (!t.isInterrupted() && w.tryLock()) {
                try {
                    t.interrupt();
                } catch (SecurityException ignore) {
                } finally {
                    w.unlock();
                }
            }
            if (onlyOne)
                break;
        }
    } finally {
        mainLock.unlock();
    }
}

监控线程池

通过线程池提供的参数进行监控
getTaskCount：线程池任务总数。
getCompletedTaskCount：线程池已完成的任务数量，小于等于completedTaskCount。
getPoolSize：线程池当前的核心线程数量。
getLargestPoolSize：线程池曾经创建过的最大线程数量。
getActiveCount：当前线程池中正在执行任务的线程数量。
最后编辑于 ：
©著作权归作者所有,转载或内容合作请联系作者
  • 人面猴
    序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
    沈念sama阅读 212,686评论 6赞 492
  • 死咒
    序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
    沈念sama阅读 90,668评论 3赞 385
  • 救了他两次的神仙让他今天三更去死
    文/潘晓璐 我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
    开封第一讲书人阅读 158,160评论 0赞 348
  • 道士缉凶录：失踪的卖姜人
    文/不坏的土叔 我叫张陵，是天一观的道长。 经常有香客问我，道长，这世上最难降的妖魔是什么？ 我笑而不...
    开封第一讲书人阅读 56,736评论 1赞 284
  • 港岛之恋（遗憾婚礼）
    正文 为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
    茶点故事阅读 65,847评论 6赞 386
  • 恶毒庶女顶嫁案：这布局不是一般人想出来的
    文/花漫 我一把揭开白布。 她就那样静静地躺着，像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上，一...
    开封第一讲书人阅读 50,043评论 1赞 291
  • 城市分裂传说
    那天，我揣着相机与录音，去河边找鬼。 笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。 我是一名探鬼主播，决...
    沈念sama阅读 39,129评论 3赞 410
  • 双鸳鸯连环套：你想象不到人心有多黑
    文/苍兰香墨 我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
    开封第一讲书人阅读 37,872评论 0赞 268
  • 万荣杀人案实录
    序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
    沈念sama阅读 44,318评论 1赞 303
  • 护林员之死
    正文 独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 36,645评论 2赞 327
  • 白月光启示录
    正文 我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 38,777评论 1赞 341
  • 活死人
    序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
    沈念sama阅读 34,470评论 4赞 333
  • 日本核电站爆炸内幕
    正文 年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
    茶点故事阅读 40,126评论 3赞 317
  • 男人毒药：我在死后第九天来索命
    文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 30,861评论 0赞 21
  • 一桩弑父案，背后竟有这般阴谋
    文/苍兰香墨 我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 32,095评论 1赞 267
  • 情欲美人皮
    我被黑心中介骗来泰国打工， 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。 一个月前我还...
    沈念sama阅读 46,589评论 2赞 362
  • 代替公主和亲
    正文 我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
    茶点故事阅读 43,687评论 2赞 351