从Java线程池实现原理到如何配置一个合适的线程池

- 日理万妓 2024-04-17 21:23 127阅读 0赞

### 文章目录 ###

*  前言
 *  为什么要使用线程池
 *  线程池的运行原理的源码分析
 *   *  线程池状态
     *  启动线程池
     *   *  创建线程
         *  线程池的工人 “Worker”
         *  启动工人，开启线程
         *  工人获取任务
     *  拒绝策略
     *   *  默认拒绝策略（抛异常）
         *  最旧淘汰策略（丢弃）
     *  关闭线程池
     *   *  温和关闭线程池
         *  暴力关闭线程池
     *  小结
 *  配置一个合适的线程池
 *   *  合适的线程数量
     *  合适的阻塞队列
     *  低耗时任务
     *  高耗时任务
     *  定时任务

# 前言 #

> 本文会出现一些阻塞队列相关的知识，这种队列除了有队列Queue的特性之外，还有阻塞的特性，例如take方法会获取元素时若队列为空会阻塞直到有元素被添加等等，还有很多特性，可以看[这篇文章][Link 1]，带你了解几个基本的阻塞队列  
> 除了上面的文章，我还强烈推荐你读[这篇文章][Link 2]，了解JDK 1.7 新增的一个高吞吐的队列实现

本篇文章的议题如下：

*  线程池的运行原理的源码分析
 *  如何配置一个合适的线程池？

# 为什么要使用线程池 #

首先，为什么我们需要线程池呢？直接 new Thread().start() 不行吗？

在 Thread\#start 方法执行后，会执行JNI方法，其在底层创建了一个POSIX Thread，然后会在该线程内回调执行我们自定义的Thread\#run方法，在执行完run方法或执行期间抛出异常，都会导致该线程的执行结束，接着就会delete自己，释放POSIX Thread线程资源。

可以看到，在 new Thread().start() 的过程中，有两个步骤和我们执行具体任务是无关的：

*  向os申请线程资源，创建POSIX Thread
 *  delete自己，释放POSIX Thread线程资源

在有多个任务需要被多线程执行的时候，以上两步会在每一个任务执行和结束时重复执行，这两步其实可以省掉的，这也就是线程池需要做的事情。

所以，线程池第一个好处即为 **减少了创建和释放线程的开销** 。

其次，我们使用线程的场景有多种多样，有时候是IO密集型（IO操作占比较大的比重，这里的IO操作指磁盘IO、数据库IO、网络IO、大内存操作等等的CPU需要长时间等待的IO操作），有时候是CPU密集型，又有时候方法会被高并发访问，并且方法为耗时操作，亦或是低并发，耗时、高并发，短时、低并发、短时等等，场景有很多，单纯的new一个线程的操作已经不能很好的帮我们去管理线程。我们需要线程池来决定应该创建多少个线程来做，需要保持多少个线程重复工作才最合适。

所以，线程池的第二个好处即为 **帮助管理和控制线程的数量、行为**

最后，在调试的时候（例如jstack）排查多线程问题，免不了需要查看线程名称，此时线程池可以给一类工作、场景命名，比如A功能为一个名字，B功能为另一个名字，这样排查起来就可以很好的定位是哪个功能模块的线程有问题。

所以，线程池的第三个好处即为 **方便调试，定位问题**

当然，线程池还有很多好处，例如可以处理线程抛出的异常之类，这里就不一一赘述，在读者研究完线程池的源码，自己使用过线程池之后相信都会体会到。

# 线程池的运行原理的源码分析 #

JDK的线程池的标准实现一般有3种：

*  java.util.concurrent.ForkJoinPool （”窃取“任务，一定场景下高吞吐量）
 *  java.util.concurrent.ScheduledThreadPoolExecutor （定时任务，异步的线程定时执行）
 *  java.util.concurrent.ThreadPoolExecutor （普通的线程池）

在这里，我们分析最后一种线程池实现，其他的线程池都是基于同一个思想，所以我们取最common的实现来分析是最有价值的。

## 线程池状态 ##

// 表示线程池当前的状态，使用32位长的int来保存状态信息
    // 其中高3位表示线程池的状态，低29位表示线程池的数量
    // 这里将ctl控制位初始化为RUNNING状态，并且0个线程数量
    private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
    private static final int COUNT_BITS = Integer.SIZE - 3;
    // 后29位全为1，代表线程池线程的最大数量，一般不会创建这么多
    private static final int CAPACITY   = (1 << COUNT_BITS) - 1;
    
    // runState is stored in the high-order bits
    private static final int RUNNING    = -1 << COUNT_BITS; // 32位中高3位为111
    private static final int SHUTDOWN   =  0 << COUNT_BITS; // 32位中高3位为000
    private static final int STOP       =  1 << COUNT_BITS; // 32位中高3位为001
    private static final int TIDYING    =  2 << COUNT_BITS; // 32位中高3位为010
    private static final int TERMINATED =  3 << COUNT_BITS; // 32位中高3位为011

下面几个方法来辨别当前的状态信息：

// Packing and unpacking ctl
    // ~表示取反操作，这里~CAPACITY就代表高3位为1，低29位为0的一个数字
    // 所以这里就只是判断c这个数的高3位有哪些为1，不关心低29位
    // 返回的值就可以与上面5个状态做比较，所以这个方法是判断线程池的运行状态
    private static int runStateOf(int c)     {  return c & ~CAPACITY; }
    // 同理可得，这个方法可以判断c的低29位数量，判断线程池当前线程数量
    private static int workerCountOf(int c)  {  return c & CAPACITY; }
    // 从上面初始化ctl控制位的ctlOf(RUNNING, 0)使用就可以看出其作用了
    private static int ctlOf(int rs, int wc) {  return rs | wc; }

## 启动线程池 ##

我们知道，调用线程池的execute、submit方法都可以提交一个任务，线程池就会调度线程去执行其run方法，那么就以此为入口方法来看看线程池到底在提交任务之后都做了什么：

public void execute(Runnable command) { 
      if (command == null)
        throw new NullPointerException();
      // 获取控制位，刚创建线程池的时候ctl控制位为RUNNING状态，0个线程数
      int c = ctl.get();
      // workerCountOf获取此时线程数量，上面有提到
      // 判断线程数量是否小于核心线程数量
      if (workerCountOf(c) < corePoolSize) { 
        // 如果此时线程数量小于核心数量，则addWorker创建线程
        if (addWorker(command, true))
          return;
        // 创建线程失败了，重新获取控制位
        c = ctl.get();
      }
      // 判断线程池是否是Running状态，如果是则将任务入workQueue这个任务队列
      if (isRunning(c) && workQueue.offer(command)) { 
        // 到这里的话可以说明任务入队成功
        int recheck = ctl.get();
        // 线程池如果此时被shutDown，此时需要拒绝任务
        if (! isRunning(recheck) && remove(command))
          reject(command);
        // 需要确认一下线程数量，如果没有线程则需要新增一个线程
        else if (workerCountOf(recheck) == 0)
          addWorker(null, false);
      }
      // 如果到这里，一般说明核心线程满了且任务入workQueue队列失败了
      // 此时继续创建一个线程
      else if (!addWorker(command, false))
        // 创建线程失败了，只能拒绝了
        reject(command);
    }

这里的大致脉络还算比较清晰，只不过我们需要深入以下3个细节分支：

1.  addWorker：创建线程过程
2.  workQueue：此队列有什么用？为什么往里面塞任务就可以执行？
3.  reject：拒绝策略

我们按顺序依次分析他们的作用

### 创建线程 ###

这里直接来看addWorker方法，是如何创建新的线程的

private boolean addWorker(Runnable firstTask, boolean core) { 
      retry:
      for (;;) { 
        // 获取此时控制位
        int c = ctl.get();
        // 获取运行状态（只取高3位的信息）
        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);
          // core代表此时是否创建核心线程，false代表创建临时线程
          // 所以这里在判断是否超过最大核心线程数或是否超过最大临时线程数
          if (wc >= CAPACITY ||
              wc >= (core ? corePoolSize : maximumPoolSize))
            // 超过则创建失败
            return false;
          // 到这里则数量够创建，则将ctl控制位增加1，代表增加一个线程数量
          if (compareAndIncrementWorkerCount(c))
            // 跳出到最外层的retry语句
            break retry;
          // 如果到这里，说明上面ctl控制位增1冲突了，则需要重新获取最新的ctl控制位
          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 { 
        // 到这里说明ctl增1成功了，所以直接来创建工作线程了
        // 在Worker的构造函数中会创建一个Thread
        w = new Worker(firstTask);
        // 此thread为构造函数中创建出来的
        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());
    
            // 若状态为Running
            if (rs < SHUTDOWN ||
                (rs == SHUTDOWN && firstTask == null)) { 
              if (t.isAlive()) // precheck that t is startable
                throw new IllegalThreadStateException();
              // 在workers这个数据结构中新增记录这个worker
              workers.add(w);
              int s = workers.size();
              if (s > largestPoolSize)
                largestPoolSize = s;
              // 标示添加worker成功了
              workerAdded = true;
            }
          } finally { 
            mainLock.unlock();
          }
          // 仅在添加worker成功且成功启动线程之后，workerStarted才为true
          // workerStarted为true才代表最终的添加成功
          if (workerAdded) { 
            t.start();
            workerStarted = true;
          }
        }
      } finally { 
        // 如果添加没有成功，需要删除此worker的一系列信息
        if (! workerStarted)
          addWorkerFailed(w);
      }
      // 返回最终新增worker的结果
      return workerStarted;
    }

这里按正常流程来说，会经历以下几个阶段：

1.  更新ctl控制位信息，数量+1
2.  创建Worker，并添加到workers这个数据结构以记录此worker
3.  线程池状态确认为RUNING状态，则执行t.start()方法，启动线程

由此可见，addWorker的作用就是创建Worker类保存下来，并启动一个新的线程

### 线程池的工人 “Worker” ###

值得一提到是刚刚代码里的t.start()方法，其开启了worker创建的thread，在介绍之前，先来看看Worker类的结构![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNzM3NzE2_size_16_color_FFFFFF_t_70]  
可以看到，Worker其实是一个Runnable，其也使用了AQS的特性（关于AQS，可以看[这篇文章][Link 3]）

我们先来回顾一下Worker的构造函数是如何创建线程的

Worker(Runnable firstTask) { 
      setState(-1); // inhibit interrupts until runWorker
      this.firstTask = firstTask;
      // 获取线程池的ThreadFactory，传入Worker对象，创建线程
      this.thread = getThreadFactory().newThread(this);
    }

关于ThreadFactory，其在创建线程池时可以配置一个，所以由此可见这个是一个入口，我们可以自定义线程池创建的每一个线程的信息，值得一提的是这里传入了this

@Override
    public Thread newThread(Runnable r) { 
      return new Thread(r, "customThreadName");
    }

这里简略的给出了一个newThread的简单实现，这里将Worker作为一个Runnable传入此方法，也就是说，Worker类的成员变量thread线程，其实就是它自己。

### 启动工人，开启线程 ###

所以上面的addWorker方法中的t.start()方法其实就是启动一个线程，然后此线程回调了自己的run方法（因为Worker就是一个Runnable），所以到这里我们需要看Worker的run方法

public void run() { 
      runWorker(this);
    }
    
    final void runWorker(Worker w) { 
      Thread wt = Thread.currentThread();
      // 一般来说，这里第一个任务就是创建worker时携带的需要执行的任务
      Runnable task = w.firstTask;
      // 清空
      w.firstTask = null;
      // worker其实也充当了锁的角色，这从它AQS的特性可以看出来
      // 至于为什么这里要锁，后面会详细讲解
      w.unlock(); // allow interrupts
      boolean completedAbruptly = true;
      try { 
        // 第一次task不为空，直接执行任务
        // 后面的话都会去getTask方法里取任务
        // 换句话说，在while循环中，task这个局部变量都是不为null的，都是有任务的
        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) { 
              thrown = x; throw x;
            } catch (Error x) { 
              thrown = x; throw x;
            } catch (Throwable x) { 
              thrown = x; throw new Error(x);
            } finally { 
              // 期间出现异常，都会进入afterExecute方法处理
              // 此方法为空方法，留给用户自定义出现异常时的操作或是没异常的完成任务操作
              afterExecute(task, thrown);
            }
          } finally { 
            task = null;
            w.completedTasks++;
            w.unlock();
          }
        }
        completedAbruptly = false;
      } finally { 
        processWorkerExit(w, completedAbruptly);
      }
    }

分析到这里我们可以知道，addWorker创建并启动的这个线程，其实就是在一直跑，也就是runWorker方法中不停的做while循环，直到循环条件 getTask() 为空才会停下。

### 工人获取任务 ###

工人除了去执行第一个创建时顺便携带的任务之外，还会一直去获取”上级“分配下来的任务，这里的上级就是workQueue这个阻塞队列。这里我们重点看看 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.
        // 如果线程池正在被shutdown，就不取任务了，返回null停止worker的无限循环
        if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) { 
          decrementWorkerCount();
          return null;
        }
    
        // 获取线程数量
        int wc = workerCountOf(c);
    
        // Are workers subject to culling?
        // allowCoreThreadTimeOut这个参数在创建线程池时可以配置为true
        // 表示核心线程也视为临时worker，也会过期释放资源
        // 或者线程数量已经大于核心数量，则现在的worker都是临时工，过一段空闲时间会被清除
        boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
    
        // 当timed = true 表示worker为临时工
        // 且timedOut = true（在下面被设置） 表示已经超过了一段空闲时间
        // 且workQueue为空表示现在没有任务，线程池空闲状态 或wc大于1 表示留一个线程也够
        if ((wc > maximumPoolSize || (timed && timedOut))
            && (wc > 1 || workQueue.isEmpty())) { 
          // 满足以上条件，都会开始清理worker，表示当前线程池开始进入空闲状态了
          // 减少线程数
          if (compareAndDecrementWorkerCount(c))
            // 返回空，让worker线程从while循环中break出来
            return null;
          continue;
        }
    
        try { 
          // timed表示此时是否为临时工
          // 这里可以看出，如果是临时工，会从任务队列workQueue中有时限地拿任务
          // 而不是临时工的话，将一直阻塞，直到workQueue中有任务
          Runnable r = timed ?
            workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
          workQueue.take();
          if (r != null)
            // 获取到任务，返回出去执行
            return r;
          // 到这里的话，就是没获取到任务，就表示线程池有点空闲了，也只有临时工才会到这一步
          // 则设置timedOut=true，准备开始清理临时工
          timedOut = true;
        } catch (InterruptedException retry) { 
          timedOut = false;
        }
      }
    }

这里的逻辑就会分为正式员工和临时工了，我们知道，正式员工与公司是有合同的，不会轻易辞退，而临时工则是在公司比较忙的时候临时添加的人手，在公司不忙的时候其实临时工就没有多大存在的必要了。以此类推也可以看出以上代码也是符合这个逻辑的，正式工人会一直阻塞去取workQueue里的任务，而临时工只会阻塞一会时间，这个时间是我们创建线程池时可以自定义的一个空闲时间，在超过这个空闲时间后还没任务，那些临时工人就会消亡（本质是worker线程的run方法执行结束，底层JVM会delete释放os线程，读者只需要知道，线程的run方法结束就会释放线程资源）。

## 拒绝策略 ##

在最开头中，线程池的execute方法我们可以看到，在线程大于核心线程数且队列也塞不下任务且增加临时工也失败（超过最大线程阈值，自行配置的值）的情况下，就会拒绝此任务，执行reject方法

final void reject(Runnable command) { 
        handler.rejectedExecution(command, this);
    }

可以看到，这里拒绝的逻辑交给了handler类的rejectedExecution方法去做，而handler是在哪里被初始化的呢？我们全局搜索可以看到，其在线程池的构造函数中被初始化

public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue,
                              ThreadFactory threadFactory,
                              RejectedExecutionHandler handler) { 
        ...
        this.handler = handler;
    }

也就是说，我们可以自定义拒绝策略，这里我们看看JDK自带的几个默认实现

### 默认拒绝策略（抛异常） ###

public static class AbortPolicy implements RejectedExecutionHandler { 
        public AbortPolicy() {  }
      
        public void rejectedExecution(Runnable r, ThreadPoolExecutor e) { 
            throw new RejectedExecutionException("Task " + r.toString() +
                                                 " rejected from " +
                                                 e.toString());
        }
    }

如果不配置拒绝策略，默认在任务满的时候会抛出一个异常

### 最旧淘汰策略（丢弃） ###

public static class DiscardOldestPolicy implements RejectedExecutionHandler { 
        public DiscardOldestPolicy() {  }
      
        public void rejectedExecution(Runnable r, ThreadPoolExecutor e) { 
            if (!e.isShutdown()) { 
                e.getQueue().poll();
                e.execute(r);
            }
        }
    }

再介绍一个前段时间刚用的策略，场景是定时更新某个数据，若任务满了，可以将最旧的任务从队列中丢弃，然后执行我这个任务，因为最旧的更新任务可能已经过了某个时效，可以允许被丢弃

> 读者还可以查看RejectedExecutionHandler的子类，查看更多JDK自带实现，从而吸收其思想，相信自定义一个符合业务场景的拒绝策略实现并不是一件难事

## 关闭线程池 ##

在需要释放线程池中的线程资源时我们通常会调用以下方法关闭线程池，释放线程资源

*  shutdown：让线程先完成手头上的工作，再设置interrupt中断标识，结束时机交给工人决定
 *  shutdownNow：直接设置interrupt中断标识，结束时机依然是工人决定

这两个方法的区别仅仅是设置interrupt的时机，而真正的结束由Worker去判断interrupt标识，下面直接来看代码看看两个方法是如何实现的

### 温和关闭线程池 ###

首先看看shutdown方法，其关闭过程还是比较温柔的

public void shutdown() { 
      final ReentrantLock mainLock = this.mainLock;	
      mainLock.lock();
      try { 
        checkShutdownAccess();
        // 设置线程池状态为SHUTDOWN
        advanceRunState(SHUTDOWN);
        // 仅中断Idle（空闲）的工人
        interruptIdleWorkers();
        // 线程池关闭钩子（hook），空方法
        // 其在ScheduledThreadPoolExecutor线程池中有实现
        onShutdown(); // hook for ScheduledThreadPoolExecutor
      } finally { 
        mainLock.unlock();
      }
      tryTerminate();
    }

关键在interruptIdleWorkers方法，是如何判断哪些工人是空闲的

private void interruptIdleWorkers() { 
      interruptIdleWorkers(false);
    }
    
    private void interruptIdleWorkers(boolean onlyOne) { 
      final ReentrantLock mainLock = this.mainLock;
      mainLock.lock();
      try { 
        // 遍历工人集合
        for (Worker w : workers) { 
          Thread t = w.thread;
          // 如果工人还没有被中断，此时就来获取工人锁
          if (!t.isInterrupted() && w.tryLock()) { 
            try { 
              // 能到这里，表示工人此时是空闲的
              // 因为工人如果在忙，工人锁是会被工人独占的
              // 中断工人
              t.interrupt();
            } catch (SecurityException ignore) { 
            } finally { 
              w.unlock();
            }
          }
          if (onlyOne)
            break;
        }
      } finally { 
        mainLock.unlock();
      }
    }

我们可以回忆一下，在创建Worker并启动线程中，会启动worker线程的run方法，其本质是在一段while循环中不断获取任务并执行，在while循环中，若获取到了任务，则会将自身作为锁锁住，表示正在执行任务，若此时shutdown方法被调用，则获取此“忙工人”这把锁会被阻塞住，直到工人执行完这个任务，释放了工人锁才会被设置中断标识，等全部工人都被设置了中断标识后线程池就真正被关闭了。

### 暴力关闭线程池 ###

说是暴力，其实也不会特别的暴力，话不多说，直接进入shutdownNow方法

public List<Runnable> shutdownNow() { 
      List<Runnable> tasks;
      final ReentrantLock mainLock = this.mainLock;
      mainLock.lock();
      try { 
        checkShutdownAccess();
        // 直接设置为STOP
        advanceRunState(STOP);
        // 直接中断工人
        interruptWorkers();
        // 将没执行的任务返回出去
        tasks = drainQueue();
      } finally { 
        mainLock.unlock();
      }
      tryTerminate();
      return tasks;
    }

和上面不同的，这里会将还未执行的任务返回出去，并且中断是直接了当的

private void interruptWorkers() { 
      final ReentrantLock mainLock = this.mainLock;
      mainLock.lock();
      try { 
        for (Worker w : workers)
          // 直接中断，并不获取锁
          w.interruptIfStarted();
      } finally { 
        mainLock.unlock();
      }
    }

但归根结底，何时中断返回还是由工作线程决定的，若工作线程中没有判断中断标识的地方，则也不会立马返回，还是会执行完手头上的工作才会返回，若线程执行的方法中有阻塞线程的方法（大部分的阻塞方法都会被interrupt方法唤醒且判断标识，抛出一个中断异常），则此时会抛出异常，强行结束。

> 若线程中存在阻塞方法并且忽略中断异常，则此时线程永远也关闭不了！
> 
> 亦或是线程中有无限阻塞方法，若只调用shutdown是关闭不了的！
> 
> 综上所述，任务需要注意以下两点：
> 
> 1.  不能忽略中断异常！需要被合理的处理
> 2.  尽量避免无限阻塞方法（大概率也不会出现），若有这个可能，需要有意识转为调用shutdownNow来关闭线程池

## 小结 ##

到这里，线程池的流转就大致结束了，我这里为了总结线程池的工作原理，画了下面这张图，希望读者也可以跟着思路，也画出这么一张流转图，相信会更好的理解整个过程  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNzM3NzE2_size_16_color_FFFFFF_t_70 1]

# 配置一个合适的线程池 #

关于如何配置一个合适的线程池，其实这个话题很大很广，视不同业务，不同场景，不同机器配置而定，所以这里没有通用公式，只是介绍几个场景，希望读者可以触类旁通。

## 合适的线程数量 ##

一般来说：

*  CPU密集型：CPU总是会需要等待IO，假设CPU满载（CPU利用率100%，最理想情况），则此时为CPU密集型，可配置线程池数量为：Core核心数 + 1，可以在每个CPU核上各自跑一个线程，理想情况下没有上下文切换开销
 *  IO密集型：若一个任务的执行有很多的时间都在等待IO操作（例如数据库的IO、磁盘的IO、网络IO等等），此时称之为IO密集型。在执行IO操作时系统会将线程暂时休眠，切换别的线程继续执行任务，等IO操作完之后才会切换回来原线程继续执行，所以为了最大剥削CPU能力，在每个核上的每个时间都在跑线程，我们需要配置线程数量（当一半时间都在等待时）为：（Core核心数 \* 2 ）+ 1 ，这是由于等待时间占一半时的线程数量，**并不是IO密集型都是这个数量**，通用的，如果IO操作（会阻塞你线程的操作）比较多，你需要用以下公式计算合适的线程数：
    
    *任务执行时间 /（任务执行时间-IO等待时间）✖️ CPU内核数*
    
    例如：一个任务需要100ms执行完成，其中IO的时间需要花去50ms来等待，则：
    
    100 / （100 - 50） = 2，则最终需要两倍的CPU核心数
    
    若业务中还需要等待网络IO之类的不定的等待时间，有时候是50，有时候100，你可以取一个平均数

以上线程数是一个大致的估测值，并不能说是最终最合适的，你需要在这之后进行测试，然后再适当调整，一般来说在复杂场景下需要不断微调才可以知道大致的合适线程数量

## 合适的阻塞队列 ##

线程池配置怎么和阻塞队列还有关系？不同的阻塞队列都有不同的特点，其决定了线程池中任务的存取，也是比较关键的一个配置项，下面介绍几个常用的阻塞队列

*  在JDK 1.7 更新了一个新的队列实现 `LinkedTransferQueue` ，具体教程可以看[这里][Link 2]，其是以上几个队列的超集，又兼备ConcurrentLinkedQueue的吞吐量，不过此队列是无界的，而hand-off特性由新的api提供，如果线程池需要用到此队列的hand-off特性，需要自己另外封装api，建议读者可以了解并学习，将此队列列为首选
 *  SynchronousQueue：传球手，本身不存数据，若没有worker在take任务（表示没有空闲的线程），线程池在接受一个任务时会直接创建worker线程（因为其offer方法没有接收方，此方法不阻塞直接返回），这个阻塞队列的吞吐量是很高的，缺点就在会一直创建线程，在接受任务的时候（若设置了max，则会一直拒绝，不太合适）
 *  LinkedBlockingQueue：构造函数中若有传值则为有界队列，使用默认构造器则为无界队列，其可以存放任务元素，起到任务缓冲的作用，一般与ArrayBlockingQueue进行比较，其不需要分配一段连续内存，而是用链表形式保存一个链表引用，所以在不确定任务数量的情况下可以使用，没有初始化一大段内存的开销。也是很多一般线程池的选择
 *  ArrayBlockingQueue：有界队列，在初始化之后会分配一段额定的连续内存，其缺点就在刚开始就需要分配一段数组内存，但优点在于连续，可以很好的利用CPU缓存特性load一段连续内存作为缓存，在任务比较确定其吞吐量时可以使用，一般比较稳定

## 低耗时任务 ##

> 这里不论高并发，还是低并发，都可以一并来讲
> 
> 若使用SynchronousQueue或LinkedTransferQueue 吞吐量优先的队列，一般不太设置核心线程，随着并发量让线程数保持一个配合并发量的合适的值  
> 但要注意，使用LinkedTransferQueue的hand-off特性需要自己封装api，因为其handoff的api改变了

如果一个接口任务耗时比较低，那此时队列的吞吐量就显得比较重要了，所以会偏向使用SynchronousQueue或LinkedTransferQueue队列

如果并发量比较高，网关是需要控制并发量，做好适当的限流的，不然这个可怕的队列会让你的线程池无限创建线程，将会造成内存溢出的异常，导致系统崩溃。最好在做好限流的同时设置一个max线程数量，以防线程创建过多

## 高耗时任务 ##

> 这里需要考虑线程数量的配置

此时任务耗时比较高，需要知道是因为CPU计算量大而导致的高耗时，还是IO或其他阻塞动作导致的高耗时，如果是后者，需要使用ArrayBlockingQueue或是LinkedBlockingQueue这类的队列（熟悉的话首选LinkedTransferQueue，缺点是无界），然后合适的确定一个线程数量配置在线程池中，让线程池和队列很好管理和控制线程数量，做到最合适的吞吐。如果是前者，可能需要考虑算法的复杂度和CPU的核心数量的增加了，不然也顶不住这高并发…

## 定时任务 ##

线程数量：如果是定时任务，并不考虑即时性，配置1-2个线程都是可以的

回收线程资源：若场景以天、周等等大时间为单位，需要将线程池中的线程都设置为临时工，不然会一直占用系统内存等等资源，其只是定期跑一跑而已，没有必要一直占用资源。可以调用线程池的allowCoreThreadTimeOut方法，将所有线程都设置为临时工

> 当然了，以上结论都只是一个参考，给读者提供的一些思路，若死板的应用在项目中，是不会轻易达到最适合的要求的。这需要读者了解线程池运行原理，多配置几次积攒一些经验才可以更好的配置一个比较合适的线程池。

[Link 1]: https://blog.csdn.net/qq_41737716/article/details/100183193
[Link 2]: https://blog.csdn.net/qq_41737716/article/details/100561485
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNzM3NzE2_size_16_color_FFFFFF_t_70]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/17/652dfe0366ee4b8aa3cefa5e0147a1ca.png
[Link 3]: https://blog.csdn.net/qq_41737716/article/details/89524380
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNzM3NzE2_size_16_color_FFFFFF_t_70 1]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/17/ab84b64f6f1740ee85c5533fb61c6515.png