Java并发编程之线程池创建线程任务的过程

àì夳堔傛蜴生んèń 2022-05-25 07:36 144阅读 0赞

毕业后又需要找工作，因为请假时间太长，不得不辞职了，离开我实习了一年的地方。希望自己一切顺利。

[Java并发编程之线程池的使用（一）点击打开链接][Java]

[Java并发编程之线程池的使用（二）点击打开链接][Java 1]

前面自己记录创建线程池的方法：一种是通过ThreadPoolExecute来创建一个线程池，这也是阿里鼓励的一种方式，让使用的人能明白如何创建线程池并且保证不浪费资源，另外一种是使用Executors执行器类下静态方法。

那么线程是如何创建到到线程池中：

首先通过execute()方法进入到线程池中看一下源码：

public void execute(Runnable command) {
            if (command == null)//如果Runable为空，抛出空指针异常
                throw new NullPointerException();   
            int c = ctl.get();//高29位表示线程数，低三位表示状态：running shutdown stop tidying terminated
            if (workerCountOf(c) < corePoolSize) {//获取当前正在运行线程数是否小于核心线程池，是则新创建一个线程执行任务，否则将任务放到任务队列中
                if (addWorker(command, true))
                    return;
                c = ctl.get();
            }
            if (isRunning(c) && workQueue.offer(command)) {//workerCountOf(c) >= corePoolSize，所以此时将线程放到任务队列中
                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);
        }

当线程池中的线程数小于corePoolSize 时，新提交的任务直接新建一个线程执行任务（不管是否有空闲线程）   
当线程池中的线程数等于corePoolSize 时，新提交的任务将会进入阻塞队列（workQueue）中，等待线程的调度   
当阻塞队列满了以后，如果corePoolSize < maximumPoolSize ,则新提交的任务会新建线程执行任务，直至线程数达到maximumPoolSize   
当线程数达到maximumPoolSize 时，新提交的任务会由(拒绝策略)管理

addWorker方法（好长啊。。。）

private boolean addWorker(Runnable firstTask, boolean core) {
            retry:
            for (;;) {//死循环
                int c = ctl.get();
                int rs = runStateOf(c);
    //如果当前线程池的状态是SHUTDOWN，STOP，TIDYING，TERMINATED并且为SHUTDOWN状态时任务队列为空，那么就返回false            
                if (rs >= SHUTDOWN &&
                    ! (rs == SHUTDOWN &&
                       firstTask == null &&
                       ! workQueue.isEmpty()))
                    return false;//原因：如果调用了shutdown方法，此时的线程池还会继续工作并且会在任务队列中的所有任务执行完成后才会结束线程池
               for (;;) {
                    int wc = workerCountOf(c);//core是在execute方法中传的参数，true表示 核心线程，false表示最大线程 
                    if (wc >= CAPACITY || //CAPACITY 容量 = (1 << COUNT_BITS) - 1;也就是1*2^29-1
                        wc >= (core ? corePoolSize : maximumPoolSize))
                        return false;
                    if (compareAndIncrementWorkerCount(c))//如果增加任务数量成功那么退出这个循环执行下面的代码，否则继续 
                        break retry;
                    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;//重入锁（ReentrantLock）是一种递归无阻塞的同步机制
                    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.add(w); //向正在执行的任务队列(workers)中添加work 
                            int s = workers.size();
                            if (s > largestPoolSize)
                                largestPoolSize = s;
                            workerAdded = true;
                        }
                    } finally {
                        mainLock.unlock();//finally块释放内置锁   
                    }
                    if (workerAdded) {//如果任务添加成功那么开始执行任务
                        t.start();
                        workerStarted = true;
                    }
                }
            } finally {
                if (! workerStarted)
                    addWorkerFailed(w);
            }
            return workerStarted;
        }

`这里我的个人理解是：firstTask进入addWork后如果走过死循环的部分（增加任务数量成功），就会向Worker中加入任务`

Worker(Runnable firstTask) {
                setState(-1); // inhibit interrupts until runWorker
                this.firstTask = firstTask;//提交的任务 在 runWorker中会用到
                this.thread = getThreadFactory().newThread(this);
            }

所以这里worker正在执行的任务队列,而workqueue是等待执行的阻塞队列。而这个Worker重新了run方法

public void run() {
                runWorker(this);
            }

final void runWorker(Worker w) {
            Thread wt = Thread.currentThread();
            Runnable task = w.firstTask;  //得到worker对象中我们提交的任务
            w.firstTask = null;
            w.unlock(); // allow interrupts
            boolean completedAbruptly = true;
            try {
                while (task != null || (task = getTask()) != null) {//while 循环 如果当前任务为空  那么就从getTask中获得任务
                    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())//测试当前线程是否已经中断 这里可以看一下这三个方法interrupted(清除中断状态) isInterrupted(没有清除状态的功能) interrupt 方法
                        wt.interrupt();//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;//执行完 task设置为null 那也就是第一次传进来task执行完之后，设置为null然后通过gettask()任务去获取task
                        w.completedTasks++;
                        w.unlock();
                    }
                }
                completedAbruptly = false;
            } finally {
                processWorkerExit(w, completedAbruptly);
            }
        }

执行完 task设置为null 那也就是第一次传进来task执行完之后，设置为null然后通过gettask()任务去获取task，那么看一下

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.
                if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
                    decrementWorkerCount();
                    return null;
                }
    
                int wc = workerCountOf(c);
    
                // Are workers subject to culling?
                boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
    
                if ((wc > maximumPoolSize || (timed && timedOut))
                    && (wc > 1 || workQueue.isEmpty())) {
                    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;
                }
            }
        }

Runnable r = timed ? workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :  workQueue.take();

这里从阻塞任务队列中取任务，如果设置了allowCoreThreadTimeOut(true) 或者当前运行的任务数大于设置的核心线程数，那么timed =true 。此时将使用workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS)从任务队列中取任务，而如果没有设置，那么使用workQueue.take();取任务，对于阻塞队列，poll(long timeout, TimeUnit unit) 将会在规定的时间内去任务，如果没取到就返回null此时  if (r != null)  return r。也就不会反悔r,take()会一直阻塞，等待任务的添加。 我们的线程池能够一直等待任务的执行而不被销毁了，其实也就是进入了阻塞状态而已。

[Java]: https://blog.csdn.net/qq442270636/article/details/80056801
[Java 1]: https://blog.csdn.net/qq442270636/article/details/80101671