Java 线程池初识

╰半夏微凉° 2022-11-18 04:55 211阅读 0赞

### 文章目录 ###

*   *  线程池的优势
     *   *  降低资源消耗
         *  提升系统响应速度
         *  提高线程的可管理性
     *  线程池的工作原理
     *   *  任务提交流程图
         *  任务提交的流程代码
         *  任务的提交方法
     *  线程池的关闭
     *  线程池创建
     *   *  线程池的参数(7 个)
         *  handler 处理策略
         *   *  AbortPolicy策略
             *  CallerRunsPolicy策略
             *  DiscardOleddestPolicy策略
             *  DiscardPolicy策略
         *  线程池的参数设计
         *   *  corePoolSize
             *  BlockingQueue
         *  ThreadPoolExecutor 的意义
         *   *  FixedThreadPool
             *  SingleThreadExecutor
             *  ScheduledThreadPoolExecutor
             *  ForkJoinPool
             *  newCachedThreadPool
     *  java 线程池工具类
     *   *  ExecutorService 接口
         *  Executors 工具类
     *  线程的回收与复用
     *   *  线程的复用原理
         *  线程池线程的回收
     *  阿里对线程池的使用建议

## 线程池的优势 ##

### 降低资源消耗 ###

*  通过复用已存在的线程和降低线程关闭的次数来尽可能降低系统性能损耗；

### 提升系统响应速度 ###

*  通过复用线程，省去创建线程的过程，因此整体上提升了系统的响应速度；

### 提高线程的可管理性 ###

*  线程是稀缺资源，如果无限制的创建，不仅会消耗系统资源，还会降低系统的稳定性，因此，需要使用线程池来管理线程

## 线程池的工作原理 ##

### 任务提交流程图 ###

![image-20201126163046323][]

*  判断线程池中核心线程池所有的线程是否都在执行任务。如果不是，则新创建一个线程执行刚提交的任务，否则，核心线程池中所有的线程都在执行任务，则进入第2步；
 *  判断当前阻塞队列是否已满，如果未满，则将提交的任务放置在阻塞队列中；否则，则进入第3步；
 *  判断线程池中所有的线程是否都在执行任务，如果没有，则创建一个新的线程来执行任务，否则，则交给饱和策略进行处理

### 任务提交的流程代码 ###

public void execute(Runnable command) { 
        if (command == null)
            throw new NullPointerException();
        /* * Proceed in 3 steps: * * 1. If fewer than corePoolSize threads are running, try to * start a new thread with the given command as its first * task. The call to addWorker atomically checks runState and * workerCount, and so prevents false alarms that would add * threads when it shouldn't, by returning false. * * 2. If a task can be successfully queued, then we still need * to double-check whether we should have added a thread * (because existing ones died since last checking) or that * the pool shut down since entry into this method. So we * recheck state and if necessary roll back the enqueuing if * stopped, or start a new thread if there are none. * * 3. If we cannot queue task, then we try to add a new * thread. If it fails, we know we are shut down or saturated * and so reject the task. */
        int c = ctl.get();
        //如果线程池的线程个数少于corePoolSize则创建新线程执行当前任务
        if (workerCountOf(c) < corePoolSize) { 
            if (addWorker(command, true))
                return;
            c = ctl.get();
        }
        //如果线程个数大于corePoolSize或者创建线程失败，则将任务存放在阻塞队列workQueue中
        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);
    }

![image-20201126163106812][]

*  果当前运行的线程少于corePoolSize，则会创建新的线程来执行新的任务；
 *  如果运行的线程个数等于或者大于corePoolSize，则会将提交的任务存放到阻塞队列workQueue中；
 *  如果当前workQueue队列已满的话且没有超过maximumPoolSize，则会创建新的线程来执行任务
 *  如果线程个数已经超过了maximumPoolSize，则会使用饱和策略RejectedExecutionHandler来进行处理。

需要注意的是，线程池的设计思想就是使用了核心线程池corePoolSize，阻塞队列workQueue和线程池maximumPoolSize，这样的缓存策略来处理任务，实际上这样的设计思想在需要框架中都会使用。

### 任务的提交方法 ###

<table> 
 <thead> 
  <tr> 
   <th>提交方式</th> 
   <th>是否关心返回结果</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td><code>Future&lt;T&gt; submit(Callable&lt;T&gt; task)</code></td> 
   <td>是</td> 
  </tr> 
  <tr> 
   <td><code>void execute(Runnable command)</code></td> 
   <td>否</td> 
  </tr> 
  <tr> 
   <td><code>Future&lt;?&gt; submit(Runnable task)</code></td> 
   <td>否，虽然返回Future，但是其get()方法总是返回null</td> 
  </tr> 
 </tbody> 
</table>

## 线程池的关闭 ##

*  关闭线程池，可以通过shutdown和shutdownNow这两个方法。它们的原理都是遍历线程池中所有的线程，然后依次中断线程。shutdown和shutdownNow还是有不一样的地方：
 *  shutdownNow首先将线程池的状态设置为STOP,然后尝试停止所有的正在执行和未执行任务的线程，并返回等待执行任务的列表；
 *  shutdown只是将线程池的状态设置为SHUTDOWN状态，然后中断所有没有正在执行任务的线程
 *  可以看出shutdown方法会将正在执行的任务继续执行完，而shutdownNow会直接中断正在执行的任务。调用了这两个方法的任意一个，isShutdown方法都会返回true，当所有的线程都关闭成功，才表示线程池成功关闭，这时调用isTerminated方法才会返回true。

## 线程池创建 ##

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

*  corePoolSize和maximumPoolSize设置不当会影响效率，甚至耗尽线程；
 *  workQueue设置不当容易导致OOM；
 *  handler设置不当会导致提交任务时抛出异常。

### 线程池的参数(7 个) ###

*  corePoolSize： 线程池维护线程的最少数量，线程池长期维持的线程数，即使线程处于Idle状态，也不会回收。
 *  maximumPoolSize：线程池维护线程的最大数量
 *  keepAliveTime： 线程池维护线程所允许的空闲时间，超过corePoolSize的线程的idle时长
 *  unit： 线程池维护线程所允许的空闲时间的单位
 *  workQueue： 线程池所使用的缓冲队列，任务的排队队列
 *  ThreadFactory threadFactory, // 新线程的产生方式
 *  handler： 线程池对拒绝任务的处理策略

### handler 处理策略 ###

*  正式使用中一般都会设置一个最大缓冲队列容量，如果线程池满它会对继续添加的任务线程执行指定的拒绝策略，ThreadPoolExcetor的最后一个参数指定了拒绝策略，JDK提供了四种拒绝策略

![image-20201126163122845][]

#### AbortPolicy策略 ####

*  该策略会直接抛出异常，阻止系统正常工作。

#### CallerRunsPolicy策略 ####

*  只要线程池未关闭，该策略直接在调用者线程中，运行当前的被丢弃的任务。

#### DiscardOleddestPolicy策略 ####

*  该策略将丢弃最老的一个请求，也就是即将被执行的任务，并尝试再次提交当前任务。

#### DiscardPolicy策略 ####

*  该策略默默的丢弃无法处理的任务，不予任何处理。

### 线程池的参数设计 ###

#### corePoolSize ####

*  池中所保存的线程数，包括空闲线程。需要注意的是在初创建线程池时线程不会立即启动，直到有任务提交才开始启动线程并逐渐使线程数目达到corePoolSize。若想一开始就创建所有核心线程需调用prestartAllCoreThreads方法。
 *  程池长期维持的线程数，即使线程处于Idle状态，也不会回收。
 *  二八定律下，可以使得任务在单位时间内执行完毕

80% 的情况下，每秒产生100个任务，每个任务执行0.1s,这种情况下100*0.1=10,也就是核心线程数的大小应该为10

#### BlockingQueue ####

*  你可以接受的一个最大等待时间，来决定等待队列的大小

假设你希望一个任务最多等待2s,那么这个时候缓冲队列的大小就是200

### ThreadPoolExecutor 的意义 ###

*  其他java内置的线程池都是对这个类的封装和参数条件进行变化得到的
 *  Executors中创建线程池的快捷方法，实际上是调用了ThreadPoolExecutor的构造方法（定时任务使用的是ScheduledThreadPoolExecutor  
    ![image-20201126163142196][]

#### FixedThreadPool ####

*  创建固定大小的线程池

ExecutorService fixedThreadPool = Executors.newFixedThreadPool(3);
    
    public static ExecutorService newFixedThreadPool(int nThreads) {
        return new ThreadPoolExecutor(
            nThreads, 
            nThreads,
            0L, 
            TimeUnit.MILLISECONDS,
            newLinkedBlockingQueue<Rnnable>());
    }
    public static ExecutorService newFixedThreadPool(int nThreads, ThreadFactory threadFactory) {
            return new ThreadPoolExecutor(nThreads, nThreads,
                                          0L, TimeUnit.MILLISECONDS,
                                          new LinkedBlockingQueue<Runnable>(),
                                          threadFactory);
        }

#### SingleThreadExecutor ####

*  特点：池中只有一个线程，如果扔5个任务进来，那么有4个任务将排队；作用是保证任务的顺序执行

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

#### ScheduledThreadPoolExecutor ####

*  定时器线程池
 *  创建一个定长线程池，支持定时及周期性任务执行
 *

#### ForkJoinPool ####

*  支持大任务分解成小任务的线程池，这是Java8新增线程池，通常配合ForkJoinTask接口的子类RecursiveAction或RecursiveTask使用。

> ForkJoinPool 主要用于实现“分而治之”的算法，特别是分治之后递归调用的函数，例如 quick sort 等

#### newCachedThreadPool ####

> 其实特点就是没有固定大小，上限是整数最大值

*  用newCachedThreadPool()方法创建该线程池对象，创建之初里面一个线程都没有，当execute方法或submit方法向线程池提交任务时，会自动新建线程；如果线程池中有空余线程，则不会新建；这种线程池一般最多情况可以容纳几万个线程，里面的线程空余60s会被回收
 *  用于应对短时间的流量突发场景或者每天的任务高峰期

## java 线程池工具类 ##

### ExecutorService 接口 ###

*  Java 内置线程池的接口，

ExecutorService
    awaitTermination
    invokeAll
    invokeAll
    invokeAny
    invokeAny
    isShutdown
    isTerminated
    shutdown
    shutdownNow
    submit
    submit
    submit

### Executors 工具类 ###

*  内置的线程池工具类，用来创建实现了ExecutorService接口的线程池
 *  Java中创建线程池很简单，只需要调用Executors中相应的便捷方法即可，比如Executors.newFixedThreadPool(int nThreads)，但是便捷不仅隐藏了复杂性，也为我们埋下了潜在的隐患（OOM，线程耗尽）
 *  Executors中创建线程池的快捷方法，实际上是调用了ThreadPoolExecutor的构造方法（定时任务使用的是ScheduledThreadPoolExecutor
 *  构造一个线程池确实需要这么多参数。这些参数中，比较容易引起问题的有corePoolSize, maximumPoolSize, workQueue以及handler：  
    corePoolSize和maximumPoolSize设置不当会影响效率，甚至耗尽线程；  
    workQueue设置不当容易导致OOM；handler设置不当会导致提交任务时抛出异常。

## 线程的回收与复用 ##

### 线程的复用原理 ###

任务提交的时候，通过线程池的规则进行判断，是否需要创建线程来运行该任务，需要的话，则调用addWorker() 方法来创建线程，并执行任务，这里我们就以addWorker 方法作为入口，看一下线程复用的原理

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.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` 对象上了，所以我们这里看一下`Worker`的实现，我们看到Worker实现了Runnable接口

private final class Worker
         extends AbstractQueuedSynchronizer
         implements Runnable
     { 
         /** * This class will never be serialized, but we provide a * serialVersionUID to suppress a javac warning. */
         private static final long serialVersionUID = 6138294804551838833L;
    
         /** 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;
    
         /** * Creates with given first task and thread from ThreadFactory. * @param firstTask the first task (null if none) */
         Worker(Runnable firstTask) { 
             setState(-1); // inhibit interrupts until runWorker
             this.firstTask = firstTask;
             this.thread = getThreadFactory().newThread(this);
         }
    
         /** Delegates main run loop to outer runWorker */
         public void run() { 
             runWorker(this);
         }
    
         // Lock methods
         //
         // The value 0 represents the unlocked state.
         // The value 1 represents the locked state.
    
         protected boolean isHeldExclusively() { 
             return getState() != 0;
         }
    
         protected boolean tryAcquire(int unused) { 
             if (compareAndSetState(0, 1)) { 
                 setExclusiveOwnerThread(Thread.currentThread());
                 return true;
             }
             return false;
         }
    
         protected boolean tryRelease(int unused) { 
             setExclusiveOwnerThread(null);
             setState(0);
             return true;
         }
    
         public void lock()        {  acquire(1); }
         public boolean tryLock()  {  return tryAcquire(1); }
         public void unlock()      {  release(1); }
         public boolean isLocked() {  return isHeldExclusively(); }
    
         void interruptIfStarted() { 
             Thread t;
             if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) { 
                 try { 
                     t.interrupt();
                 } catch (SecurityException ignore) { 
                 }
             }
         }
     }

然后我们看到run 方法里面调用了`runWorker` 方法，这里我们看一下这个方法的实现,这个方法很重要，而且有大段的注释信息

/** * Main worker run loop. Repeatedly gets tasks from queue and * executes them, while coping with a number of issues: * 循环地从队列里面取出task 然后执行它们 * 1. We may start out with an initial task, in which case we * don't need to get the first one. Otherwise, as long as pool is * running, we get tasks from getTask. If it returns null then the * worker exits due to changed pool state or configuration * parameters. Other exits result from exception throws in * external code, in which case completedAbruptly holds, which * usually leads processWorkerExit to replace this thread. * * 2. Before running any task, the lock is acquired to prevent * other pool interrupts while the task is executing, and then we * ensure that unless pool is stopping, this thread does not have * its interrupt set. * * 3. Each task run is preceded by a call to beforeExecute, which * might throw an exception, in which case we cause thread to die * (breaking loop with completedAbruptly true) without processing * the task. * * 4. Assuming beforeExecute completes normally, we run the task, * gathering any of its thrown exceptions to send to afterExecute. * We separately handle RuntimeException, Error (both of which the * specs guarantee that we trap) and arbitrary Throwables. * Because we cannot rethrow Throwables within Runnable.run, we * wrap them within Errors on the way out (to the thread's * UncaughtExceptionHandler). Any thrown exception also * conservatively causes thread to die. * * 5. After task.run completes, we call afterExecute, which may * also throw an exception, which will also cause thread to * die. According to JLS Sec 14.20, this exception is the one that * will be in effect even if task.run throws. * * The net effect of the exception mechanics is that afterExecute * and the thread's UncaughtExceptionHandler have as accurate * information as we can provide about any problems encountered by * user code. * * @param w the worker */
    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);
        }
    }

可以看出线程复用的原理就是while 循环，这个线程(worker)一但启动，就会进入while 循环，只要task 或者getTask()的返回值不为null ,则该worker 一直处理队列中的task,也就是我们通过线程池提交进来的runable 对象。我们看到worker运行的第一个任务也就是firstTask 是我们创建worker时候的runable 对象，后面一直处理的都是队列里的对象

### 线程池线程的回收 ###

当线程池中的线程数，超过核心线程数时。此时如果任务量下降，肯定会出现有一些线程处于无任务执行的空闲状态。那么如果这个线程的空闲时间超过了 keepAliveTime&unit 配置的时长后，就会被回收。

需要注意的是，对于线程池来说，它只负责管理线程，对于创建的线程是不区分所谓的「核心线程」和「非核心线程」的，它只对线程池中的线程总数进行管理，当回收的线程数达到 corePoolSize 时，回收的过程就会停止，也就是说其实没有核心线程和非核心线程之分，当回收后的线程数量等于corePoolSize，那么这些未被回收的线程就是核心线程

对于线程池的核心线程数中的线程，也有回收的办法，可以通过 `allowCoreThreadTimeOut(true)` 方法设置，在核心线程空闲的时候，一旦超过 keepAliveTime&unit 配置的时间，也将其回收掉。

public void allowCoreThreadTimeOut(boolean value) {
      if (value && keepAliveTime <= 0)
        thrownew IllegalArgumentException("Core threads must have nonzero keep alive times");
      if (value != allowCoreThreadTimeOut) {
        allowCoreThreadTimeOut = value;
        if (value)
          interruptIdleWorkers();
      }
    }

`allowCoreThreadTimeOut()` 能被设置的前提是 keepAliveTime 不能为 0

好了，现在我们看一下，到底如何进行线程的回收，我们这里的入口方法还是`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) { 
              // 执行线程
            }
            completedAbruptly = false;
        } finally { 
          	// 线程回收
            processWorkerExit(w, completedAbruptly);
        }
    }

其实我们可以看到，当getTask() 的返回值是null 的时候，那这个时候开始进行线程回收，下面我们具体看一下getTask 方法什么时候返回null

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

第一种情况，当线程池的状态已经是STOP，TIDYING, TERMINATED，或者是SHUTDOWN且工作队列为空；

第二种情况，工作线程数已经大于最大线程数或当前工作线程已超时，且还有其他工作线程或任务队列为空。

下面我们看一下线程回收的这个方法processWorkerExit

private void processWorkerExit(Worker w, boolean completedAbruptly) { 
        if (completedAbruptly) // If abrupt, then workerCount wasn't adjusted
            decrementWorkerCount();
    
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try { 
            completedTaskCount += w.completedTasks;
            workers.remove(w);
        } finally { 
            mainLock.unlock();
        }
    
        tryTerminate();
    
        int c = ctl.get();
        if (runStateLessThan(c, STOP)) { 
            if (!completedAbruptly) { 
                int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
                if (min == 0 && ! workQueue.isEmpty())
                    min = 1;
                if (workerCountOf(c) >= min)
                    return; // replacement not needed
            }
            addWorker(null, false);
        }
    }

首先就从worker 集合中将这个worker 移除，这个操作和worker 的添加是对应的

## 阿里对线程池的使用建议 ##

不要使用Executors.newXXXThreadPool()快捷方法创建线程池，**因为这种方式会使用无界的任务队列，为避免OOM，我们应该使用ThreadPoolExecutor的构造方法手动指定队列的最大长度**

ExecutorService executorService = new ThreadPoolExecutor(2, 2, 
                    0, TimeUnit.SECONDS, 
                    new ArrayBlockingQueue<>(512), // 使用有界队列，避免OOM
                    new ThreadPoolExecutor.DiscardPolicy());

int poolSize = Runtime.getRuntime().availableProcessors() * 2;
    BlockingQueue<Runnable> queue = new ArrayBlockingQueue<>(512);
    RejectedExecutionHandler policy = new ThreadPoolExecutor.DiscardPolicy();
    executorService = new ThreadPoolExecutor(poolSize, poolSize,
        0, TimeUnit.SECONDS,
                queue,
                policy);

其实很多场景下，即使不OOM,也不能使用无解的队列，因为这个时候我们应该做的是及时发现，然后增加资源，提升系统的处理能力，而不是为了维护表象上的稳定，使得任务大量堆积。

[image-20201126163046323]: /images/20221022/85112c909ad2472da420c26b201e4f95.png
[image-20201126163106812]: /images/20221022/81f25ecaf983498481129a6a5612ff54.png
[image-20201126163122845]: /images/20221022/bb9e8c190fe24aa6b94f015e8e3c0e52.png
[image-20201126163142196]: /images/20221022/e6ca523fb21548c683af79777a0c367e.png