深刻理解Java线程池

时间  2021-01-02
标签 java web 算法 数据库 数组 安全 微信 并发 app 异步 栏目 Java 繁體版
原文   原文链接

编者注：Java中的线程池是运用场景最多的并发组件，几乎全部须要异步或并发执行任务的程序均可以使用线程池。java

在开发过程当中，合理地使用线程池可以带来至少如下几个好处。web

  • 下降资源消耗：经过重复利用已建立的线程下降线程建立和销毁形成的消耗。算法

  • 提升响应速度：当任务到达时，任务能够不须要等到线程建立就能当即执行。数据库

  • 提升线程的可管理性：线程是稀缺资源，若是无限制地建立，不只会消耗系统资源，还会下降系统的稳定性，使用线程池能够进行统一分配、调优和监控。可是，要作到合理利用线程池，必须了解其实现原理。数组

  • 代码解耦：好比生产者消费者模式。安全

线程池实现原理

当提交一个新任务到线程池时，线程池的处理流程以下：微信

  1. 若是当前运行的线程少于corePoolSize，则建立新线程来执行任务（注意，执行这一步骤须要获取全局锁）。并发

  2. 若是运行的线程等于或多于corePoolSize，则将任务加入BlockingQueue。app

  3. 若是没法将任务加入BlockingQueue（队列已满），则建立新的线程来处理任务（注意，执行这一步骤也须要获取全局锁）。异步

  4. 若是建立新线程将使当前运行的线程数超出maximumPoolSize，该任务将被拒绝，并调用相应的拒绝策略来处理（RejectedExecutionHandler.rejectedExecution()方法，线程池默认的饱和策略是AbortPolicy，也就是抛异常）。

ThreadPoolExecutor采起上述步骤的整体设计思路，是为了在执行execute()方法时，尽量地避免获取全局锁（那将会是一个严重的可伸缩瓶颈）。在ThreadPoolExecutor完成预热以后（当前运行的线程数大于等于corePoolSize），几乎全部的execute()方法调用都是执行步骤2，而步骤2不须要获取全局锁。

线程池任务 拒绝策略包括 抛异常、直接丢弃、丢弃队列中最老的任务、将任务分发给调用线程处理。

线程池的建立：经过ThreadPoolExecutor来建立一个线程池。

new ThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime, timeUnit, runnableTaskQueue, handler);

建立一个线程池时须要输入如下几个参数：

  • corePoolSize（线程池的基本大小）：当提交一个任务到线程池时，线程池会建立一个线程来执行任务，即便其余空闲的基本线程可以执行新任务也会建立线程，等到线程池的线程数等于线程池基本大小时就再也不建立。若是调用了线程池的prestartAllCoreThreads()方法，线程池会提早建立并启动全部基本线程。

  • maximumPoolSize（线程池最大数量）：线程池容许建立的最大线程数。若是队列满了，而且已建立的线程数小于最大线程数，则线程池会再建立新的线程执行任务。值得注意的是，若是使用了无界的任务队列这个参数就没什么效果。

  • keepAliveTime（线程活动保持时间）：线程池的工做线程空闲后，保持存活的时间。因此，若是任务不少，而且每一个任务执行的时间比较短，能够调大时间，提升线程的利用率。

  • TimeUnit（线程活动保持时间的单位）：可选的单位有天（DAYS）、小时（HOURS）、分钟（MINUTES）、毫秒（MILLISECONDS）、微秒（MICROSECONDS，千分之一毫秒）和纳秒（NANOSECONDS，千分之一微秒）。

  • runnableTaskQueue（任务队列）：用于保存等待执行的任务的阻塞队列。能够选择如下几个阻塞队列。

  • - ArrayBlockingQueue：是一个基于数组结构的有界阻塞队列，此队列按FIFO（先进先出）原则对元素进行排序。

  • LinkedBlockingQueue：一个基于链表结构的阻塞队列，此队列按FIFO排序元素，吞吐量一般要高于ArrayBlockingQueue。静态工厂方法Executors.newFixedThreadPool()使用了这个队列。

  • SynchronousQueue：一个不存储元素的阻塞队列。每一个插入操做必须等到另外一个线程调用移除操做，不然插入操做一直处于阻塞状态，吞吐量一般要高于LinkedBlockingQueue，静态工厂方法Executors.newCachedThreadPool使用了这个队列。

  • PriorityBlockingQueue：一个具备优先级的无界阻塞队列。

线程的状态

在HotSpot VM线程模型中，Java线程被一对一映射到本地系统线程，Java线程启动时会建立一个本地系统线程；当Java线程终止时，这个本地系统线程也会被回收。操做系统调度全部线程并把它们分配给可用的CPU。

thread运行周期中，有如下6种状态，在 java.lang.Thread.State 中有详细定义和说明：

// Thread类
public enum State {
    /**
     * 刚建立还没有运行
     */
    NEW,

    /**
     * 可运行状态，该状态表示正在JVM中处于运行状态，不过有多是在等待其余资源，好比CPU时间片，IO等待
     */
    RUNNABLE,

    /**
     * 阻塞状态表示等待monitor锁（阻塞在等待monitor锁或者在调用Object.wait方法后从新进入synchronized块时阻塞）
     */
    BLOCKED,

    /**
     * 等待状态，发生在调用Object.wait、Thread.join (with no timeout)、LockSupport.park
     * 表示当前线程在等待另外一个线程执行某种动做，好比Object.notify()、Object.notifyAll()，Thread.join表示等待线程执行完成
     */
    WAITING,

    /**
     * 超时等待，发生在调用Thread.sleep、Object.wait、Thread.join (in timeout)、LockSupport.parkNanos、LockSupport.parkUntil
     */
    TIMED_WAITING,

    /**
     *线程已执行完成，终止状态
     */
    TERMINATED;
}

线程池操做

向线程池提交任务，可使用两个方法向线程池提交任务，分别为execute()和submit()方法。execute()方法用于提交不须要返回值的任务，因此没法判断任务是否被线程池执行成功。经过如下代码可知execute()方法输入的任务是一个Runnable类的实例。

threadsPool.execute(new Runnable() {
    @Override
    public void run() {
        // TODO Auto-generated method stub
    }
});

submit()方法用于提交须要返回值的任务。线程池会返回一个future类型的对象，经过这个future对象能够判断任务是否执行成功，经过future的get()方法来获取返回值，future的get()方法会阻塞当前线程直到任务完成，而使用get（long timeout，TimeUnit unit）方法则会阻塞当前线程一段时间后当即返回，这时候有可能任务尚未执行完。

Future<Object> future = executor.submit(harReturnValuetask);
try {
    Object s = future.get();
} catch (InterruptedException e) {
    // 处理中断异常
} catch (ExecutionException e) {
    // 处理没法执行任务异常
} finally {
    // 关闭线程池
    executor.shutdown();
}

合理配置线程池

要想合理配置线程池，必须先分析任务的特色，能够从如下几个角度分析：

  • 任务的性质：CPU密集型任务、IO密集型任务和混合型任务。

  • 任务的优先级：高、中和低。

  • 任务的执行时间：长、中和短。

  • 任务的依赖性：是否依赖其余系统资源，如数据库链接。

性质不一样的任务能够用不一样规模的线程池分开处理。CPU密集型任务应配置尽量少的线程，如配置Ncpu+1个线程的线程池。因为IO密集型任务线程并非一直在执行任务，则应配置多一点线程，如2*Ncpu。混合型的任务，若是能够拆分，将其拆分红一个CPU密集型任务和一个IO密集型任务，只要这两个任务执行的时间相差不是太大，那么分解后执行的吞吐量将高于串行执行的吞吐量。若是这两个任务执行时间相差太大，则不必进行分解。能够经过Runtime.getRuntime().availableProcessors()方法得到当前设备的CPU个数。

优先级不一样的任务可使用优先级队列PriorityBlockingQueue来处理。它可让优先级高的任务先执行。执行时间不一样的任务能够交给不一样规模的线程池来处理，或者可使用优先级队列，让执行时间短的任务先执行。依赖数据库链接池的任务，由于线程提交SQL后须要等待数据库返回结果，等待的时间越长，则CPU空闲时间就越长，那么线程数应该设置得越大，这样才能更好地利用CPU。

线程池中线程数量未达到coreSize时，这些线程处于什么状态？

这些线程处于RUNNING或者WAITING，RUNNING表示线程处于运行当中，WAITING表示线程阻塞等待在阻塞队列上。当一个task submit给线程池时，若是当前线程池线程数量还未达到coreSize时，会建立线程执行task，不然将任务提交给阻塞队列，而后触发线程执行。（从submit内部调用的代码也能够看出来）

ScheduledThreadPoolExecutor

ScheduledThreadPoolExecutor继承自ThreadPoolExecutor。它主要用来在给定的延迟以后运行任务，或者按期执行任务。ScheduledThreadPoolExecutor的功能与Timer相似，但ScheduledThreadPoolExecutor功能更强大、更灵活。Timer对应的是单个后台线程，而ScheduledThreadPoolExecutor能够在构造函数中指定多个对应的后台线程数。

ScheduledThreadPoolExecutor继承自ThreadPoolExecutor，ScheduledThreadPoolExecutor和ThreadPoolExecutor的区别是，ThreadPoolExecutor获取任务时是从BlockingQueue中获取的，而ScheduledThreadPoolExecutor是从DelayedWorkQueue中获取的（注意，DelayedWorkQueue是BlockingQueue的实现类）。

ScheduledThreadPoolExecutor把待调度的任务（ScheduledFutureTask）放到一个DelayQueue中，其中ScheduledFutureTask主要包含3个成员变量：

  1. sequenceNumber：任务被添加到ScheduledThreadPoolExecutor中的序号；

  2. time：任务将要被执行的具体时间；

  3. period：任务执行的间隔周期。

ScheduledThreadPoolExecutor会把待执行的任务放到工做队列DelayQueue中，DelayQueue封装了一个PriorityQueue，PriorityQueue会对队列中的ScheduledFutureTask进行排序，具体的排序比较算法实现以下：

ScheduledFutureTask在DelayQueue中被保存在一个PriorityQueue（基于数组实现的优先队列，相似于堆排序中的优先队列）中，在往数组中添加/移除元素时，会调用siftDown/siftUp来进行元素的重排序，保证元素的优先级顺序。

static class DelayedWorkQueue extends AbstractQueue<Runnable>
    implements BlockingQueue<Runnable> {

    private static final int INITIAL_CAPACITY = 16;
    private RunnableScheduledFuture<?>[] queue =
        new RunnableScheduledFuture<?>[INITIAL_CAPACITY];
    private final ReentrantLock lock = new ReentrantLock();
    private int size = 0;

    private Thread leader = null;
    private final Condition available = lock.newCondition();
}

从DelayQueue获取任务的主要逻辑就在take()方法中，首选获取lock，而后获取queue[0]，若是为null则await等待任务的来临，若是非null查看任务是否到期，是的话就执行该任务，不然再次await等待。这里有一个leader变量，用来表示当前进行awaitNanos等待的线程，若是leader非null，表示已经有其余线程在进行awaitNanos等待，本身await等待，不然本身进行awaitNanos等待。

// DelayedWorkQueue
public RunnableScheduledFuture<?> take() throws InterruptedException {
    final ReentrantLock lock = this.lock;
    lock.lockInterruptibly();
    try {
        for (;;) {
            RunnableScheduledFuture<?> first = queue[0];
            if (first == null)
                available.await();
            else {
                long delay = first.getDelay(NANOSECONDS);
                if (delay <= 0)
                    return finishPoll(first);
                first = null; // don't retain ref while waiting
                if (leader != null)
                    available.await();
                else {
                    Thread thisThread = Thread.currentThread();
                    leader = thisThread;
                    try {
                        available.awaitNanos(delay);
                    } finally {
                        if (leader == thisThread)
                            leader = null;
                    }
                }
            }
        }
    } finally {
        if (leader == null && queue[0] != null)
            available.signal();
        lock.unlock();
    }
}

获取到任务以后，就会执行task的run()方法了，即ScheduledFutureTask.run()：

public void run() {
    boolean periodic = isPeriodic();
    if (!canRunInCurrentRunState(periodic))
        cancel(false);
    else if (!periodic)
        ScheduledFutureTask.super.run();
    else if (ScheduledFutureTask.super.runAndReset()) {
        setNextRunTime();
        reExecutePeriodic(outerTask);
    }
}


 推荐阅读 


欢迎小伙伴 关注【TopCoder】 阅读更多精彩好文。

本文分享自微信公众号 - TopCoder（gh_12e4a74a5c9c）。
若有侵权，请联系 support@oschina.cn 删除。
本文参与“OSC源创计划”，欢迎正在阅读的你也加入，一块儿分享。

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