深刻浅出Java线程池ThreadPoolExecutor

时间  2019-11-24
标签 深刻 java 线程 threadpoolexecutor 栏目 Java 繁體版
原文   原文链接

简书江溢Jonny，转载请注明原创出处，谢谢！java

本文基于JDK1.7的源码进行分析并解读。咱们将以一个简单的案例开始本话题的讨论，而且在文章的结尾，笔者将会给出一些经验之谈，避免后来者踩坑。编程

若是你喜欢我，能够关注个人公众号~更多干货~ 缓存

ThreadPoolExecutor是JUC提供的一类线程池工具，也是Java语言中应用场景最多的并发框架，能够说，几乎全部须要异步或者并发执行的，均可以使用Java线程池。那么首先，咱们一块儿来比较一下“单纯使用线程的方案”和“使用ThreadPoolExecutor线程池的方案”，在解决问题上有什么区别吧。安全

案例：抄写员

在中世纪，有一种叫作抄写员（Scribe）的工做，他们的职责就好像是复印机，抄写一本又一本书。假如这个时候有一个抄写员工做室，只有2个抄写员，他们要抄写10本书。bash

咱们在本例中分别“本身写线程管理”和“由ThreadPoolExecutor作线程管理”服务器

public static class Book {
 
    private static AtomicInteger id = new AtomicInteger(0); // 书名生成器
    private String bookName; // 书名
     
    public void copy() { // 抄写书籍
        System.out.println("start copy " + bookName);
        try {
            Thread.sleep(100L); // sleep 100ms
        } catch (Exception e) {
            // ignore
        }
        System.out.println("end copy " + bookName);
    }
 
    public Book() {
        bookName = "book-" + String.valueOf(id.incrementAndGet()); // 书名自动生成
    }
}
复制代码

本身实现线程管理

// 提早准备好十本书
final BlockingQueue<Book> books = new LinkedBlockingDeque<Book>(10);
for (int i = 0; i < 10; i++) {
    try {
        books.put(new Book());
    } catch (Exception e) {
        // ignore
    }
}
 
 
System.out.println("start work...");
// 建立两个书籍抄写员线程
Thread[] scribes = new Thread[2];
for (int scribeIndex = 0; scribeIndex < 2; scribeIndex++) {
    scribes[scribeIndex] = new Thread(new Runnable() {
        public void run() {
            for (; ; ) {
                if (Thread.currentThread().isInterrupted()) {
                    System.out.println("time arrives, stop writing...");
                }
                try {
                    Book currentBook = books.poll(5, TimeUnit.SECONDS);
                    currentBook.copy();
                } catch (Exception e) {
                    System.out.println("time arrives, stop writing...");
                    return;
                }
            }
        }
    });
    scribes[scribeIndex].setDaemon(false); // 设置为非守护线程
    scribes[scribeIndex].start();
}
 
// 工做已经安排下去了，安心等待就行了
try {
    Thread.sleep(10000l);
} catch (Exception e) {
    // ignore
}
 
// 时间到了，提醒两个抄写员中止抄写
for (int scribeIndex = 0; scribeIndex < 2; scribeIndex++) {
    scribes[scribeIndex].interrupt();
}
 
System.out.println("end work...");
复制代码

写了一大堆代码来完成上述的功能，让咱们一块儿来看看用了ThreadPoolExecutor是怎么完成的。并发

System.out.println("start work...");
ExecutorService executorService = Executors.newFixedThreadPool(2);
for (int i = 0; i < 10; i ++) {
    executorService.submit(new Runnable() {
        public void run() {
            new Book().copy();
        }
    });
}
 
// 工做已经安排下去了，安心等待就行了
try {
    Thread.sleep(10000l);
} catch (Exception e) {
    // ignore
}
 
executorService.shutdownNow();
System.out.println("end work...");
复制代码

整个流程很是清晰，分别是：任务编写、线程建立、线程启动、终止线程。框架

可是不少时候，问题并不只限于上述的内容。异步

开发者困境

最先的并发编程的开发者不少事情都须要亲力亲为，而经过使用Java线程池，能够完成如下工做：函数

1）线程管理，线程的建立、启动、销毁等工做；

2）线程复用，线程的建立是会给服务器带来必定开销的，如何减小频繁重复建立线程的开销；

3）弹性伸缩，服务器一般有高峰期也有低峰期，线程池是否能够弹性伸缩，好比线程建立成功后长时间不使用是否能够回收，以减小系统资源的浪费，或者线程池的容量是否能够随时增加；

4）拒绝策略，线程数量有限而须要处理的任务不少，超出系统承载范围的任务是拒绝仍是阻塞等待；

5）异常处理，线程在执行过程当中可能遇到异常或者错误，开发者如何正确应对这些异常或者错误；

6）任务分配，任务的分配是基于先入先出仍是基于某种优先级的策略。

等等如是，不一而足，这个时候咱们就要介绍Doug Lea大神开发的ThreadPoolExecutor线程池框架，看看大神是怎么解决上面这些问题的。

ThreadPoolExecutor源码简析

首先，在解读源码以前，要引入ThreadPoolExecutor的一些重要概念

生命周期

在ThreadPoolExecutor线程池的设计中，把整个任务执行框架线程池划分为5个生命周期：

RUNNING：容许接收新任务而且处理队列中的任务

SHUTDOWN：再也不接收新的任务，仅消化完队列中的任务

STOP：不只再也不接收新的任务，连队列中的任务都再也不消化处理了，而且尝试中断正在执行任务的线程

TIDYING：全部任务被终止了，工做线程数workCount也被设为0，线程的状态也被设为TIDYING，并开始调用钩子函数terminated()

TERMINATED：钩子函数terminated()执行完毕

各个生命周期的转化图以下：

线程池的生命周期

从图中能够看到，整个生命周期的变化是不可逆的。

状态字

ThreadPoolExecutor把线程池状态和线程池容量打包成一个int型变量，以下图所示

状态字

线程池状态位

状态 高位值枚举 正负性
RUNNING 111 负数（-536870912）
SHUTDOWN 000 0
STOP 001 正数（536870912）
TIDYING 010 正数（1073741824）
TERMINATED 011 正数（1610612736）

所以在状态值的排布上能够知道 TERMINATED > TIDYING > STOP >SHUTDOWN > RUNNING

ThreadPoolExecutor中的代码以下所示：

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;
 
// 状态字的高比特位存放线程池状态信息
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;
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; }
  
// 判断当前线程池是否正在执行
private static boolean isRunning(int c) {
    return c < SHUTDOWN;
}
复制代码

线程池主要执行流程

线程池主要执行流程

代码介绍

首先，咱们建立一个线程池。

1.线程池建立
ExecutorService executorService = Executors.newFixedThreadPool(2);
复制代码

这里使用了Executors提供的工厂方法，能够建立如下四种类型线程池：

newFixedThreadPool。该方法将用于建立一个固定大小的线程池（此时corePoolSize = maxPoolSize），每提交一个任务就建立一个线程池，直到线程池达到最大数量，线程池的规模在此后不会发生任何变化；

newCachedThreadPool。该方法建立了一个可缓存的线程池，（此时corePoolSize = 0，maxPoolSize = Integer.MAX_VALUE），空闲线程超过60秒就会被自动回收，该线程池存在的风险是，若是服务器应用达到请求高峰期时，会不断建立新的线程，直到内存耗尽；

newSingleThreadExecutor。该方法建立了一个单线程的线程池，该线程池按照任务在队列中的顺序串行执行（如：FIFO、LIFO、优先级）；

newScheduledThreadPool。该方法建立了一个固定长度的线程池，能够以延迟或者定时的方式执行任务；

二、任务提交

任务提交的大概逻辑以下：

1）当线程池小于corePoolSize时，新提交任务将建立一个新线程执行任务，即便此时线程池中存在空闲线程；

2）当线程池达到corePoolSize时，新提交任务将被放入workQueue中，等待线程池中任务调度执行；

3）当workQueue已满，且maximumPoolSize > corePoolSize时，新提交任务会建立新线程执行任务；

4）当提交任务数超过maximumPoolSize时，新提交任务由RejectedExecutionHandler处理；

5）当线程池中超过corePoolSize线程，空闲时间达到keepAliveTime时，关闭空闲线程；

那么接下来咱们看看源代码是怎么实现上面的描述的

线程池建立成功之后，咱们提交任务到线程池中：

executorService.submit(new Runnable() {
    public void run() {
        new Book().copy();
    }
});
复制代码

submit到线程池之后：

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;
}
复制代码

能够看到ThreadPoolExecutor的入口方法就是execute(Runnable commad)。该方法的执行逻辑以下：

int c = ctl.get();
// 1. 若是当前线程池中线程总数少于核心线程数，则添加新线程到线程池中，
// 而且由新线程执行刚提交进来的任务
if (workerCountOf(c) < corePoolSize) {
    if (addWorker(command, true))
        return;
    c = ctl.get();
}
  
if (isRunning(c) && workQueue.offer(command)) {
    int recheck = ctl.get();
    // 2. 可能刚才在建立新线程成功的同时，线程池被关闭了，所以须要double-check，
    // 若是此时线程池已经被关闭了，那么回滚刚才被添加进来的任务
    if (! isRunning(recheck) && remove(command))
        reject(command);
    else if (workerCountOf(recheck) == 0)
        addWorker(null, false);
}
// 3. 若是此时核心线程数（corePoolSize）已经满了，而且任务队列也满了，
// 尝试增长线程到maximumPoolSize大小，若是仍然失败，执行拒绝策略
else if (!addWorker(command, false))
    reject(command);
复制代码

在上面的代码里面，ctl.get()方法、workerCountOf()、以及isRunning() 方法都是对以前提到的状态字进行读写的操做，这部分咱们就再也不展开给读者看了，有兴趣的读者能够本身了解一下。

接下来，咱们看看addWorker都作了什么工做：

private boolean addWorker(Runnable firstTask, boolean core) {
    // 这部分省略的代码都是对状态字进行修改，添加并建立线程以前，
    // 须要递增work记数（此时须要线程安全地操做）
    ...
 
    boolean workerStarted = false;
    boolean workerAdded = false;
    Worker w = null;
    try {
        ...
        w = new Worker(firstTask); // 此处封装出了一个新的Work，这个类咱们稍后会介绍
        final Thread t = w.thread;
        if (t != null) {
            ...
             
                // 得到线程池状态，若是线程池已经被关闭了，则再也不建立新的线程
                int c = ctl.get();
                int rs = runStateOf(c);
 
                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();
                    ...
                    workerAdded = true;
            ...
            if (workerAdded) {
                t.start();
                workerStarted = true;
            }
        }
    } finally {
        if (! workerStarted)
            // 若是任务启动或者提交到线程池失败，
            // 则执行回滚操做（从工做线程池中移除失败添加的worker、减小状态字中的任务计数）
            addWorkerFailed(w);
    }
    return workerStarted;
}

复制代码
三、任务执行

任务执行在Worker类中，而Worker类是一个继承了Runnable接口的类。

private final class Worker extends AbstractQueuedSynchronizer implements Runnable {
    ...
    public void run() {
        runWorker(this);
    }
    ...
}
复制代码

能够看到Worker类中调用了外部的runWorker()方法。所以能够了解到，任务执行的主要逻辑，就是在外部的runWorker()方法中执行的

final void runWorker(Worker w) {
    Thread wt = Thread.currentThread();
    Runnable task = w.firstTask;
    w.firstTask = null;
    ...
    boolean completedAbruptly = true;
    try {
        while (task != null || (task = getTask()) != null) { // 循环读取任务
            ...
            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++;
                ...
            }
        }
        completedAbruptly = false;
    } finally {
        processWorkerExit(w, completedAbruptly);
    }
}

复制代码

beforeExecute和afterExecute是两个钩子方法，在里面指定了当线程开始执行和完成执行之后执行的动做，须要开发者实现。

另外须要注意的还有runWorker方法内调用的getTask()方法，在该方法内部，若是发生如下状况将会返回null，终止工做线程的执行循环： 1）当前线程数即将超过maxPoolSize 2）线程池被关闭 3）当前线程数大于corePoolSize且小于maxPoolSize，并关切从BlockingQueue取数据超过了超时时间（默认60秒）

代码实现以下：

private Runnable getTask() {
    boolean timedOut = false; // Did the last poll() time out?
 
    retry:
    for (;;) {
        int c = ctl.get();
        int rs = runStateOf(c);
 
        // 校验当前线程池状态
        if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
            decrementWorkerCount();
            return null;
        }
 
        boolean timed;      // Are workers subject to culling?
 
        for (;;) {
            int wc = workerCountOf(c);
            timed = allowCoreThreadTimeOut || wc > corePoolSize;
 
            if (wc <= maximumPoolSize && ! (timedOut && timed))
                break;
            if (compareAndDecrementWorkerCount(c))
                return null;
            c = ctl.get();  // Re-read ctl
            if (runStateOf(c) != rs)
                continue retry;
            // else CAS failed due to workerCount change; retry inner loop
        }
 
        try {
            // 若是线程超过指定时间内（默认60秒）没有获取到任务，说明有线程即将过时
            Runnable r = timed ?
                workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                workQueue.take();
            if (r != null)
                return r;
            timedOut = true;
        } catch (InterruptedException retry) {
            timedOut = false;
        }
    }
}

复制代码
四、任务拒绝

若是线程被提交到线程池时，当前线程池出现如下状况的任一一种状况： 1）线程池任务队列已经满了 2）线程池被关闭了（调用了shutdown函数或者shutdownNow函数） 都将会调用提早设置好的回调策略，ThreadPoolExecutor中总共提供了四种策略：

1）AbortPolicy（停止）：该策略将会直接抛出RejectedExecutionException异常，调用者将会得到异常；

2）DiscardPolicy（抛弃）：使用该策略，线程池将会悄悄地丢弃这个任务而不被调用者知道；

3）CallerRunsPolicy（调用者运行）：该策略既不会抛弃任务也不会抛出异常，而是将这个任务退回给调用者，从而下降新任务的流量；

4）DiscardOldestPolicy（抛弃最旧的）：该策略将会抛弃下一个即将轮到执行的任务，那么“抛弃最旧”的将致使抛弃优先级最高的任务，所以最好不要把“抛弃最旧的”饱和策略和优先级队列放在一块儿使用； 这里，代码实现咱们将只展现**CallerRunsPolicy（调用者运行）**策略：

public static class CallerRunsPolicy implements RejectedExecutionHandler {
    /** * Creates a {@code CallerRunsPolicy}. */
    public CallerRunsPolicy() { }
     
    // 策略实现
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        if (!e.isShutdown()) {
            r.run();
        }
    }
}

复制代码

固然开发者也能够选择，根据业务需求，定义本身的饱和策略。

五、线程池销毁

ThreadPoolExecutor提供了两种方法销毁线程池，分别是shutdown()和shutdownNow()

shutdown()方法仅仅是把线程池的状态置为SHUTDOWN，而且拒绝以后尝试提交进来的全部请求，可是已经在任务队列里的任务会仍然会正常消费。

而shutdownNow()方法的表现显得更加简单粗暴，它会强行关闭ExecutorService，也会尝试取消正在执行的任务，而且返回全部已经提交但还没有开始的任务，开发者能够将这些任务写入日志保存起来以便以后进行处理，另外尝试取消正在执行的任务仅仅是尝试对执行线程进行中断，具体的线程响应中断策略须要用户本身编写。代码实现以下：

public List<Runnable> shutdownNow() {
    List<Runnable> tasks;
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        checkShutdownAccess();
        advanceRunState(STOP);
        interruptWorkers();
        tasks = drainQueue();
    } finally {
        mainLock.unlock();
    }
    tryTerminate();
    return tasks;
}
复制代码

当心踩坑：线程池经验谈

不要使用ThreadLocal

不要在ThreadPoolExecutor线程池中使用ThreadLocal，由于在ThreadPoolExecutor中，线程是复用的，所以在这里使用ThreadLocal会被多个task共享，所以可能会带来脏数据污染。须要当心使用

合理设置corePoolSize的值

以一段代码为例：

// 10个线程,由于任务多,这里用LinkedBlockingQueue
private static final LinkedBlockingQueue<Runnable> queue = new LinkedBlockingQueue<Runnable>();
private static final ExecutorService service = new ThreadPoolExecutor(0, 10,
                                                60L, TimeUnit.SECONDS, queue
);
复制代码

代码中的corePoolSize=0，也就是核心线程数是1，若是任务数多于10个，那么会先建立maximumPoolSize个线程执行，其他的任务加入 queue 中等待执行。

而在ThreadPoolExecutor的实现中，当workQueue已满，且maximumPoolSize>corePoolSize时，新提交任务会建立新线程执行任务。

所以，queue 是不会满的，那么永远不会有maximumPoolSize个线程被建立，也就是说咱们的任务一直仍是一个线程在跑，没法达到能够同时使用多个线程的预期。

线程中断

虽然ThreadPoolExecutor提供了shutdownNow()方法，在调用该方法后会尝试中断全部线程，可是该中断并不能保证线程必定会就此终止，所以，须要开发者实现线程中断的策略。关于这部分的内容，在Doug Lea的《Java Concurrency In Practice》的7.1.2节已经进行了完整的讨论，笔者在这里就再也不赘述了。

finalize函数

尤为须要注意的是，ThreadPoolExecutor有一个finalize函数，具体实现以下：

protected void finalize() {
    shutdown();
}
复制代码

在该方法中调用了shutdown()函数，所以，若是你并非真正但愿中止线程池的执行，那么就不要让线程池离开你代码的做用范围。

我是江溢Jonny。

相关文章
相关标签/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公众号
   欢迎关注本站公众号,获取更多信息
联系我们 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程