临沂大众网:线程池源码探讨

admin 5个月前 (04-17) 科技 37 0

1.线程池简介

使用线程池,一样平常会使用JDK提供的几种封装类型,即:newFixedThreadPoolnewSingleThreadExecutornewCachedThreadPool等,这些线程池的界说在Executors类中,来看看相关的源码:

    public static ExecutorService newCachedThreadPool(ThreadFactory threadFactory) {
        return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                      60L, TimeUnit.SECONDS,
                                      new SynchronousQueue<Runnable>(),
                                      threadFactory);
    }

    public static ExecutorService newSingleThreadExecutor() {
        return new FinalizableDelegatedExecutorService
            (new ThreadPoolExecutor(1, 1,
                                    0L, TimeUnit.MILLISECONDS,
                                    new LinkedBlockingQueue<Runnable>()));
    }

    public static ExecutorService newFixedThreadPool(int nThreads) {
        return new ThreadPoolExecutor(nThreads, nThreads,
                                      0L, TimeUnit.MILLISECONDS,
                                      new LinkedBlockingQueue<Runnable>());
    }

这些方式内部都使用了ThreadPoolExecutor的组织方式,区别只是传入的参数差别。ThreadPoolExecutor有四个重载的组织方式,最终挪用的是由7个参数的组织器,其源码如下:

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

参数注释:

  • corePoolSize:焦点池巨细,默认情形下,线程池启动之后,并不会立刻建立线程,而是要等到义务到来之后,才建立线程去执行义务(除非设置了allowCoreThreadTimeOut参数,该参数会在线程池启动之后立马建立焦点池数目的线程)。随着义务的不停增添,现有线程无法知足要求,就会不停的建立新线程,直到线程数到达corePoolSize的值,后续新来的义务会放入壅闭行列;
  • maximumPoolSize: 最大池巨细,当义务太多,壅闭行列满了之后,若是线程数目还没有跨越该参数的值,就会继续建立新线程,直到线程数到达该参数划定的值,后续再来的义务会使用拒绝计谋举行处置;
  • keepAliveTime: 若是线程数跨越corePoolSize的值,那么多余的线程在空闲keepAliveTime时间后会被销毁;
  • unit: keepAliveTime参数的单元;
  • workQueue: 壅闭行列;
  • threadFactory: 线程工厂,建立线程时需要使用到该工厂;
  • handler: 拒绝计谋。

2.焦点字段

ThreadPoolExecutor的焦点字段如下:

    //ctl低29位示意线程的数目,高3位示意线程池状态,因此当前线程池允许的最大线程数目是2^29-1
    private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
    //固定值29
    private static final int COUNT_BITS = Integer.SIZE - 3;
    //线程最大容量
    private static final int CAPACITY   = (1 << COUNT_BITS) - 1;

    // runState is stored in the high-order bits
    //线程池的运行时状态,负数示意正在运行,正数示意终止情形
    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;

3.线程池状态

线程池的状态有5种,状态之间的转换关系如下图:

初始情形下,线程池建立完毕后会处于RUNNING状态,可以正常的接受新义务;当挪用shutdown()时,线程池酿成SHUTDOWN状态,此时无法接受新义务,然则会继续执行壅闭行列中的义务;当挪用shutdownNow()时,线程由RUNNING状态酿成STOP状态,此时不能接受新义务,而且会中止正在执行的义务;当线程池中的线程数削减为0时,就会转成TIDYING状态;在TIDYING状态会自动挪用terminated()使线程池转为TERMINATED状态。

  • shutdown()
    shutdown()方式的逻辑分别由5个差别的方式来实现,这里将这些方式整理在一起,如下:
    public void shutdown() {
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            //检查security manager是否允许挪用方执行此方式
            checkShutdownAccess();
            //将线程池状态更新为SHUTDOWN
            advanceRunState(SHUTDOWN);
            //中止空闲线程
            interruptIdleWorkers();
            //这是一个空实现,允许子类举行重写
            onShutdown(); // hook for ScheduledThreadPoolExecutor
        } finally {
            mainLock.unlock();
        }
        tryTerminate();
    }

    private void advanceRunState(int targetState) {
        for (;;) {
            int c = ctl.get();
            //若是线程池已经处在targetState及之后的状态则直接竣事循环,否则使用CAS操作将线程池状态更新为targetState
            if (runStateAtLeast(c, targetState) ||
                ctl.compareAndSet(c, ctlOf(targetState, workerCountOf(c))))
                break;
        }
    }

    private void interruptIdleWorkers() {
        interruptIdleWorkers(false);
    }

    //onlyOne示意是否只终止一个空闲线程
    private void interruptIdleWorkers(boolean onlyOne) {
        final ReentrantLock mainLock = this.mainLock;
        //加可重入锁
        mainLock.lock();
        try {
            for (Worker w : workers) {
                Thread t = w.thread;
                //若是线程没有被中止,则实验获取锁,获取乐成后将线程中止
                if (!t.isInterrupted() && w.tryLock()) {
                    try {
                        t.interrupt();
                    } catch (SecurityException ignore) {
                    } finally {
                        //释放锁
                        w.unlock();
                    }
                }
                if (onlyOne)
                    break;
            }
        } finally {
            mainLock.unlock();
        }
    }

    final void tryTerminate() {
        //自旋
        for (;;) {
            int c = ctl.get();
            //线程池还在运行,或者已经是TIDYING或TERMINATED状态,或者壅闭行列不为空,这几种情形不再继续执行
            if (isRunning(c) ||
                runStateAtLeast(c, TIDYING) ||
                (runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty()))
                return;
            //线程数不为0时,终止一个空闲线程
            if (workerCountOf(c) != 0) { // Eligible to terminate
                interruptIdleWorkers(ONLY_ONE);
                return;
            }

            final ReentrantLock mainLock = this.mainLock;
            mainLock.lock();
            try {
                //将线程池设置为DIDYING状态
                if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {
                    //设置乐成后,执行terminated()方式
                    try {
                        //这也是一个空实现,子类可以根据需要举行重写
                        terminated();
                    } finally {
                        //将线程池设置为TERMINATED状态
                        ctl.set(ctlOf(TERMINATED, 0));
                        termination.signalAll();
                    }
                    return;
                }
            } finally {
                mainLock.unlock();
            }
            // else retry on failed CAS
        }
    }
  • shutdownNow()
    public List<Runnable> shutdownNow() {
        List<Runnable> tasks;
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            //检查security manager是否允许挪用方执行此方式
            checkShutdownAccess();
            //将线程池状态更新为STOP
            advanceRunState(STOP);
            //与shutdown的区别是,这里会中止所有线程,而不仅仅是空闲线程
            interruptWorkers();
            //将义务从workQueue中移除,转移到一个ArrayList中,此操作后,workQueue为空,已有的义务无法继续执行
            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();
        }
    }

4.执行义务

线程池通过execute()方式执行义务,其源码如下:

    public void execute(Runnable command) {
        if (command == null)
            throw new NullPointerException();

        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);
            //若是活跃线程为0,则建立一个非焦点线程,并将firstTask设置为null
            else if (workerCountOf(recheck) == 0)
                addWorker(null, false);
        }
        //若是添加非焦点线程失败,则执行拒绝计谋
        else if (!addWorker(command, false))
            reject(command);
    }
    
    //获取活跃的线程数
    private static int workerCountOf(int c)  { return c & CAPACITY; }
    //获取线程池运行状态
    private static int runStateOf(int c)     { return c & ~CAPACITY; }

来看看addWorker()方式的实现:

    //core示意要建立的是否是焦点线程,true示意建立焦点线程,false示意建立非焦点线程
    private boolean addWorker(Runnable firstTask, boolean core) {
        retry:
        for (;;) {
            int c = ctl.get();
            //获取线程池状态
            int rs = runStateOf(c);

            // Check if queue empty only if necessary.
            //rs >= SHUTDOWN,示意线程池已终止
            //rs>=SHUTDOWN,说明已经挪用了shutdown()或者shutdownNow()方式,在此条件知足的情形下,第二项条件等同于
            //rs!=SHUTDOWN || firstTask != null || workQueue.isEmpty(),知足这三个条件的任何一个都不会再添加新义务
            //rs!=SHUTDOWN,说明是STOP、TIDYING、TERMINATE这三种
            if (rs >= SHUTDOWN &&
                ! (rs == SHUTDOWN &&
                   firstTask == null &&
                   ! workQueue.isEmpty()))
                return false;

            //执行到这里说明:
            //① rs<SHUTDOWN,即线程池是运行状态
            //② rs=SHUTDOWN,farstTask=null, 而且壅闭行列不为空
            for (;;) {
                int wc = workerCountOf(c);
                //有三种情形会返回false:1)线程数到达最大值;2)当前建立焦点线程,然则线程数已经到达焦点池巨细;
                //3)当前建立非焦点线程,而且线程数到达最大池巨细
                if (wc >= CAPACITY ||
                    wc >= (core ? corePoolSize : maximumPoolSize))
                    return false;
                //若是使用CAS操作乐成将ctl的值加1,则跳出最外层循环
                if (compareAndIncrementWorkerCount(c))
                    break retry;
                //走到这里说明无法使用CAS更新ctl的值,说明此时发生了多线程竞争,需要重新查看线程池的状态
                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;
        Worker w = null;
        try {
            w = new Worker(firstTask);
            final Thread t = w.thread;
            if (t != null) {
                final ReentrantLock mainLock = this.mainLock;
                //加重入锁
                mainLock.lock();
                try {
                    // Recheck while holding lock.
                    // Back out on ThreadFactory failure or if
                    // shut down before lock acquired.
                    int rs = runStateOf(ctl.get());

                    if (rs < SHUTDOWN ||
                        (rs == SHUTDOWN && firstTask == null)) {
                        if (t.isAlive()) // precheck that t is startable
                            throw new IllegalThreadStateException();
                        //workers是个HashSet类型,只在重入锁代码中被接见
                        workers.add(w);
                        //更新当前活跃线程的最大值
                        int s = workers.size();
                        if (s > largestPoolSize)
                            largestPoolSize = s;
                        workerAdded = true;
                    }
                } finally {
                    mainLock.unlock();
                }
                if (workerAdded) {
                    //启动线程,挪用Worker类的run()方式
                    t.start();
                    workerStarted = true;
                }
            }
        } finally {
            //乐成建立新线程时,才会设置workerStarted=true,这里处置没有建立新线程的情形
            if (! workerStarted)
                addWorkerFailed(w);
        }
        return workerStarted;
    }

addWorker()方式中用到了Worker类,这是ThreadPoolExecutor的内部类,对线程举行了包装,线程池建立或者启动的线程,现实都是Worker类型的实例,其源码如下(省略了无关代码):

    private final class Worker extends AbstractQueuedSynchronizer implements Runnable
    {

        /** Thread this worker is running in.  Null if factory fails. */
        final Thread thread;
        /** Initial task to run.  Possibly null. */
        Runnable firstTask;
        /** Per-thread task counter */
        volatile long completedTasks;

        Worker(Runnable firstTask) {
            setState(-1); 
            this.firstTask = firstTask;
            this.thread = getThreadFactory().newThread(this);
        }

        /** Delegates main run loop to outer runWorker  */
        public void run() {
            runWorker(this);
        }

当启动Worker线程时,会挪用runWorker()方式,每一个启动的线程都市在该方式的while循环中不停获取义务去执行,该方式源码如下:

    final void runWorker(Worker w) {
        Thread wt = Thread.currentThread();
        Runnable task = w.firstTask;
        w.firstTask = null;
        w.unlock(); // allow interrupts
        boolean completedAbruptly = true;
        try {
            //若是能够乐成拿到义务,则执行下面的代码块,若是getTask()方式返回null,当前线程就会执行退出逻辑
            while (task != null || (task = getTask()) != null) {
                //若是能将state字段设置为1,示意乐成拿到锁,就接着向下执行,否则线程会加入守候行列,不再继续执行
                //注重这里是在乐成拿到新义务之后才会加锁,连系shutdown()方式的逻辑
                w.lock();
                // If pool is stopping, ensure thread is interrupted;
                // if not, ensure thread is not interrupted.  This
                // requires a recheck in second case to deal with
                // shutdownNow race while clearing interrupt
                //若是线程池正在关闭,需要中止当前线程
                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 {
            processWorkerExit(w, completedAbruptly);
        }
    }

beforeExecute()afterExecute()protected类型,而且默认是空实现,很明显是留给子类去实现钩子逻辑。上面的代码使用getTask()从壅闭行列中取义务,其实现如下:

    private Runnable getTask() {
        boolean timedOut = false; // Did the last poll() time out?

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

            // Check if queue empty only if necessary.
            //线程池正在关闭,或者壅闭行列空了,就削减线程数,并返回null
            if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
                decrementWorkerCount();
                return null;
            }

            int wc = workerCountOf(c);

            // Are workers subject to culling?
            //在设置了allowCoreThreadTimeOut参数后,跨越给定的时间,会将空闲的焦点线程清算掉
            //或者线程数目跨越了焦点池数目,会在一定时间后清算掉多余的线程
            boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
            //1)线程数目跨越最大池数目,或者超时; 2)线程数大于1,或者壅闭行列为空; 这两个条件都建立时,就将ctl值减1
            if ((wc > maximumPoolSize || (timed && timedOut))
                && (wc > 1 || workQueue.isEmpty())) {
                if (compareAndDecrementWorkerCount(c))
                    return null;
                continue;
            }

            try {
                //若是设置了超时状态,则使用poll方式取义务,跨越keepAliveTime还没有义务到来就返回true
                //否则使用take取义务,在壅闭行列为空时会一直守候
                Runnable r = timed ?
                    workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                    workQueue.take();
                if (r != null)
                    return r;
                timedOut = true;
            } catch (InterruptedException retry) {
                //线程有可能在守候新义务的到来而壅闭,然则在守候的过程中挪用shutdownNow()关闭线程时,线程会抛出中止异常,在这里被捕捉
                timedOut = false;
            }
        }
    }

现在来整理一下runWorker()方式的思绪:每一个新建立的线程都市在runWorker()方式里通过while循环不停地从壅闭行列中获取义务,取到义务之后就执行义务的run()方式,取不到义务就会一直壅闭,或者守候一定的时间之后,空闲线程超时需要接纳,就会执行processWorkerExit()方式。

5.线程池是若何关闭的

  • shutdown()
    在先容shutdown()方式时有一个疑问,该方式只会中止空闲线程,然则非空闲的线程不会被中止,纵然该线程被壅闭,因此该方式有可能无法关闭那些一直处在守候状态的非空闲线程,这一点在使用时需要注重。在runWorker()方式中,while循环会在乐成拿到义务后才会加锁,因此那些由于壅闭行列为空拿不到义务而壅闭的线程也会被shutdown()方式中止
while (task != null || (task = getTask()) != null) {
    //若是能将state字段设置为1,示意乐成拿到锁,就接着向下执行,否则线程会加入守候行列,不再继续执行
    //注重这里是在乐成拿到新义务之后才会加锁,连系shutdown()方式的逻辑
    w.lock();
    //忽略其他代码
}
  • shutdownNow()
    shutdownNow()会中止所有的存活线程,岂论这些线程是否空闲,因此可能会导致义务在执行的过程中抛出异常,这点需要注重。

岂论是挪用哪个方式来关闭线程池,最终线程的退出是要根据getTask()方式来决议。当getTask()方式返回null,即当前壅闭行列已经没有义务时,线程会退出,而且在getTask()方式的自旋代码会首先检查线程池的状态,如下:

    if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
        decrementWorkerCount();
        return null;
    }

在挪用shutdownNow()方式关闭线程池后,rs >= STOP逻辑建立,直接返回null,而shutdown()方式会继续执行壅闭行列中的义务,直到workQueue.isEmpty()条件为真,getTask()返回null导致线程一个个竣事,岂论是哪种情形,最终线程池中的线程数目都市酿成0。

,

sunbet 申博

Sunbet 申博www.1888ss.com秉承以客为先的理念,多年运营、专业团队、专注服务、值得信赖。

皇冠体育声明:该文看法仅代表作者自己,与本平台无关。转载请注明:临沂大众网:线程池源码探讨

网友评论

  • (*)

最新评论

文章归档

站点信息

  • 文章总数:625
  • 页面总数:0
  • 分类总数:8
  • 标签总数:1148
  • 评论总数:268
  • 浏览总数:8678