Android(安卓)Java 线程池 ThreadPoolExecutor源码篇

线程池简单点就是任务队列+线程组成的。接下来我们来简单的了解下ThreadPoolExecutor的源码。

先看ThreadPoolExecutor的简单类图，对ThreadPoolExecutor整体来个简单的认识。

为了分析ThreadPoolExecutor我们得下扯点队列和队列里面的任务这东西。

常见三种BlockingQueue阻塞队列SynchronousQueue，LinkedBlockingQueue，ArrayBlockingQueue当然还有其他的，简单类图(只画了SynchronousQueue的简单类图)

队列里面的任务FutureTask 简单类图

按前面所说对于ThreadPoolExecutor我们先关注两个东西。

BlockingQueueu 队列他决定了任务的调度方式，我们主要关注BlockingQueue的offer, poll,take三个方法offer往队列里面添加任务如果队列已经满了话返回false，poll在规定的时间内从队列里面取出任务如果队列是空的就返回null, take也是从队列里面取出任务如果队列是空的则阻塞(保证线程池核心线程一直存在的时候有妙用)
SynchronousQueue：这种queue你直接offer()是没有用的，必须要有另外一个线程还在poll()的时候才能offer成功。要两个地方配合使用(对于SynchronousQueue来说这个元素只是走了一个过场罢了一下子就取出来了SynchronousQueue的长度一直是0)。SynchronousQueue在什么地方用呢，比如Executors.newCachedThreadPool() 这一类线程池的队列就是用的SynchronousQueue，本来这类线程池的初衷是不用队列的submit一个任务就开一个线程，任务运行完线程结束。使用SynchronousQueue是为了不用每次submit一个任务的时候都去另开线程，如果submit的时候正好有一个线程执行完了一个任务在poll的时候还是由这个线程来执行这个任务。
ArrayBlockingQueue：基于数组的queue的，先进先出。要设置queue的大小。
LinkedBlockingQueue：基于链表的queue，先进先出，可以设置也可以不设置queue的大小，不设置就是默认的大小。
BlockingQueue 队列里面放得是FutureTask，从队列里面把FutureTask任务拿出来之后调用的是FutureTask的run方法。run方法里面会调用FutureTask里面Callable的call方法，call方法调用完之后保存住了call的返回值。这样FutureTask就可以通过get方法得到这个返回值。

先说下ExecutorService(ThreadPoolExecutor实现了这个接口，从ThreadPoolExecutor的类图可以看出)接口里面一些方法具体作用。

    // 不让再submit新的任务了，但是之前提交的还是会继续执行完的。    void shutdown();    // 不让再submit新的任务了，并且尝试去停止线程池里面所有的任务，不管是正在执行的还是没有执行的，并且返回没有执行的任务列表    List<Runnable> shutdownNow();    // 线程池是否shut down了    boolean isShutdown();    // 线程池是否终止了    boolean isTerminated();    // 等待线程池终止，如果在timeout时间之内终止了就返回ture，否则返回false。一般配合shutdown函数使用    boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException;    // 提交任务    <T> Future<T> submit(Callable<T> task);    <T> Future<T> submit(Runnable task, T result);    Future<?> submit(Runnable task);    // 执行tasks里面所有的callable，返回所有的情况结果    <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks) throws InterruptedException;    // 在timeout时间内执行tasks里面所有的callable，返回所有的情况结果(包括没执行的也会返回)    <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit) throws InterruptedException;    // 执行tasks里面所有的callable，当有一个处理完了就结束    <T> T invokeAny(Collection<? extends Callable<T>> tasks) throws InterruptedException, ExecutionException;    // 在timeout时间内执行tasks里面所有的callable，当有一个处理完了就结束    <T> T invokeAny(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException;

下面正式开始ThreadPoolExecutor的源码分析（只是分析了部分函数）按照线程池的使用流程来看ThreadPoolExecutor的源码。先构造函数，在submit方法, 然后在shutdown方法。
我们得先知道线程池总共有5中状态如下所示

// runState is stored in the high-order bitsprivate 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;private static final int TERMINATED =  3 << COUNT_BITS;

RUNNING：表示线程池能够接受任务，并且可以运行队列中的任务。
SHUTDOWN：表示不接受新的任务，但是之前队列里面的任务还是会被调用(调用了shutdown()之后的状态)
STOP：表示不接受新的任务，不会执行队列中的任务，并且尝试去中断正在运行的任务(调用了shutdownNow()的状态)
TIDYING：表示所有任务都已经终止，workCount值为0(workCount可以理解成线程的个数)转到TIDYING状态的线程即将要执行terminated()钩子方法。
TERMINATED：表示terminated()方法执行结束。
5中状态的转换有以下几种方式。
RUNNING -> SHUTDOWN：调用了shutdown方法，或者线程池实现了finalize方法，在里面调用了shutdown方法。
(RUNNING or SHUTDOWN) -> STOP：调用了shutdownNow方法
SHUTDOWN -> TIDYING：当队列和线程池均为空的时候
STOP -> TIDYING：当线程池为空的时候
TIDYING -> TERMINATED：terminated()钩子方法调用完毕

ThreadPoolExecutor4个构造函数不管是从哪个构造函数进来的最后走的都是最后一个

    ...    public ThreadPoolExecutor(int corePoolSize,                              int maximumPoolSize,                              long keepAliveTime,                              TimeUnit unit,                              BlockingQueue<Runnable> workQueue) {        this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,             Executors.defaultThreadFactory(), defaultHandler);    }    public ThreadPoolExecutor(int corePoolSize,                              int maximumPoolSize,                              long keepAliveTime,                              TimeUnit unit,                              BlockingQueue<Runnable> workQueue,                              ThreadFactory threadFactory) {        this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,             threadFactory, defaultHandler);    }    public ThreadPoolExecutor(int corePoolSize,                              int maximumPoolSize,                              long keepAliveTime,                              TimeUnit unit,                              BlockingQueue<Runnable> workQueue,                              RejectedExecutionHandler handler) {        this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,             Executors.defaultThreadFactory(), handler);    }    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;    }    ...

corePoolSize：核心线程个数。
maximumPoolSize：最大线程个数最大线程个数要大于核心线程个数(maximumPoolSize>corePoolSize)
workQueue: 线程池任务队列(线程池关键地方，有时候注重任务出队的顺序或者任务有优先级都要靠他来实现)。
keepAliveTime：核心线程之外的线程如果达到了这个空闲时间线程自动关闭(当然也可以作用于核心线程通过allowsCoreThreadTimeOut()函数)。
unit：keepAliveTime时间的单位。
threadFactory：线程工程用来创建线程的，把这个暴露给我们是为了让我们可以控制创建线程的一些行为，比如设置线程的优先级，名字，debug等等。
handler：对于reject的任务该怎么处理就是靠这个来完成的(当workQueue满了并且达到了最大线程的个数的时候会拒绝加进来的任务，或者调用了shutdown函数之后再加入任务也是会reject的)。

submit函数(AbstractExecutorService类中)

    ...    public Future<?> submit(Runnable task) {        if (task == null) throw new NullPointerException();        RunnableFuture<Void> ftask = newTaskFor(task, null);        execute(ftask);        return ftask;    }    public <T> Future<T> submit(Runnable task, T result) {        if (task == null) throw new NullPointerException();        RunnableFuture<T> ftask = newTaskFor(task, result);        execute(ftask);        return ftask;    }    public <T> Future<T> submit(Callable<T> task) {        if (task == null) throw new NullPointerException();        RunnableFuture<T> ftask = newTaskFor(task);        execute(ftask);        return ftask;    }    ...

不管调用的是哪个submit方法都是先构造出一个RunnableFuture(FutureTask) 然后调用execute方法。不管你submit的时候传入的是Runnable还是Callable最后RunnableFuture(FutureTask)里面都会生成Callable对象。任务调用的时候调用RunnableFuture(FutureTask)的run方法，run方法调用Callable对象的call方法。接着看execute方法。

    ...    public void execute(Runnable command) {        if (command == null)            throw new NullPointerException();        /*        * Proceed in 3 steps:        *        * 1. If fewer than corePoolSize threads are running, try to        * start a new thread with the given command as its first        * task.  The call to addWorker atomically checks runState and        * workerCount, and so prevents false alarms that would add        * threads when it shouldn't, by returning false.        *        * 2. If a task can be successfully queued, then we still need        * to double-check whether we should have added a thread        * (because existing ones died since last checking) or that        * the pool shut down since entry into this method. So we        * recheck state and if necessary roll back the enqueuing if        * stopped, or start a new thread if there are none.        *        * 3. If we cannot queue task, then we try to add a new        * thread.  If it fails, we know we are shut down or saturated        * and so reject the task.        */        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);    }    ...

注释中我们也看到了总共分为三步。1. 如果当前线程池中的线程个数小于核心线程数则调用addWorker方法第一个参数是任务，第二个参数表示是不是核心线程（addWorker等下再看）。2. 如果当前的线程池是RUNNING状态则任务加入线程池，加入到队列之后再检测一次状态如果不是RUNNING状态，把这个任务从任务中移除 reject这个任务(reject等下再看)，else如果线程池中线程为0表示没有指定核心线程个数，还是addWorker 注意addWorker的参数。3. 可能是队列满了用核心线程之外的线程去处理任务还是addWorker。

这里我们两个函数我们没有分析reject 和 addWorker函数，先看简单的reject函数。reject简单点

    final void reject(Runnable command) {        handler.rejectedExecution(command, this);    }

handler 是RejectedExecutionHandler 对于reject的任务可以做不同的处理，抛出异常或者不做任何处理。随你如果你想自己处理，这个应该还好说。

addWorker通过这个函数去开线程，第一个参数firstTask如果不为空则开的线程直接执行这个Runnable，如果为空则开的线程去队列里面拿任务，第二个参数core表示准备开的线程是不是核心线程判断线程个数用的。

    private boolean addWorker(Runnable firstTask, boolean core) {        retry:        for (;;) {            int c = ctl.get();            int rs = runStateOf(c);            // Check if queue empty only if necessary.            if (rs >= SHUTDOWN &&                ! (rs == SHUTDOWN &&                   firstTask == null &&                   ! workQueue.isEmpty()))                return false;            for (;;) {                int wc = workerCountOf(c);                if (wc >= CAPACITY ||                    wc >= (core ? corePoolSize : maximumPoolSize))                    return false;                if (compareAndIncrementWorkerCount(c))                    break retry;                c = ctl.get();  // Re-read ctl                if (runStateOf(c) != rs)                    continue retry;                // 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) {                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());                    if (rs < SHUTDOWN ||                        (rs == SHUTDOWN && firstTask == null)) {                        if (t.isAlive()) // precheck that t is startable                            throw new IllegalThreadStateException();                        workers.add(w);                        int s = workers.size();                        if (s > largestPoolSize)                            largestPoolSize = s;                        workerAdded = true;                    }                } finally {                    mainLock.unlock();                }                if (workerAdded) {                    t.start();                    workerStarted = true;                }            }        } finally {            if (! workerStarted)                addWorkerFailed(w);        }        return workerStarted;    }

8~12行 if (rs >= SHUTDOWN && !(rs == SHUTDOWN && firstTask == null &&! workQueue.isEmpty())) 转换成if (rs >= SHUTDOWN && （(rs ！= SHUTDOWN || firstTask ！= null || workQueue.isEmpty()))）前半部分状态是这四种才能进入SHUTDOWN，STOP，TIDYING，TERMINATED 这四种状态才会进入（这里注意前面说的线程池各种状态的含义哦），后半部分第一个条件rs ！= SHUTDOWN 又给我们去掉了一种状态STOP，TIDYING，TERMINATED这三种状态时既不能去执行新加的任务也不能执行队列里面的任务直接return false。第二个条件状态为SHUTDOWN且firstTask ！= null意思是说在SHUTDOWN状态还想添加新的任务return false(SHUTDOWN状态是不能添加新任务吧)，第三个添加workQueue.isEmpty()表示SHUTDOWN且 firstTask == null 且队列为空,表示SHUTDOWN状态去队列里面取任务执行，但是这个时候队列里面没有任务了return false。这里可能说的有点乱总结下这几行代码。三种情况
1）线程池状态是STOP，TIDYING，TERMINATED 不能再去增开线程，不管你开的这个线程是去取队列的任务还是直接执行你submit的任务都是不可以的。
2）线程池状态是SHUTDOWN的你又想去开线程执行你submit的任务对不起reject
3）线程池状态是SHUTDOWN 队列里面没有任务，这个时候你又想去队列里面取出任务执行。对不起不行。队列为空你肯定取不到这个任务。
16~18行，要开的线程是核心线程线程池个数肯定要小于核心线程个数吧，不是核心线程，线程个数肯定要小于最大线程个数吧。
19行，线程池里面的线程个数加一。
32行，出现了Worker worker里面有一个Thread 这个我们等下再看。先往下看
47行，放到workers里面去，workers 里面放的是所有的线程。先这么理解吧。
57行，t.start() Worker里面的线程池跑起来了。

接下来就该到了Worker 类了。
Worker 类实现了Runnable方法。注意Worker的构造函数里面this.thread = getThreadFactory().newThread(this); newThread的参数是this是Worker本身。所以前面addWorker函数里面t.start() Worker里面的线程跑起来了。直接调用的是Woker类里面自身的run方法。那就直接看Worker类里面的run方法。

    public void run() {        runWorker(this);    }

恩 runWorker(this)，又是把自身给传入进去了，没什么说的进去看吧，这里面干的事情就是没执行完一个任务又去队列里面取下一个任务执行，如果没取到线程结束线程个数减掉1。看具体的实现。ThreadPoolExecutor 类里面runWorker函数

    final void runWorker(Worker w) {        Thread wt = Thread.currentThread();        Runnable task = w.firstTask;        w.firstTask = null;        w.unlock(); // allow interrupts        boolean completedAbruptly = true;        try {            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                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);        }    }

第8行 while (task != null || (task = getTask()) != null) 先判断是不是去执行submit进来的任务如果不是则是去队列里面取任务执行(while完之后task又会赋值null让他去队列里面取任务执行)。getTask函数我们等下再看，我们知道了他是去队列里面取任务的。
第20行和第31行 beforeExecute(wt, task); afterExecute(task, thrown);给我们上层重写用的每个任务的执行前和执行后都会调用者两个方法。
第23行 task.run(); 真正每个任务要做的逻辑在这个里面。而且我们前面也说过task是FutureTask，调用FutureTask里面的run方法会调用FutureTask里面Callable的call方法，call方法调用完之后保存住了call的返回值。FutureTask 可以通过get方法得到这个返回值。
第34行看task = null;了吧。

到这里就差getTask() 方法了。

    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?            boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;            if ((wc > maximumPoolSize || (timed && timedOut))                && (wc > 1 || workQueue.isEmpty())) {                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;            }        }    }

9~12行，两种情况第一种线程池状态是STOP，TIDYING，TERMINATED不能再去队列里面拿任务执行了，第二种线程池状态是SHUTDOWN 队列里面又没有任务不能再提供任务这个线程了。workcount减掉1
17行，判断这个线程有没有timeout，如果没有timeout的线程是不会自动停掉的会一直存在(因为有的时候想一直保持核心线程的个数如果没有特殊的设置)。allowCoreThreadTimeOut如果设置了则所有的线程都有timeout包括核心线程，wc > corePoolSize 为了让核心线程之外的线程能够停掉。
27~29行从队列中去取任务 poll在规定的时间内从队列里面取出任务如果队列是空的就返回null(表示没取到线程也就结束了), take也是从队列里面取出任务如果队列是空的则阻塞保证线程池里面的核心线程数量的线程一直存在。队列里面poll和take方法的妙用。

submit函数的调用过程就说完了，下面是shutdown函数

    public void shutdown() {        final ReentrantLock mainLock = this.mainLock;        mainLock.lock();        try {            checkShutdownAccess();            advanceRunState(SHUTDOWN);            interruptIdleWorkers();            onShutdown(); // hook for ScheduledThreadPoolExecutor        } finally {            mainLock.unlock();        }        tryTerminate();    }

5行，使得线程可以shutdown（interrupt）
6行，切换线程池的状态
7行，interruptIdleWorkers方法中断空闲的线程。接下来分析
8行，ouShutdown（）空方法给我们重写用的。
12行，尝试去terminate线程池，接下来分析。
这里我们要分析interruptIdleWorkers和tryTerminate方法

先interruptIdleWorkers 这个方法是去中断空闲线程。

    private void interruptIdleWorkers() {        interruptIdleWorkers(false);    }

在进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();        }    }

第6行，拿到worker对应的Thread。
第7行，如果当前线程没有被中断且可以拿到worker的锁，则中断worker对应的线程。如果我们拿到worker的锁说明worker对应的线程是空闲的，为什么这么说呢，看worker的run方法。lock加锁是在while里面的。

tryTerminate方法当线程池的状态为SHUTDOWN且任务队列为空，需要将池的状态转变为TERMINATED；当池的状态为STOP且池中的当前活动线程数为0，要将池的状态转换成TERMINATED。

    final void tryTerminate() {        for (;;) {            int c = ctl.get();            if (isRunning(c) ||                runStateAtLeast(c, TIDYING) ||                (runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty()))                return;            if (workerCountOf(c) != 0) { // Eligible to terminate                interruptIdleWorkers(ONLY_ONE);                return;            }            final ReentrantLock mainLock = this.mainLock;            mainLock.lock();            try {                if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {                    try {                        terminated();                    } finally {                        ctl.set(ctlOf(TERMINATED, 0));                        termination.signalAll();                    }                    return;                }            } finally {                mainLock.unlock();            }            // else retry on failed CAS        }    }

4~7行，表示有三种情况是不用设置线程池的状态到TERMINATED，第一种当前线程是RUNNING的状态，第二种当前状态为TIDYING或TERMINATED，池中的活动线程数已经是0，第三种当前状态是SHUTDWON，但队列中还有任务也不用做处理因为这种情况下任务还是要处理掉的。
剩下的情况就两种吧才会继续往下走吧一种当前的状态是STOP不管队列里面有没有任务，还有一种是当前状态是SHUTDOWN状态且队列里面没有任务了。这两种情况会想办法切换到TERMINATED状态去。
8行，线程池中的线程数不为0，只是尝试去中断一个空闲线程，为什么这么干还没理解。
16行，把线程池状态切换到TIDYING。
18行，terminated();给我们重写用的。
20行，把线程池状态切换到TERMINATED。

总结。

线程池先看队列是什么形式的队列，是先进先出的，还是在入队的时候会做排序操作。关注队列的offer，poll，take方法。offer入队的时候调用，poll当我们对线程设置了timeout的时候会调用poll方法去队列里面去任务如果指定时间内没取到改线程也就结束了。take阻塞的形式去取任务线程不会退出有的时候用了保证核心线程个数的线程一直存在。
线程池队列里面的任务FutureTask 里面实现了run方法，run方法里面又会调用FutureTask里面Callable对象的call方法，所以每个任务在入队的时候不管你submit方法传入的是Runnable 还是 Callable 最后都会同意成Callable。
只有当达到了核心线程数，并且队列满了的时候才会去启动其他的线程。

更多相关文章

随机推荐