资讯专栏INFORMATION COLUMN

ThreadPool实现原理

spacewander / 1750人阅读

摘要:所以,并不代表线程池就一定立即就能退出,它也可能必须要等待所有正在执行的任务都执行完成了才能退出。

本文主要分析java.util.concurrent.ThreadPoolExecutor的实现原理,首先看它的构造函数:

public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue 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:线程池中最大线程数

keepAliveTime and unit:大于最小线程数的线程空闲后存活时间

workQueue:用于存放任务的阻塞队列

threadFactory:用于创建线程的工厂类

handler:当任务队列满了且线程数达到了最大时的饱和策略

对于IO密集型任务,线程数一般设为CPU数*2,对于计算密集型任务,线程数一般设为CPU数。

当调用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);
}

其流程如图:

创建线程是通过addWorker创建内部Worker类,其中调用getThreadFactory().newThread(this)来创建执行自己的线程,之后在addWorker中start该线程,执行Worker run方法中的runWorker会不断的从任务队列中获取任务或阻塞,并且每次执行任务前会执行beforeExecute,之后会afterExecute,可以通过重写beforeExecute方法来给执行线程重命名。

线程池状态变化如图:

RUNNING: Accept new tasks and process queued tasks

SHUTDOWN: Don"t accept new tasks, but process queued tasks

STOP: Don"t accept new tasks, don"t process queued tasks, and interrupt in-progress tasks

TIDYING: All tasks have terminated, workerCount is zero, the thread transitioning to state TIDYING will run the terminated() hook method

TERMINATED: terminated() has completed

shutdownNow终止线程的方法是通过调用Thread.interrupt()方法来实现的:

 * 

If this thread is blocked in an invocation of the {@link * Object#wait() wait()}, {@link Object#wait(long) wait(long)}, or {@link * Object#wait(long, int) wait(long, int)} methods of the {@link Object} * class, or of the {@link #join()}, {@link #join(long)}, {@link * #join(long, int)}, {@link #sleep(long)}, or {@link #sleep(long, int)}, * methods of this class, then its interrupt status will be cleared and it * will receive an {@link InterruptedException}. * *

If this thread is blocked in an I/O operation upon an {@link * java.nio.channels.InterruptibleChannel InterruptibleChannel} * then the channel will be closed, the thread"s interrupt * status will be set, and the thread will receive a {@link * java.nio.channels.ClosedByInterruptException}. * *

If this thread is blocked in a {@link java.nio.channels.Selector} * then the thread"s interrupt status will be set and it will return * immediately from the selection operation, possibly with a non-zero * value, just as if the selector"s {@link * java.nio.channels.Selector#wakeup wakeup} method were invoked. * *

If none of the previous conditions hold then this thread"s interrupt * status will be set.

可以看到如果线程处于正常活动状态,那么会将该线程的中断标志设置为true,而无法中断当前的线程。所以,shutdownNow并不代表线程池就一定立即就能退出,它也可能必须要等待所有正在执行的任务都执行完成了才能退出。

文章版权归作者所有,未经允许请勿转载,若此文章存在违规行为,您可以联系管理员删除。

转载请注明本文地址:https://www.ucloud.cn/yun/77876.html

相关文章

  • Java SDK 并发包全面总结

    摘要:一和并发包中的和主要解决的是线程的互斥和同步问题,这两者的配合使用,相当于的使用。写锁与读锁之间互斥,一个线程在写时,不允许读操作。的注意事项不支持重入,即不可反复获取同一把锁。没有返回值,也就是说无法获取执行结果。 一、Lock 和 Condition Java 并发包中的 Lock 和 Condition 主要解决的是线程的互斥和同步问题,这两者的配合使用,相当于 synchron...

    luckyyulin 评论0 收藏0
  • Java多线程(3):取消正在运行的任务

    摘要:比如上一篇文章提到的线程池的方法,它可以在线程池中运行一组任务,当其中任何一个任务完成时,方法便会停止阻塞并返回,同时也会取消其他任务。 当一个任务正在运行的过程中,而我们却发现这个任务已经没有必要继续运行了,那么我们便产生了取消任务的需要。比如 上一篇文章 提到的线程池的 invokeAny 方法,它可以在线程池中运行一组任务,当其中任何一个任务完成时,invokeAny 方法便会停...

    terro 评论0 收藏0
  • Java线程池

    摘要:中的线程池是运用场景最多的并发框架。才是真正的线程池。存放任务的队列存放需要被线程池执行的线程队列。所以线程池的所有任务完成后,它最终会收缩到的大小。饱和策略一般情况下,线程池采用的是,表示无法处理新任务时抛出异常。 Java线程池 1. 简介 系统启动一个新线程的成本是比较高的,因为它涉及与操作系统的交互,这个时候使用线程池可以提升性能,尤其是需要创建大量声明周期很短暂的线程时。Ja...

    jerry 评论0 收藏0
  • 深入剖析ThreadPool的运行原理

    摘要:而且,线程池中的线程并没有睡眠,而是进入了自旋状态。普通的线程被中断会导致线程继续执行,从而方法运行完毕,线程退出。线程死亡超过时间,任务对列没有数据而返回。线程死亡保证了线程池至少留下个线程。 线程在执行任务时,正常的情况是这样的: Thread t=new Thread(new Runnable() { @Override ...

    Pines_Cheng 评论0 收藏0

发表评论

0条评论

spacewander

|高级讲师

TA的文章

阅读更多
最新活动
阅读需要支付1元查看
<