并发编程--线程池ThreadPoolExecutor实现原理（二）

爱被打了一巴掌 2022-06-11 00:59 168阅读 0赞

ThreadPoolExecutor是线程池类。对于线程池，可以通俗的将它理解为"存放一定数量线程的一个线程集合。线程池允许若个线程同时允许，允许同时运行的线程数量就是线程池的容量；当添加的到线程池中的线程超过它的容量时，会有一部分线程阻塞等待。线程池会通过相应的调度策略和拒绝策略，对添加到线程池中的线程进行管理。"

1. workers

workers是HashSet<Work>类型，即它是一个Worker集合。而一个Worker对应一个线程，也就是说线程池通过workers包含了"一个线程集合"。当Worker对应的线程池启动时，它会执行线程池中的任务；当执行完一个任务后，它会从线程池的阻塞队列中取出一个阻塞的任务来继续运行。

wokers的作用是，线程池通过它实现了"允许多个线程同时运行"。

2. workQueue

workQueue是BlockingQueue类型，即它是一个阻塞队列。当线程池中的线程数超过它的容量的时候，线程会进入阻塞队列进行阻塞等待。

通过workQueue，线程池实现了阻塞功能。

3. mainLock

mainLock是互斥锁，通过mainLock实现了对线程池的互斥访问。

4. corePoolSize和maximumPoolSize

corePoolSize是"核心池大小"，maximumPoolSize是"最大池大小"。它们的作用是调整"线程池中实际运行的线程的数量"。  
    例如，当新任务提交给线程池时(通过execute方法)。  
          -- 如果此时，线程池中运行的线程数量< corePoolSize，则创建新线程来处理请求。  
          -- 如果此时，线程池中运行的线程数量> corePoolSize，但是却< maximumPoolSize；则仅当阻塞队列满时才创建新线程。

如果设置的 corePoolSize 和 maximumPoolSize 相同，则创建了固定大小的线程池。如果将 maximumPoolSize 设置为基本的无界值（如 Integer.MAX\_VALUE），则允许池适应任意数量的并发任务。在大多数情况下，核心池大小和最大池大小的值是在创建线程池设置的；但是，也可以使用 setCorePoolSize(int) 和 setMaximumPoolSize(int) 进行动态更改。

5. poolSize

poolSize是当前线程池的实际大小，即线程池中任务的数量。

6. allowCoreThreadTimeOut和keepAliveTime

allowCoreThreadTimeOut表示是否允许"线程在空闲状态时，仍然能够存活"；而keepAliveTime是当线程池处于空闲状态的时候，超过keepAliveTime时间之后，空闲的线程会被终止。

7. threadFactory

threadFactory是ThreadFactory对象。它是一个线程工厂类，"线程池通过ThreadFactory创建线程"。

8. handler

handler是RejectedExecutionHandler类型。它是"线程池拒绝策略"的句柄，也就是说"当某任务添加到线程池中，而线程池拒绝该任务时，线程池会通过handler进行相应的处理"。

9. Worker

Worker是ThreadPoolExecutor线程池来的私有类，其实现了Runnable接口，因此Worker也是一个线程，添加到ThreadPoolExecutor中的线程最终都会构造成为一个worker任务，并且Worker提供了一个thread变量，在线程池中会获取到thread并执行start方法，  
最终执行的是woker的run方法，在worker中实现了线程的重用功能，当调用worker的run方法时，如果此时线程池的workQueue阻塞队列中还有阻塞线程，则直接从阻塞队列中获取任务并调用其run方法，这样就可以在一个线程中依次执行多个线程任务，通过重复利用已创建的线程降低线程创建和销毁造成的消耗。  
综上所说，线程池通过workers来管理"线程集合"，每个线程在启动后，会执行线程池中的任务；当一个任务执行完后，它会从线程池的阻塞队列中取出任务来继续运行。阻塞队列是管理线程池任务的队列，当添加到线程池中的任务超过线程池的容量时，该任务就会进入阻塞队列进行等待。

接下来我们通过分析ThreadPoolExecutor的源码来了解线程池的实现原理。

1、提交线程任务

在线程池中提交线程，如果线程池未满，执行的线程数小于corePoolSize，则调用addWorker方法，将线程添加到执行队列中并执行线程，如果线程池已满，则将线程添加到workQueue阻塞线程队列中，将线程挂起。

public void execute(Runnable command) { // 如果任务为null，则抛出异常。 if (command == null) throw new NullPointerException(); // 获取ctl对应的int值。该int值保存了"线程池中任务的数量"和"线程池状态"信息 int c = ctl.get(); // 当线程池中的任务数量 < "核心池大小"时，即线程池中少于corePoolSize个任务。 // 则通过addWorker(command, true)新建一个线程，并将任务(command)添加到该线程中；然后，启动该线程从而执行任务。 if (workerCountOf(c) < corePoolSize) { if (addWorker(command, true)) return; c = ctl.get(); } // 当线程池中的任务数量 >= "核心池大小"时， // 而且，"线程池处于允许状态"时，则尝试将任务添加到阻塞队列中。 if (isRunning(c) && workQueue.offer(command)) { // 再次确认“线程池状态”，若线程池异常终止了，则删除任务；然后通过reject()执行相应的拒绝策略的内容。 int recheck = ctl.get(); if (! isRunning(recheck) && remove(command)) reject(command); // 否则，如果"线程池中任务数量"为0，则通过addWorker(null, false)尝试新建一个线程，新建线程对应的任务为null。 else if (workerCountOf(recheck) == 0) addWorker(null, false); } // 通过addWorker(command, false)新建一个线程，并将任务(command)添加到该线程中；然后，启动该线程从而执行任务。 // 如果addWorker(command, false)执行失败，则通过reject()执行相应的拒绝策略的内容。 else if (!addWorker(command, false)) reject(command); }

addworker中做的操作简单来说将线程添加到workers队列中，并执行这个线程，当线程执行结束之后删除掉此线程。

private boolean addWorker(Runnable firstTask, boolean core) { retry: // 更新"线程池状态和计数"标记，即更新ctl。 for (;;) { // 获取ctl对应的int值。该int值保存了"线程池中任务的数量"和"线程池状态"信息 int c = ctl.get(); // 获取线程池状态。 int rs = runStateOf(c); // 有效性检查 if (rs >= SHUTDOWN && ! (rs == SHUTDOWN && firstTask == null && ! workQueue.isEmpty())) return false; for (;;) { // 获取线程池中任务的数量。 int wc = workerCountOf(c); // 如果"线程池中任务的数量"超过限制，则返回false。 if (wc >= CAPACITY || wc >= (core ? corePoolSize : maximumPoolSize)) return false; // 通过CAS函数将c的值+1。操作失败的话，则退出循环。 if (compareAndIncrementWorkerCount(c)) break retry; c = ctl.get(); // Re-read ctl // 检查"线程池状态"，如果与之前的状态不同，则从retry重新开始。 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 { final ReentrantLock mainLock = this.mainLock; // 新建Worker，并且指定firstTask为Worker的第一个任务。 w = new Worker(firstTask); // 获取Worker对应的线程。 final Thread t = w.thread; if (t != null) { // 获取锁 mainLock.lock(); try { int c = ctl.get(); int rs = runStateOf(c); // 再次确认"线程池状态" if (rs < SHUTDOWN || (rs == SHUTDOWN && firstTask == null)) { if (t.isAlive()) // precheck that t is startable throw new IllegalThreadStateException(); // 将Worker对象(w)添加到"线程池的Worker集合(workers)"中 workers.add(w); // 更新largestPoolSize 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; }

线程的执行是调用task.start()方法，最终调用的是Worker类（worker是一个线程实现类）的run方法

//调用start方法最终会执行线程的run方法 public void run() { runWorker(this); }

runWorker是实现线程池实现的关键，首先从worker中获取线程task，如果task不为空则执行task.run方法，这样当前线程就执行完毕了，接下来还是为执行while循环，调用task = getTask，简单来说getTask获取到的线程为之前添加到阻塞队列中的线程，如果存在阻塞队列则直接获取阻塞队列并调用task.run方法，这样就实现了在一个线程中执行多个线程的逻辑方法run，就不需要系统调用线程的start方法操作系统创建新的线程和销毁线程操作。

final void runWorker(Worker w) { Thread wt = Thread.currentThread(); //获取线程 Runnable task = w.firstTask; w.firstTask = null; w.unlock(); // allow interrupts boolean completedAbruptly = true; try { //如果当前任务task执行完之后，会设置task为空，接下来执行task=getTask()是从阻塞队列中获取新的线程任务，这样阻塞的线程任务就不需要操作系统初始化新的线程，接下来在当前线程中直接执行run方法实现的逻辑 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 { //执行线程的run方法 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); } }

简单来说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. 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; } } }

接下来我们通过一个用例图来说明一下，当用户添加一个任务线程，会通过Worker及ThreadFactory包装成一个worker对象，接下来调用线程的start方法，其实是调用Worker的run方法，在worker的run方法中如果阻塞线程队列workerQueue中有阻塞线程任务则通过getTask方法获取到线程，然后直接调用线程任务的run函数，这样就可以通过在一个线程中依次执行多个线程的逻辑处理方法run，这样操作系统就减少了线程创建和销毁资源消耗

![Image 1][]

参考：http://www.cnblogs.com/skywang12345/p/3509941.html

[Image 1]: