java多线程之ThreadPoolExecutor

构造函数

public ThreadPoolExecutor(int corePoolSize,
                             int maximumPoolSize,
                             long keepAliveTime,
                             TimeUnit unit,
                             BlockingQueue<Runnable> workQueue,
                             ThreadFactory threadFactory,
                             RejectedExecutionHandler handler)

解释一下这几个参数

corePoolSize：线程池核心线程数量；线程池刚刚创建的时候线程池中有0个线程在执行任务，当任务提交到线程池时，如果当前线程数量小于corePoolSize，就会创建新的线程执行任务；如果当前线程已经等于corePoolSize，就会把任务放在workQueue任务队列中。

maximumPoolSize：线程池最大线程数量；当workQueue任务队列已满，此时如果添加新的任务，线程池就会创建新的线程来执行任务，但是最大不会超过maximumPoolSize

keepAliveTime：线程池中线程的数量大于corePoolSize时，空闲线程的超时回收时间。如果 allowCoreThreadTimeOut为true 则核心线程也会超时回收。

threadFactory：创建线程的工厂。

handler：拒绝策略，当workQueue已满而且线程数达到maximumPoolSize，无法处理新来的任务时的拒绝策略。

##

addWorker方法

private boolean addWorker(Runnable firstTask, boolean core) {
    retry:
    for (;;) {
        int c = ctl.get();
        int rs = runStateOf(c);

        // Check if queue empty only if necessary.
        if (rs >= SHUTDOWN &&
            ! (rs == SHUTDOWN &&
               firstTask == null &&
               ! workQueue.isEmpty()))
            return false;

        for (;;) {
            int wc = workerCountOf(c);
            if (wc >= CAPACITY ||
                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;
            mainLock.lock();
            try {
                // Recheck while holding lock.
                // Back out on ThreadFactory failure or if
                // shut down before lock acquired.
                int rs = runStateOf(ctl.get());
                  //判断线程是否被SHUTDOWN
                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;
}

在里面完成了状态检查、新建任务、执行任务等一系列动作。可以看到上面创建线程的流程和上面画的图一样。

接下来看t.start();到底做了什么

Worker(Runnable firstTask) {
          setState(-1); // inhibit interrupts until runWorker
          this.firstTask = firstTask;
          this.thread = getThreadFactory().newThread(this);
      }

在Worker 的构造函数里面可以看到 Worker.thread 是使用自己来构造的一个线程，所以上面的t.start() 所执行的就是work的run方法。也就是runWorker

runWorker

final void runWorker(Worker w) {
       Thread wt = Thread.currentThread();
       Runnable task = w.firstTask;
       w.firstTask = null;
       w.unlock(); // allow interrupts
       boolean completedAbruptly = true;
       try {
           while (task != null || (task = getTask()) != null) {
               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);
       }
   }

可以看到这个方法一直在执行我们提交的任务的run() 方法，第一次会执行创建这个线程的任务的run方法，后面会执通过getTask()方法重队列里面的任务。