java多线程-线程池

大纲：

1. 线程池的状态
2. 构造函数
3. 线程池执行任务的过程
4. Worker
5. 线程池执行任务的主要方法
6. 中断线程池主要方法

简要说明：

1. 版本java1.8
2. 以ThreadPoolExecutor线程池为代表介绍线程池。

一、线程池的状态

线程池由5个状态(ThreadPoolExecutor类中由5个常量标识5种状态)

1. RUNNING：正常活跃的线程池，可以接受任务并执行。
2. SHUTDOWN：不接受新的任务，但会完成已经进入阻塞队列的任务。
3. STOP：不接受新的任务，不处理进入阻塞队列的任务，并interrupt当前正在执行的任务.
4. TIDYING：所有任务已经终止,工作线程数为0，将要执行terminated钩子方法。
5. TERMINATED：terminated钩子方法执行完毕。

二、构造函数

2.1看一个比较全的构造函数，剩下的大同小异

public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue,
                              ThreadFactory threadFactory,
                              RejectedExecutionHandler handler) 

• corePoolSize：核心线程数
• maximumPoolSize：最大线程数
• keepAliveTime：默认是非核心线程的空闲存活时间（根据成员变量allowCoreThreadTimeOut来决定是否包括核心线程的空闲存活时间）
• workQueue：存放任务的阻塞队列
• threadFactory：创建线程的线程工厂
• handler：拒绝策略处理器

2.2拒绝策略

其中ThreadPoolExecutor类中内置了4个拒绝策略处理器

1. AbortPolicy：拒绝任务并抛出异常。场景：应用不能承受大并发，及时抛出异常使系统发现。
2. CallerRunsPolicy：线程池状态为running时，不用线程池执行任务，用调用线程执行。非running状态拒绝任务。场景：线程池仅仅是帮助增加吞吐量的场景，线程不够则调用线程自己完成任务。
3. DiscardPolicy：直接丢弃任务，不做任何响应。场景：执行无关紧要的任务的时候。
4. DiscardOldestPolicy：线程池状态为running时，丢弃队列头的任务，然后然后重新尝试执行刚刚进来提交的任务。非running状态拒绝任务。场景：根据实际需求来判定是否能够丢弃队列头的任务。

这4中还不能满足的话，可以自己实现RejectedExecutionHandler接口。

2.3阻塞队列

阻塞队列是存放任务的地方，核心线程被占用满了之后，任务会放进阻塞队列中。

几种特殊队列：

1. synchronousQueue:这个队列没有容量，核心线程占用满了后就直接开始创建非核心线程。
2. PriorityBlockingQueue:支持实现comparable接口的任务的排序。
3. DelayQueue:支持延迟获取任务。

三、线程池执行任务的过程

1. 任务提交给线程池后，线程池首先创建核心线程来处理任务，核心线程默认是不销毁的，处理完任务后阻塞等待后续任务。
2. 有一个属性allowCoreThreadTimeOut默认false，方为true时，核心线程超过等待时间也会消亡。
3. 当任务较多时，核心线程数满且每个核心线程都在执行任务，这时再提交进来的任务则被放进阻塞队列中。
4. 当阻塞队列也满时（用有界队列的时候，无界队列不存在这个问题），线程池开始创建非核心线程来执行任务，非核心任务有最大空闲时间，没有任务的时候非核心线程阻塞至空闲时间后被消亡。

四、内部类worker

继承了aqs，同时实现了Runnable，这个内部类封装了工作线程，所有Worker对象存在一个名为workers的HashSet<Worker>中。因此这个workers相当于持有所有worker线程的引用。

private final class Worker extends AbstractQueuedSynchronizer implements Runnable {

    final Thread thread;//工作线程
    Runnable firstTask;//第一个任务
    Worker(Runnable firstTask) {
        setState(-1); // 不允许打断直到执行runWorker
        this.firstTask = firstTask;
        this.thread = getThreadFactory().newThread(this);
    }

    public void run() {
        runWorker(this);//线程开始时将执行runWorker
    }
}

 五、线程池执行任务的主要方法

5.1提交任务submit

向线程池中提交一个任务:executor.submit(()->System.out.println(1))

submit方法在ThreadPoolExecutor继承的抽线类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;
    }
    //创建FutureTask
    protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
        return new FutureTask<T>(callable);
    }

submit方法就是将我们提交的任务包装成要给FutureTask，并提交execute方法。

execute方法是顶层接口Executor中的方法，由具体的线程池实现，这里自然讨论的是由ThreadPoolExecutor线程池实现，Executor中只有这一个方法，足以见得这个方法对线程池的重要性。

5.2执行任务execute

有一个ctl成员变量比较特殊，是一个AtomicInteger，低29位存放worker数量，高3位存放线程池状态

private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));

execute方法分为三个部分

    public void execute(Runnable command) {
        if (command == null)
            throw new NullPointerException();
        int c = ctl.get();
        //当worker数量小于核心线程数，执行addWorker，自动检查线程池状态和worker数量，并把传入的任务作为worker的第一个任务执行。
        if (workerCountOf(c) < corePoolSize) {
            if (addWorker(command, true))
                return;
            c = ctl.get();
        }
        //worker数量达到核心线程数，接下来的任务开始进入阻塞队列
        if (isRunning(c) && workQueue.offer(command)) {
            int recheck = ctl.get();
            //线程池非runngin状态则拒绝任务
            if (! isRunning(recheck) && remove(command))
                reject(command);
            //当工作线程为0的时候，addWorker（这种情况可能发生在核心线程数设置为0的时候发生）
            else if (workerCountOf(recheck) == 0)
                addWorker(null, false);
        }
        //队列满后，再次尝试addWorker（任何时候Worker都有可能消亡）
        //添加失败，执行决绝策略
        else if (!addWorker(command, false))
            reject(command);
    }

5.3添加工作线程addWorker

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);
                //检查是否worker数量超过核心线程数/最大线程数
                if (wc >= CAPACITY ||
                        wc >= (core ? corePoolSize : maximumPoolSize))
                    return false;
                if (compareAndIncrementWorkerCount(c))//cas增加worker数量，后面真正add失败了会减的
                    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;
        ThreadPoolExecutor.Worker w = null;
        try {
            w = new ThreadPoolExecutor.Worker(firstTask);//创建一个worker
            final Thread t = w.thread;
            if (t != null) {
                final ReentrantLock mainLock = this.mainLock;
                mainLock.lock();//添加worker是线程安全的
                try {
                    // Recheck while holding lock.
                    // Back out on ThreadFactory failure or if
                    // shut down before lock acquired.
                    int rs = runStateOf(ctl.get());
                    //线程池状态为running或为shutdown但创建的是非核心线程的worker
                    if (rs < SHUTDOWN || (rs == SHUTDOWN && firstTask == null)) {
                        if (t.isAlive()) // precheck that t is startable
                            throw new IllegalThreadStateException();
                        workers.add(w);//添加到workers中
                        int s = workers.size();
                        if (s > largestPoolSize)
                            largestPoolSize = s;
                        workerAdded = true;
                    }
                } finally {
                    mainLock.unlock();
                }
                if (workerAdded) {
                    t.start();//开启线程，本质上是调用了worker中的runWorker方法
                    workerStarted = true;
                }
            }
        } finally {
            if (! workerStarted)
                addWorkerFailed(w);
        }
        return workerStarted;
    }

添加工作线程后，执行worker线程

5.4启动worker线程runWorker

final void runWorker(ThreadPoolExecutor.Worker w) {
        Thread wt = Thread.currentThread();
        Runnable task = w.firstTask;
        w.firstTask = null;
        w.unlock(); // 允许线程被打断
        boolean completedAbruptly = true;
        try {
            //获取任务并执行，当队列中的任务为空则阻塞
            while (task != null || (task = getTask()) != null) {
                w.lock();//这里的lock是为了后面shutdown时候，不中断正在执行的线程。
                //如果线程池是stop状态，确保线程被中断，如果不是stop，确保线程池没有被中断。
                //当我们清空中断标志时，第二种情况需要需要有一个recheck来应对shutdownNow方法。
                if ((runStateAtLeast(ctl.get(), STOP) || (Thread.interrupted() && runStateAtLeast(ctl.get(), STOP))) && !wt.isInterrupted())
                    wt.interrupt();
                try {
                    beforeExecute(wt, task);//线程执行前执行的方法，ThreadPoolExecutor中为空方法，需要执行的话继承ThreadPoolExecutor重写该方法
                    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);//线程执行后执行的方法，ThreadPoolExecutor中为空方法，需要执行的话继承ThreadPoolExecutor重写该方法
                    }
                } finally {
                    task = null;
                    w.completedTasks++;
                    w.unlock();
                }
            }
            completedAbruptly = false;
        } finally {
            //移除worker，统计执行完成任务数量等收尾工作
            processWorkerExit(w, completedAbruptly);
        }
    }

worker线程启动后，将不断从队列中获取任务并执行，直到中断线程池，或者超时。

5.5获取任务getTask

private Runnable getTask() {
        boolean timedOut = false; //最后一次获取任务是否超时

        for (;;) {
            int c = ctl.get();
            int rs = runStateOf(c);

            // shutdown状态队列为空，或者stop状态，返回null
            if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
                decrementWorkerCount();
                return null;
            }
            int wc = workerCountOf(c);
            boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;//是否获取任务超时。获取任务的是非核心worker线程，或者allowCoreThreadTimeOut被设置为true后所有worker线程，为true

            if ((wc > maximumPoolSize || (timed && timedOut))
                    && (wc > 1 || workQueue.isEmpty())) {
                if (compareAndDecrementWorkerCount(c))
                    return null;
                continue;
            }

            try {
                //从阻塞队列中获取任务，这2个都是阻塞方法，poll设置了超时
                Runnable r = timed ?
                        workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                        workQueue.take();
                if (r != null)
                    return r;
                timedOut = true;
            } catch (InterruptedException retry) {
                timedOut = false;
            }
        }
    }

5.6拒绝任务reject

    final void reject(Runnable command) {
        handler.rejectedExecution(command, this);//这里的handler处理器就是构造函数传进来的拒绝策略
    }

六、中断线程池主要方法

6.1中断shutdown

    public void shutdown() {
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            checkShutdownAccess();//检查性方法
            advanceRunState(SHUTDOWN);//把线程池状态改变为SHUTDOWN
            interruptIdleWorkers();//中断没有执行任务的线程
            onShutdown(); //  ScheduledThreadPoolExecutor线程池的钩子方法，本例为空方法
        } finally {
            mainLock.unlock();
        }
        tryTerminate();
    }

    private void interruptIdleWorkers() {
        interruptIdleWorkers(false);
    }
    private void interruptIdleWorkers(boolean onlyOne) {
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            for (ThreadPoolExecutor.Worker w : workers) {
                Thread t = w.thread;
                //上面runWorker正在执行任务的worker线程会lock，因此这里是中断没有在运行任务的空闲worker线程（tryLock成功表示runWorker方法中的任务没有在执行，这里利用了独占锁特性）
                if (!t.isInterrupted() && w.tryLock()) {
                    try {
                        t.interrupt();
                    } catch (SecurityException ignore) {
                    } finally {
                        w.unlock();
                    }
                }
                if (onlyOne)
                    break;
            }
        } finally {
            mainLock.unlock();
        }
    }

6.2中断shutdownNow

public List<Runnable> shutdownNow() {
        List<Runnable> tasks;
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            checkShutdownAccess();//检查性方法
            advanceRunState(STOP);//把线程池状态改变为STOP
            interruptWorkers();//中断所有worker
            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();
        }
    }