ThreadPoolExecutor是如何做到线程重用的

痛定思痛。 2022-09-30 12:52 174阅读 0赞

# 前言： #

> 看关于ThreadPoolExecutor参数时，看到了keepaliveTime这个参数，这个参数的意思是：“当线程数大于CorePoolSize时，如果有没有等到新的Task，到了keepaliveTime时间后，就自动终止掉”。那么如果在这个时间之前，等到了新的Task，就可以重用这个线程。到底是怎么重用线程的呢？

# 正文： #

## 原理如下： ##

> 前提条件：假如coreSize=3，maxSize=10，当前存在线程数是5。  
> （注意，存在的这5个线程，并不是你执行ExecuteService.execute/submit时的参数，而是为了执行execute/submit的参数所启动的“内部线程”。这个“内部线程”其实是通过ThreadPoolExecutor的ThreadFactory参数生成的线程，而“execute/submit的参数”是执行在这些“内部线程”里面的。）
> 
> 存在这5个“内部线程”，都访问同一个队列，从队列中去取任务执行（任务就是通过execute/submit提交的Runnable参数），当任务充足时，5个“内部线程”都持续执行。重点是没有任务时怎么办？  
> 没有任务时，这5个“内部线程”都会做下面判断：
> 
>  *  如果poolSize > coreSize，那就从队列里取任务，当过了keepaliveTime这么长时间还没有得到任务的话，当前这个“内部线程”就会结束（使用的是BlockingQueue.poll方法）。
>  *  如果poolSize <= coreSize，那就以“阻塞”的方式，去从队列里取任务，当得到任务后，就继续执行。这样的话，这个线程就不会结束掉。
> 
> 如果没有任务可以继续执行了，最后只剩下coreSize那么多的“内部线程”留在线程池里，等待重用。

## 相关代码如下： ##

1，从ExecuteService.execute方法开始说，execute方法代码如下：

int c = ctl.get();
    // 当“内部线程”数 < coreSize的话，就新建一个“内部线程”
    if (workerCountOf(c) < corePoolSize) {
        if (addWorker(command, true))
            return;
        c = ctl.get();
    }
    // 能执行到这里，说明“内部线程”数 >= coreSize。这样的话，就把任务加入队列。
    // 这个队列，就是上面说的所有“内部线程”访问的那个队列。
    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);

2，下面再看一下addWorker方法，看看“内部线程”是如何建立的。

// 上面一部分代码省略
    boolean workerStarted = false;
    boolean workerAdded = false;
    Worker w = null;
    try {
        // 创建一个Worker实例，把我们提交的任务（也就是firstTask）传入到Worker里。
        // 然后，取得Worker里的thread的属性，并在下面运行这个thread。
        // （其实这个thread属性，就是通过ThreadFactory生成的Thread，也就是所说的“内部线程”）
        // 接下来我们看看Worker类
        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.add(w);
                    int s = workers.size();
                    if (s > largestPoolSize)
                        largestPoolSize = s;
                    workerAdded = true;
                }
            } finally {
                mainLock.unlock();
            }
            if (workerAdded) {
                t.start();
                workerStarted = true;
            }
        }
    } finally {
        if (! workerStarted)
            addWorkerFailed(w);
    }
    return workerStarted;

下面我们看看Worker类，其中和本次说明没有什么关系的代码省略。

Worker(Runnable firstTask) {
        setState(-1); // inhibit interrupts until runWorker
        // 这里把提交的任务保存到Worker里，在后面执行的时候，从Worker里取得这个任务执行。
        this.firstTask = firstTask;
        // 下面就是生成“内部线程”的代码
        this.thread = getThreadFactory().newThread(this);
    }
    
    public void run() {
        // 这个方法，是“内部线程”实际上执行的方法。原理是这样的：
        // 上面构造函数在生成“内部线程”时，把Worker本身，当做参数传给了“内部线程”
        // “内部线程”在自己的run方法里，执行的是worker的run方法，而worker的run方法执行的是runWorker这个方法。
        // runWorker方法是ThreadPoolExectuor的方法，也就是说，线程池里的“内部线程”在调用run方法时，都是执行的这个方法
        //（所以这个方法要注意同步）
        runWorker(this);
    }

3，最后我们看看runWorker方法。

Thread wt = Thread.currentThread();
    Runnable task = w.firstTask;
    w.firstTask = null;
    w.unlock(); // allow interrupts
    boolean completedAbruptly = true;
    try {
        // task!=null是为了运行“内部线程”启动时，提交的给它的那个线程。
        // (task = getTask()) != null) 是当提交给他的任务执行完后，看看队列里还有没有任务可以执行。
        // 如果有的话，这个“内部线程”就可以重用了。
        // 而“内部线程都会做的判断”，就是getTask()方法里面做的，下面看一下getTask()方法
        while (task != null || (task = getTask()) != null) {
            w.lock();
            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) {
       // 省略异常处理
                } finally {
                    afterExecute(task, thrown);
                }
            } finally {
                task = null;
                w.completedTasks++;
                w.unlock();
            }
        }
        completedAbruptly = false;
    } finally {
        processWorkerExit(w, completedAbruptly);
    }

getTask()方法原代码如下

// 这个语句后半部分来判断“内部线程数 > coreSize”
    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;
    }

## 最后 ##

如果要测试看看线程是不是被重用的话，可以通过自己实现ThreadFactory，加个打印语句，来查看newThread方法被调用的次数。下面的代码是参考Executors的代码，加了一条打印语句

static class DefaultThreadFactory implements ThreadFactory {
        private static final AtomicInteger poolNumber = new AtomicInteger(1);
        private final ThreadGroup group;
        private final AtomicInteger threadNumber = new AtomicInteger(1);
        private final String namePrefix;
    
        DefaultThreadFactory() {
            SecurityManager s = System.getSecurityManager();
            group = (s != null) ? s.getThreadGroup() :
                    Thread.currentThread().getThreadGroup();
            namePrefix = "pool-" +
                    poolNumber.getAndIncrement() +
                    "-thread-";
        }
    
        public Thread newThread(Runnable r) {
            Thread t = new Thread(group, r,
                    namePrefix + threadNumber.getAndIncrement(),
                    0);
            if (t.isDaemon())
                t.setDaemon(false);
            if (t.getPriority() != Thread.NORM_PRIORITY)
                t.setPriority(Thread.NORM_PRIORITY);
            // 通过查看调用的次数，可以来判断new了几个内部线程
            System.out.println(namePrefix + threadNumber.get() + " is created " + " hashcode:" + t.hashCode());
            return t;
        }
    }

## 其它 ##

下面的这篇文章是对线程池的一个说明，里面有一个简单的线程池的实现，和ThreadPoolExecutor有些类似，可以参考一下：  
[Thread pools and work queues][]

[Thread pools and work queues]: https://www.ibm.com/developerworks/library/j-jtp0730/