postgres 支持的线程数_线程池被打满了怎么处理呢，你是否真的了解线程池？

0、前言

线程池，顾名思义就是线程的池子，在每次需要取线程去执行任务的时候，没必要每次都创建新线程执行，线程池就是起着维护线程的作用，当有任务的时候就取出一个线程执行，如果任务执行完成则把线程放回到池子中，完成线程的复用，没必要每次都去创建和销毁现场，提高效率。

线程池为线程的生命周期的开销和资源不足提供的解决方案。通过对多个任务的重用线程

那么什么是时候使用多线程呢？

单个任务处理时间比较短
处理的任务数比较大，创建和销毁线程消耗资源过多

1、Exector 接口

Exector 接口：运行新任务的简单接口
ExectorService 接口：扩展了Exector接口，添加了一些用来管理执行器生命周期和任务生命周期的方法
ScheduledExecutorService 接口：扩展自ExectorService接口，支持Future和定期执行任务
Exectors 类：包装了具体的几个常用的线程池的定义，便于使用

2、ThreadPoolExecutor 类

java的线程都是在JUC包中，其中java.uitl.concurrent.ThreadPoolExecutor是最为核心的类

其包中包含的四种创建线程池的方法

public class ThreadPoolExecutor extends AbstractExecutorService { ..... public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit, BlockingQueue workQueue);  public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit, BlockingQueue workQueue,ThreadFactory threadFactory);  public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit, BlockingQueue workQueue,RejectedExecutionHandler handler);  public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit, BlockingQueue workQueue,ThreadFactory threadFactory,RejectedExecutionHandler handler); ...}

具体的调用方法又可以是在Exectors类中，只是提供更加便利的线程池调用方法，如果希望使用更加灵活自定义的线程池，还是建议使用ThreadPoolExecutor

注意到默认由executors创建的线程池的工作队列都是基于链表的阻塞队列，没有具体长度

public class Executors { public static ExecutorService newFixedThreadPool(int nThreads) { return new ThreadPoolExecutor(nThreads, nThreads, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue()); } public static ExecutorService newFixedThreadPool(int nThreads, ThreadFactory threadFactory) { return new ThreadPoolExecutor(nThreads, nThreads, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue(), threadFactory); }  public static ExecutorService newSingleThreadExecutor() { return new FinalizableDelegatedExecutorService (new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue())); }  public static ExecutorService newSingleThreadExecutor(ThreadFactory threadFactory) { return new FinalizableDelegatedExecutorService (new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue(), threadFactory)); }  public static ExecutorService newCachedThreadPool() { return new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS, new SynchronousQueue()); }  public static ExecutorService newCachedThreadPool(ThreadFactory threadFactory) { return new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS, new SynchronousQueue(), threadFactory); }

2.1 参数说明

private volatile int corePoolSize;:核心池大小，当当前线程数小于核心池数字，则新加一个任务就创建一个线程处理，当超过了核心池大小则加入到任务队列中,在创建了线程池后，默认情况下，线程池中并没有任何线程，而是等待有任务到来才创建线程去执行任务，除非调用了prestartAllCoreThreads()或者prestartCoreThread()方法，从这2个方法的名字就可以看出，是预创建线程的意思，即在没有任务到来之前就创建corePoolSize个线程或者一个线程。默认情况下，在创建了线程池后，线程池中的线程数为0，当有任务来之后，就会创建一个线程去执行任务，当线程池中的线程数目达到corePoolSize后，就会把到达的任务放到缓存队列当中
private volatile int maximumPoolSize;:最大线程数，线程池最大能创建的线程数目
private volatile long keepAliveTime;:线程没有需要处理的任务时最多保持多久会终止，默认情况下，只有当线程池中的线程数大于corePoolSize时，keepAliveTime才会起作用，直到线程池中的线程数不大于corePoolSize，即当线程池中的线程数大于corePoolSize时，如果一个线程空闲的时间达到keepAliveTime，则会终止，直到线程池中的线程数不超过corePoolSize。但是如果调用了allowCoreThreadTimeOut(boolean)方法，在线程池中的线程数不大于corePoolSize时，keepAliveTime参数也会起作用，直到线程池中的线程数为0

什么时候会出现核心线程数呢？如果使用了默认的Executors类提供的默认的线程池生产方法，则生成的任务队列都是链表阻塞队列，都是没有长度限制的，那么这样也就暗示了使用链表阻塞队列的线程池，线程池个数最多是核心线程数个，列表阻塞队列才有可能出现填充满了的情况，才进而出现实际线程个数超过核心线程数的情况

private volatile ThreadFactory threadFactory;:创建线程的工厂bean，其中Executors.defaultThreadFactory()提供默认的线程工厂bean
private final BlockingQueue workQueue;:任务队列
private volatile RejectedExecutionHandler handler:拒绝处理任务时的策略

ThreadPoolExecutor.AbortPolicy:丢弃任务并抛出RejectedExecutionException异常。
ThreadPoolExecutor.DiscardPolicy：也是丢弃任务，但是不抛出异常。
ThreadPoolExecutor.DiscardOldestPolicy：丢弃队列最前面的任务，然后重新尝试执行任务(重复此过程)

ThreadPoolExecutor.CallerRunsPolicy：由调用线程处理该任务

2.2 重要方法

execute()

Executor中声明的方法，在ThreadPoolExecutor进行了具体的实现，这个方法是ThreadPoolExecutor的核心方法，通过这个方法可以向线程池提交一个任务，交由线程池去执行。

submit()

ExecutorService中声明的方法，在AbstractExecutorService就已经有了具体的实现，在ThreadPoolExecutor中并没有对其进行重写，这个方法也是用来向线程池提交任务的，但是它和execute()方法不同，它能够返回任务执行的结果，去看submit()方法的实现，会发现它实际上还是调用的execute()方法

shutdown()

关闭线程池的方法，不会接受新的任务，会等到已经存在的任务执行完成

shutdownNow()

关闭线程池的方法，不会接受新的任务，已经运行的任务也会被中断执行

注意！当线程处于关闭、关闭中的状态时，不会再接收新的任务，接收到的任务也是会被直接拒绝的，后续由拒绝Policy执行调用

3、线程池实现原理

3.1 线程池状态

线程池的状态，表示当前线程池的工作状态，例如会不会接受新的任务，之前存在的任务是如何处理的

 private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0)); private static final int COUNT_BITS = Integer.SIZE - 3; 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; private static final int SHUTDOWN = 0 << COUNT_BITS; private static final int STOP = 1 << COUNT_BITS; private static final int TIDYING = 2 << COUNT_BITS; private static final int TERMINATED = 3 << COUNT_BITS; // Packing and unpacking ctl private static int runStateOf(int c) { return c & ~CAPACITY; } private static int workerCountOf(int c) { return c & CAPACITY; } private static int ctlOf(int rs, int wc) { return rs | wc; }

ctl字段是组合为线程池的运行状态和有效线程数异或生成的，高三位保存了runState(运行状态)，低29位保存了workerCount(工作线程个数)

RUNNING : 能接受新提交的任务，并且能处理阻塞队列中的任务
SHUTDOWN : 关闭线程池，不再接受新提交的任务，但是可以继续处理在阻塞线程中已经存在的任务，线程在 RUNNING 状态调用shutdown()方法进入到SHUTDOWN状态
STOP : 不接受新任务，也不处理阻塞队列的任务，中断正在处理的线程，线程处于 RUNNING 或者 SHUTDOWN 状态调用 shutdownNow()方法进入到STOP 状态
TIDYING 所有的任务都终止了，workCount为0，线程池如果处于该情况下，调用terminated()方法，进入到 TERMINATED 状态
TERMINATED TODO

3.2 任务添加执行

3.2.1 execute(Runnable command)

 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(); if (workerCountOf(c) < corePoolSize) { if (addWorker(command, true)) return; c = ctl.get(); } 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); }

1、当前有效线程数小于核心线程数时，应该添加一个新线程addWorker(command, true)，成功则退出，否则继续进程

2、再次获取ctl值(状态、线程数可能发生变化)

3、如果线程池处于运行状态，并且当前任务成功加入到workQueue中，再次进行double-check，理由同上

如果发现线程池不再处于运行状态，但是任务已经添加到队列中了，尝试使用remove移除，如果移除成功，(存在已经被执行的情况)就拒绝该任务(这里就是上面说的当前状态是关闭时或者关闭，则拒绝该任务)
否则当当前有效线程数为0时，创建一个空的线程，会自动从workQueue中获取任务去执行

4、运行最后一个else则有如下情况

线程不处于RUNNING状态
线程是RUNNING状态，但是工作线程数目已经超过了核心线程数，而且阻塞队列已经满了
这个时候通过调用addWorker(command, false),false表示线程上线设置为最大线程数去添加该任务

5、如果没有正常添加，则拒绝该任务

这就是整个的线程池工作流程，先是核心线程，再是任务队列，最后是最大线程

3.2.2 addWorker(Runnable firstTask, boolean core)

创建一个新线程去执行任务

firstTask表示需要执行的任务，如果为null，则是单独的创建一个新线程

core 则是判断线程数是否需要超过核心线程数目，true表示新增线程数前判断是否小于核心线程数，false表示新增线程数前判断是否小于最大线程数

 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;  /* 如果状态 > SHUTDOWN 直接返回false，此时不执行新任务，也不执行阻塞队列的任务 但是 在 = SHUTDOWN的状态，不会处理新任务，还是可以继续执行在阻塞队列的任务 情况列为如下三种 1、如果状态 > SHUTDOWN 直接返回false 2、如果状态 = SHUTDOWN 而且 firstTask 不为 null 直接返回false(拒绝处理) 3、如果状态 = SHUTDOWN 而且 firstTask 为 null 而且workQueue为空 直接返回false(workQueue都已经没有数据了，也就没必要创建新的空工作线程) */  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 }  /* 很明显采用的是经典的**死循环加CAS模式**去判断当前是否可以新加线程 如果当前工作线程数目超过了最大值，就返回false，不再处理 采取CAS去添加线程数目，如果成功就跳出第一个for循环，这里需要注意下，这里**线程池状态可能会发生变化**，所以有如下代码  ```java private boolean compareAndIncrementWorkerCount(int expect) { return ctl.compareAndSet(expect, expect + 1); } ``` 在进行CAS操作的时候，是针对整个的ctl值进行操作，如果成功，**肯定能够确保线程状态没有发生变化**！  CAS失败，再次获取ctl值，如果当前线程池状态不等于rs，则说明线程池状态发生变化，需要从for循环重新开始，否则一直在这个for循环内运行  for循环break之后就是表示当前任务可以被创建成一个线程去执行操作了  */ } boolean workerStarted = false; boolean workerAdded = false; Worker w = null; try { w = new Worker(firstTask); /* ```java Worker(Runnable firstTask) { setState(-1); // inhibit interrupts until runWorker this.firstTask = firstTask; this.thread = getThreadFactory().newThread(this); } ``` 新建一个Worker对象，其中Thread线程是通过线程工厂bean调用生成的 */ final Thread t = w.thread; if (t != null) { final ReentrantLock mainLock = this.mainLock; mainLock.lock(); // 利用可重入的排他锁，确保线程安全 // 需要注意到添加了锁确实不错，但是只是这线程池一块添加了锁，其他没有加锁的部分依旧可用 // 如果看线程池的showdown和shutdownnow方法，都使用了该锁，那也就意味着，进入到这里 // 就不会发生线程池关闭的情况了，AQS已经保证了该请求操作的安全性 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)) { // 如果线程池在RUNNING 运行状态或者 // 在 SHUTDOWN 而且 firstTask为null，则需要创建一个新线程 // 其实在这里有个疑问，为什么在shutdown了之后，不检查workQueue的长度？ if (t.isAlive()) // precheck that t is startable // 如果该线程已经存活，则抛出异常，肯定不正常，线程还是刚创建，压根没启动 throw new IllegalThreadStateException(); workers.add(w); // workers是一个HashSet() int s = workers.size(); if (s > largestPoolSize) largestPoolSize = s; // largestPoolSize 表示线程池中存在的最大线程数的情况 workerAdded = true;  // 往工作线程中加入了新的worker成功 } } finally { mainLock.unlock(); // 锁被释放，其他的线程可以进入了 } if (workerAdded) { t.start(); // 这里才是真正的线程启动的地方，由于Worker本身是继承Runnable // 所以这个start最终运行的也是Worker的run方法，也就是`runWorker`方法 workerStarted = true; } } } finally { if (! workerStarted) // 没有正常添加线程到worker中或者线程启动失败 addWorkerFailed(w); /* private void addWorkerFailed(Worker w) { final ReentrantLock mainLock = this.mainLock; mainLock.lock(); // 依旧可能存在线程安全问题 // 不得不说，这代码对线程安全的粒度做的确实很细 try { if (w != null) workers.remove(w); decrementWorkerCount(); tryTerminate(); // 尝试更新线程池状态到TERMINATED 状态 } finally { mainLock.unlock(); } } */ } return workerStarted; }

3.2.3 runWorker(Worker w)

新创建了一个worker，然后执行，如果worker中不包含对于的任务，可以从阻塞队列中获取，一旦一个work开始启动了，如果workQueue中一直存在数据，则其不会退出，会在while中一直循环的

 final void runWorker(Worker w) { Thread wt = Thread.currentThread(); // 当前需要执行的线程 Runnable task = w.firstTask; w.firstTask = null; w.unlock(); // allow interrupts // 解绑workder和当前任务的关系，需要执行的是task任务 // 线程已经创建完成wt，wt需要执行task boolean completedAbruptly = true; try { while (task != null || (task = getTask()) != null) { // 这是一个死循环，直到无法获得到有效的task 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(); // STOP 状态 是调用了 shutdownNow() 之后的线程池状态 try { beforeExecute(wt, task); // 这个可以自定义加一些线程监控mirror 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++; // 当前这个workder有效完成的任务数 w.unlock(); } } completedAbruptly = false; } finally { processWorkerExit(w, completedAbruptly); } }

3.2.4 getTask()

从阻塞队列workQueue中获取有效的任务

 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; } /* 1、如果当前线程并未STOP，但是却进入了SHUTDOWN状态，而且阻塞队列也没有任何，则减少workers数目，并且返回null 2、如果当前线程池>=STOP,则直接返回返回null */  int wc = workerCountOf(c); // Are workers subject to culling? boolean timed = allowCoreThreadTimeOut || wc > corePoolSize; /* 这个allowCoreThreadTimeOut 就是配套 keepAliveTime 使用 /** * If false (default), core threads stay alive even when idle. * If true, core threads use keepAliveTime to time out waiting * for work. */ private volatile boolean allowCoreThreadTimeOut; */ if ((wc > maximumPoolSize || (timed && timedOut)) && (wc > 1 || workQueue.isEmpty())) { if (compareAndDecrementWorkerCount(c)) return null; continue; } /* wc > maximumPoolSize 工作线程数大于最大线程数，可能是在此时之前设置了最大线程数 (timed && timedOut) 需要设置超时 如果这种情况下，线程数>1或者阻塞队列为空，则需要减少一个worker 成功了，则返回null，无有效task  线程池中无论是否存在有效任务，必须是存在一个线程在 */ try { Runnable r = timed ? workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) : workQueue.take(); // 从阻塞队列中获取任务，可能是需要设置超时，如果一旦时间超时，则需要在规定的时间内获取到任务 if (r != null) return r; // 这一步也感觉很妙，如果没有有效的获取到任务，(可能是这个时候阻塞队列为空，也有可能是因为没有设置超时而没有获取到)，自动的设置这个timeOut为true，加上超时设置 timedOut = true; } catch (InterruptedException retry) { // 如果发生了中断，则回复这个timeOut设置，进行重试操作 timedOut = false; } } }

至此核心的函数都已经介绍完毕，提交任务，可以立刻处理任务就立马创建线程处理，如果当前无法马上处理则加入到阻塞队列中，如果加入到阻塞队列都失败，则调用相关的拒绝策略处理任务

Worker作为处理线程任务则一直在循环接受到新的任务或者从阻塞队列获取任务

3.3 其他方法

3.3.1 processWorkerExit(Worker w, boolean completedAbruptly)

此方法是在runWorker中被调用的，当runWorker无法获取到有效执行的task或者执行任务期间抛出各种异常导致的此worker无法在继续执行下去，该worker需要被消耗

 private void processWorkerExit(Worker w, boolean completedAbruptly) { if (completedAbruptly) // If abrupt, then workerCount wasn't adjusted // 如果这个值为true，则不是没有task导致的，而是在执行任务期间出现问题，故减1 // 如果是正常执行到这里的，则意味着task=null，在getTask方法中已经进行减一操作的 decrementWorkerCount(); final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { // worker是HashSet ，非线程安全，加锁保证安全 completedTaskCount += w.completedTasks; workers.remove(w); // 移除该worker，并汇总共完成的有效任务数，当然这需要确保线程安全 } finally { mainLock.unlock(); } tryTerminate(); // 尝试使得线程池状态变更为TERMINATED int c = ctl.get(); if (runStateLessThan(c, STOP)) { // 线程池状态位RUNNING 或者 SHUTDOWN  if (!completedAbruptly) { // 如果当前worker是由于没有有效task而退出的 int min = allowCoreThreadTimeOut ? 0 : corePoolSize; if (min == 0 && ! workQueue.isEmpty()) // 如果设置了超时，而且有工作队列还有任务存在，则最少保存min=1个worker min = 1; // 这里min=1或者corePoolSize if (workerCountOf(c) >= min) // 当前的工作线程个数超过了min，那么就无需操作工作线程格式，否则需要添加一个空的工作线程，所以min=1 return; // replacement not needed } // 直接创建一个新的空的worker // 防止工作线程 减少到0了，在  addWorker(null, false); } }

3.3.2 tryTerminate()

尝试让线程池状态变更为 TERMINATED，结束线程池

 final void tryTerminate() { for (;;) { int c = ctl.get(); /* 1、线程池在RUNNING 阶段 2、线程池在TIDYING 或者 TERMINATED 阶段 3、线程池在SHUTDOWN 而且工作线程不为空  以上几种情况都直接返回，所有的情况都未发生变化 */ if (isRunning(c) || runStateAtLeast(c, TIDYING) || (runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty())) return;  // 工作线程数不为0，则中断一个空闲的工作线程(因为工作线程每次工作时都会获取锁) if (workerCountOf(c) != 0) { // Eligible to terminate interruptIdleWorkers(ONLY_ONE); return; } // 到这里来的可能就是SHUTDOWN 而且工作线程为空的情况  final ReentrantLock mainLock = this.mainLock; mainLock.lock(); // 加锁操作 try { if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) { // 状态和工作线程数变更成功 try { terminated(); // 这是一个空方法，可以自定义实现 } finally { ctl.set(ctlOf(TERMINATED, 0)); termination.signalAll(); } return; } } finally { mainLock.unlock(); } // else retry on failed CAS } }

3.3.3 interruptIdleWorkers(boolean onlyOne)

中断空闲的线程，如果onlyOne为true，则中断其中的一个

 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()) { // 如果t线程不处于中断状态，尝试获取线程成功 // 能获取锁成功，至少能够说明其没有进行工作，也就是一个空闲的工作线程 try { // 调用中断操作 t.interrupt(); } catch (SecurityException ignore) { } finally { w.unlock(); } } if (onlyOne) break; } } finally { mainLock.unlock(); } }

3.4 线程池初始化

如上代码所示，起初创建线程池的时候，是不包含任何线程的，但是可以人为的创建线程

// 创建一个线程public boolean prestartCoreThread() { return workerCountOf(ctl.get()) < corePoolSize && addWorker(null, true); // 添加一个没有任务的空现场}// 创建核心线程数目个线程，直到addWorker返回falsepublic int prestartAllCoreThreads() { int n = 0; while (addWorker(null, true)) ++n; return n;}

3.5 任务队列

workQueue的类型为BlockingQueue，通常可以取下面三种类型：