最新公告
  • 新注册用户请前往个人中心绑定邮箱以便接收相关凭证邮件!!!点击前往个人中心
  • 线程池

    内容目录

    • TreadPoolexecutor源码解析

    • 类关系图

    • Executor接口

    • ExecutorService接口

    • AbstractExecutorService

    • 成员变量

    • 构造函数

    • Worker类解析

    • Worker简介

    • 成员变量

    • 构造方法

    • execute()解析

    • addWorker()解析

    • runWorker()解析

    • getTask()解析

    • processWorkerExit()解析

    • tryTerminate()解析

    • interruptIdleWorker()解析

    • 监控线程池

    TreadPoolexecutor源码解析

    1、类关系图

    2、Executor接口

    Executor提供execute()用来启动任务

       public interface Executor {

          //用来启动任务

          void execute(Runnable command);

       }

    3、ExecutorService接口

    ExecutorService提供了一些管理线程池方法和任务执行的方法

        public interface ExecutorService extends Executor {

            //关闭线程池,队列已经存在的任务可以继续执行

            void shutdown();

            //关闭线程池,中断未执行的任务

            List<Runnable> shutdownNow();

            //判断是否关闭

            boolean isShutdown();

            //判断是否终止

            boolean isTerminated();

            //设置超时终止

            boolean awaitTermination(long timeout, TimeUnit unit)

                throws InterruptedException;

            //提交Callable任务

            <T> Future<T> submit(Callable<T> task);

            //提交Runable任务,带返回值

            <T> Future<T> submit(Runnable task, T result);

            //提交Runnable任务不带返回值

            Future<?> submit(Runnable task);

            //invokeAll()是同步的,其需要等待任务的完成,才能返回。submit()是异步的

            <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)

                throws InterruptedException;

            <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,

                                          long timeout, TimeUnit unit)

                throws InterruptedException;

            //invokeAny()取第一个任务的返回值,然后调用interrupt方法中断其它任务。

            <T> invokeAny(Collection<? extends Callable<T>> tasks)

                throws InterruptedException, ExecutionException;

            <T> invokeAny(Collection<? extends Callable<T>> tasks,

                            long timeout, TimeUnit unit)

                throws InterruptedException, ExecutionException, TimeoutException;

        }

    4、AbstractExecutorService

    AbstractExecutorService类重写某些方法。自定义了newTaskFor()用于构建RunnableFuture。

    5、成员变量

    /**

      * 线程池使用一个int变量存储线程池状态和工作线程数

      * int4个字节,32位,用高三位存储线程池状态,低29位存储工作线程数

      * 为什么使用一个变量来同时表示线程状态和线程数?就是节省空间。咨询了一下写c的朋友,他们经常这么写

      **/

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

    //COUNT_BITS=29

    private static final int COUNT_BITS = Integer.SIZE – 3;

    //理论上线程池最大线程数量CAPACITY=536870911

    private static final int CAPACITY   = (1 << COUNT_BITS) – 1;

    /**

      * 线程池状态转换

      * RUNNING -> SHUTDOWN

      * RUNNING or SHUTDOWN -> STOP

      * SHUTDOWN or STOP -> TIDYING

      * TIDYING -> TERMINATED  terminated()执行完后变为该TERMINATED

      */

    //接受新任务,可以处理阻塞队列里的任务

    private static final int RUNNING    = –1 << COUNT_BITS;

    //不接受新任务,可以处理阻塞队列里的任务。执行shutdown()会变为SHUTDOWN

    private static final int SHUTDOWN   =  0 << COUNT_BITS;

    //不接受新的任务,不处理阻塞队列里的任务,中断正在处理的任务。执行shutdownNow()会变为STOP

    private static final int STOP       =  1 << COUNT_BITS;

    //临时过渡状态,所有的任务都执行完了,当前线程池有效的线程数量为0,这个时候线程池的状态是TIDYING,执行terminated()变为TERMINATED

    private static final int TIDYING    =  2 << COUNT_BITS;

    //终止状态,terminated()调用完成后的状态

    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; }

    //初始化ctl

    private static int ctlOf(int rs, int wc) return rs | wc; }

    //用于保存等待执行的任务的阻塞队列。比如LinkedBlockQueue,SynchronousQueue等

    private final BlockingQueue<Runnable> workQueue;

    //重入锁,更新线程池核心大小、线程池最大大小等都有用到

    private final ReentrantLock mainLock = new ReentrantLock();

    //用于存储woker

    private final HashSet<Worker> workers = new HashSet<Worker>();

    //用于终止线程池

    private final Condition termination = mainLock.newCondition();

    //记录线程池中曾经出现过的最大线程数

    private int largestPoolSize;

    //完成任务数量

    private long completedTaskCount;   

    //线程工厂

    private volatile ThreadFactory threadFactory;

    /**

     *  rejectedExecutionHandler:任务拒绝策略

     *  DiscardOldestPolicy:丢弃队列里最近的一个任务,并执行当前任务

     *  AbortPolicy:抛出异常。这也是默认的策略

     *  CallerRunsPolicy:用调用者所在线程来运行任务

     *  DiscardPolicy:不处理,丢弃掉

     */

    private volatile RejectedExecutionHandler handler;

    /**

      * 线程空闲时间

      * 当线程空闲时间达到keepAliveTime,该线程会退出,直到线程数量等于corePoolSize。

      * 如果allowCoreThreadTimeout设置为true,则所有线程均会退出。

      */

    private volatile long keepAliveTime;

    //是否允许核心线程空闲超时退出,默认值为false。

    private volatile boolean allowCoreThreadTimeOut;

    /**

     * 核心线程数

     * 核心线程会一直存活,即使没有任务需要处理,当线程数小于核心线程数时。

     * 即使现有的线程空闲,线程池也会优先创建新线程来处理任务,而不是直接交给现有的线程处理。

     * 核心线程数在初始化时不会创建,只有提交任务的时候才会创建。核心线程在allowCoreThreadTimeout为true的时候超时会退出。

     */

    private volatile int corePoolSize;

     /** 最大线程数

       * 当线程数大于或者等于核心线程,且任务队列已满时,线程池会创建新的线程,直到线程数量达到maxPoolSize。

       * 如果线程数已等于maxPoolSize,且任务队列已满,则已超出线程池的处理能力,线程池会采取拒绝操作。

       */

    private volatile int maximumPoolSize;

    //默认的拒绝策略:抛出异常

    private static final RejectedExecutionHandler defaultHandler =

        new AbortPolicy();

    private static final RuntimePermission shutdownPerm =

        new RuntimePermission(“modifyThread”);

    6、构造函数

    //直接提供了一个最终调用的构造函数

    //大致逻辑就是给线程池核心参数赋值

    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;

    }

    Worker类解析

    1、Worker简介

    Woeker类是ThreadPoolExecutor一个内部类。此类继AbstractQueuedSynchronizer,

    目的在于判断线程是否空闲以及是否可以被中断。实现Runnable,在run()中调用了runWorker()。

    Worker类中firstTask用来保存传入的任务,thread是在调用构造方法时通过ThreadFactory来创建的线程,是用来处理任务的线程。注意:这个线程并不是task线程。

    2、成员变量

        /** 工作线程空间,由线程池中所设置的线程工厂创建*/

        final Thread thread;

        /** 在构造方法中传入的任务*/

        Runnable firstTask;

        /** 执行完任务的总数*/

        volatile long completedTasks;

    3、构造方法

        Worker(Runnable firstTask) {

            //state设置为-1是为了禁止在执行任务前对任务进行中断。

            setState(-1); 

            //提交的task

            this.firstTask = firstTask;

            //从线程工厂获取的线程,注意这个thread并不是用户线程

            this.thread = getThreadFactory().newThread(this);

        }

    4、execute()解析

    public void execute(Runnable command{

         //判断提交的任务是否为空

        if (command == null)

            throw new NullPointerException();

        //获取线程池状态和工作线程数量结合体(下文统称为ctl)

        int c = ctl.get();

        //判断工作线程数量是否小于核心线程数

        if (workerCountOf(c) < corePoolSize) {

            //把任务添加到worker,添加成功则返回

            if (addWorker(command, true))

                return;

            //再次获取ctl

            c = ctl.get();

        }

        //如果线程池状态是正在运行并且添加一个任务到队列成功

        if (isRunning(c) && workQueue.offer(command)) {

            int recheck = ctl.get();

            //再次校验线程池状态,如果状态不是RUNNING则需要从队列中移除任务

            if (! isRunning(recheck) && remove(command))

                 //执行拒绝策略

                reject(command);

            else if (workerCountOf(recheck) == 0)

                //进入这里说明空闲核心线程数都超时退出啦

                //因为任务已经放入队列了,所以此处不需要传入任务

                //注意事项,网上很多说这里创建一个线程不启动这是错误的。博主亲测这里创建了一个线程并且需要启动

                addWorker(nullfalse);

        }

       /*

        * 如果执行到这里有两种情况:

        * 线程池已经不是RUNNING状态

        * 线程池是RUNNING状态,workerCount>=corePoolSize并且workQueue已满

        */

        //调用addWorker(),传入false代表把线程池线程数量设置maximumPoolSize,如果添加失败则执行拒绝策略。

        else if (!addWorker(command, false))

            reject(command);

    }

    5、addWorker()解析

    /**

      *@param firstTask 表示执行的任务;

      *@param core 表示限制添加线程的数量是根据corePoolSize来判断还是maximumPoolSize来判断; 

      * 如果为true,根据corePoolSize来判断;如果为false,则根据maximumPoolSize来判断

      */

    private boolean addWorker(Runnable firstTask, boolean core) {

        retry:

        for (;;) {

            int c = ctl.get();

            //获取线程池状态

            int rs = runStateOf(c);

            /**

              * 只要满足下面任一条就直接返回false

              * 线程池状态为STOP, TYDING 或 TERMINATD 状态

              * 线程池状态为SHUTDOWN,并且firstTask != null 或者workQueue为空

              */

             //这里为什么不直接判断线程池状态呢?是因为有可能在线程池状态为RUNNING时已经把任务放入队列中,放入完成以后状态变为SHUTDOWN

            if (rs >= SHUTDOWN &&! (rs == SHUTDOWN &&

                   firstTask == null &&! workQueue.isEmpty()))

                return false;

            for (;;) {

                //得到工作的线程数量

                int wc = workerCountOf(c);

                //如果工作线程数量大于理论上线程池容量;或者工作线程数大于(corePoolSize or maximumPoolSize) 直接返回false,添加失败

                if (wc >= CAPACITY || wc >= (core ? corePoolSize : maximumPoolSize))

                    return false;

                 //cas修改工作线程数,工作线程数+1。如果修改失败需要重新执行;成功退出循环

                if (compareAndIncrementWorkerCount(c))

                    break retry;

                c = ctl.get();  

                 //如果线程池状态变化则需要重新执行

                if (runStateOf(c) != rs)

                    continue retry;

            }

        }

        //worker是否已经启动

        boolean workerStarted = false;

        //worker是否添加成功

        boolean workerAdded = false;

        Worker w = null;

        try {

            //构建worker

            w = new Worker(firstTask);

            //注意,这个thread不是firstTask,是从线程工厂造出来的

            final Thread t = w.thread;

            if (t != null) {

                //获取锁

                final ReentrantLock mainLock = this.mainLock;

                mainLock.lock();

                try {

                    //获取线程池状态

                    int rs = runStateOf(ctl.get());

                    //如果线程池状态是RUNNING或者状态是SHUTDOWN但是队列里面还有任务

                    if (rs < SHUTDOWN ||

                        (rs == SHUTDOWN && firstTask == null)) {

                        //如果t.isAlive()=true,说明是有问题的,都没有启动,t怎么会是活的呢。所以抛出异常。

                        if (t.isAlive()) 

                            throw new IllegalThreadStateException();

                        //把worker添加到set集合

                        workers.add(w);

                        //记录线程池中出现的最大线程数

                        int s = workers.size();

                        if (s > largestPoolSize)

                            largestPoolSize = s;

                        //添加成功标识

                        workerAdded = true;

                    }

                } finally {

                    //释放锁

                    mainLock.unlock();

                }

                //添加成功启动线程,启动线程是调用了runWorker()

                if (workerAdded) {

                    t.start();

                    workerStarted = true;

                }

            }

        } finally {

             //启动失败

            if (! workerStarted)

                //启动失败需要从workers中移除当前构造的woker;工作线程数减1了;执行tryTerminate()判断是否终止线程池。

                addWorkerFailed(w);

        }

        return workerStarted;

    }

    6、runWorker()解析

    final void runWorker(Worker w{

        //获取当前线程

        Thread wt = Thread.currentThread();

        //需要执行的任务

        Runnable task = w.firstTask;

        //置为null

        w.firstTask = null;

        //这里是为了把之前的state=-1设置为state=0,此时允许中断

        w.unlock(); 

        //是否异常退出循环

        boolean completedAbruptly = true;

        try {

            //如果任务不为null或者getTask()不为null

            while (task != null || (task = getTask()) != null) {

                //获取锁。这里使用锁的目的在于标识正在处理任务

                w.lock();

                 //线程池=SHUWDOWN,要保证当前线程是中断状态

                 //线程池!=SHUWDOWN,要保证当前线程不是中断状态

                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 {

             //一定要注意。执行到这里说明getTask()返回null。说明当前线程池中不需要那么多线程来执行任务了,可以把多于corePoolSize数量的工作线程干掉

            processWorkerExit(w, completedAbruptly);

        }

    }

    7、getTask()解析

    //什么情况才会执行getTask呢?说明工作线程数已经大于核心线程数才会执行getTask()。一定要记住这一点

    private Runnable getTask() {

         //表示上次从阻塞队列中取任务时是否超时

        boolean timedOut = false

        for (;;) {

            int c = ctl.get();

            //获取线程池状态

            int rs = runStateOf(c);

           //如果线程池状态rs >= SHUTDOWN并且(rs >= STOP,或者阻塞队列为空)。则将workerCount减1并返回null。

            if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {

                decrementWorkerCount();

                return null;

            }

             //获取工作线程数

            int wc = workerCountOf(c);

            //表示是否需要超时控制。allowCoreThreadTimeOut默认false;如果线程池数量超过核心线程数也是需要超时控制的

            boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;

            //校验工作线程数量和任务队列是否为空

            if ((wc > maximumPoolSize || (timed && timedOut))

                && (wc > 1 || workQueue.isEmpty())) {

                 //工作线程数量-1

                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;

            }

        }

    }

    8、processWorkerExit()解析

    private void processWorkerExit(Worker w, boolean completedAbruptly{

           //是否异常,如果异常工作线程数量-1

        if (completedAbruptly) 

            decrementWorkerCount();

        final ReentrantLock mainLock = this.mainLock;

        mainLock.lock();

        try { 

            //增加线程池完成任务数量

            completedTaskCount += w.completedTasks;

            //从worker中移除完成任务

            workers.remove(w);

        } finally {

            mainLock.unlock();

        }

        //根据线程池状态进行判断是否结束线程池

        tryTerminate();

        int c = ctl.get();

        if (runStateLessThan(c, STOP)) {

            if (!completedAbruptly) {

                int min = allowCoreThreadTimeOut ? 0 : corePoolSize;

                if (min == 0 && ! workQueue.isEmpty())

                    min = 1;

                if (workerCountOf(c) >= min)

                    return// replacement not needed

            }

            addWorker(nullfalse);

        }

    }

    9、tryTerminate()解析

     //根据线程池状态判断是否结束线程池

    final void tryTerminate() {

        for (;;) {

            int c = ctl.get();

             //RUNNING不能结束线程池

             //线程池状态是TIDYING或TERMINATED说明线程池已经处于正在终止的路上

             //状态为SHUTDOWN,但是任务队列不为空不能结束线程池

            if (isRunning(c) ||

                runStateAtLeast(c, TIDYING) ||

                (runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty()))

                return;

             //工作线程数量不等于0,中断一个空闲的工作线程,并返回

            if (workerCountOf(c) != 0) { 

                interruptIdleWorkers(ONLY_ONE);

                return;

            }

            final ReentrantLock mainLock = this.mainLock;

            mainLock.lock();

            try {

             // 设置线程池状态为TIDYING,如果设置成功,则调用terminated方法

                if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {

                    try { 

                        //子类实现

                        terminated();

                    } finally {

                       // 设置状态为TERMINATED

                        ctl.set(ctlOf(TERMINATED, 0));

                        termination.signalAll();

                    }

                    return;

                }

            } finally {

                mainLock.unlock();

            }

        }

    }

    10、interruptIdleWorker()解析

    //该方法用于给空闲工作线程一个中断标识

    private void interruptIdleWorkers(boolean onlyOne) {

        final ReentrantLock mainLock = this.mainLock;

        mainLock.lock();

        try {

            //遍历worker,根据onlyOne判断,如果为ture只中断一个空闲线程

            for (Worker w : workers) {

                Thread t = w.thread;

                //线程没有被中断并且线程是空闲状态tryLock()判断是否空闲

                if (!t.isInterrupted() && w.tryLock()) {

                    try {

                        t.interrupt();

                    } catch (SecurityException ignore) {

                    } finally {

                        w.unlock();

                    }

                }

                if (onlyOne)

                    break;

            }

        } finally {

            mainLock.unlock();

        }

    }

    监控线程池

    通过线程池提供的参数进行监控

    • getTaskCount:线程池任务总数。

    • getCompletedTaskCount:线程池已完成的任务数量,小于等于completedTaskCount。

    • getPoolSize:线程池当前的核心线程数量。

    • getLargestPoolSize:线程池曾经创建过的最大线程数量。

    • getActiveCount:当前线程池中正在执行任务的线程数量。

    本站所有文章均由网友分享,仅用于参考学习用,请勿直接转载,如有侵权,请联系网站客服删除相关文章。若由于商用引起版权纠纷,一切责任均由使用者承担
    极客文库 » 线程池

    常见问题FAQ

    如果资源链接失效了怎么办?
    本站用户分享的所有资源都有自动备份机制,如果资源链接失效,请联系本站客服QQ:2580505920更新资源地址。
    如果用户分享的资源与描述不符怎么办?
    可以联系客服QQ:2580505920,如果要求合理可以安排退款或者退赞助积分。
    如何分享个人资源获取赞助积分或其他奖励?
    本站用户可以分享自己的资源,但是必须保证资源没有侵权行为。点击个人中心,根据操作填写并上传即可。资源所获收益完全归属上传者,每周可申请提现一次。
    如果您发现了本资源有侵权行为怎么办?
    及时联系客服QQ:2580505920,核实予以删除。

    参与讨论

    • 211会员总数(位)
    • 3737资源总数(个)
    • 0本周发布(个)
    • 0 今日发布(个)
    • 862稳定运行(天)

    欢迎加入「极客文库」,成为原创作者从这里开始!

    立即加入 了解更多
    成为赞助用户享有更多特权立即升级