如何扩展和优化线程池？
1. Executors 线程池工厂能创建哪些线程池
先来一个最简单的线程池使用例子：
static class MyTask implements Runnable {
@Override
public void run() {
System.out
.println(System.currentTimeMillis() + ": Thread ID :" + Thread.currentThread().getId());
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
public static void main(String[] args) {
MyTask myTask = new MyTask();
ExecutorService service1 = Executors.newFixedThreadPool(5);
for (int i = 0; i < 10; i++) {
service1.submit(myTask);
}
service1.shutdown();
}运行结果：
我们创建了一个线程池实例，并设置默认线程数量为5，并向线程池提交了10任务，分别打印当前毫秒时间和线程ID，从结果中，我们可以看到结果中有5个相同 id 的线程打印了毫秒时间。
其它线程创建方式。
1. 固定线程池 ExecutorService service1 = Executors.newFixedThreadPool(5); 该方法返回一个固定线程数量的线程池。该线程池中的线程数量始终不变。
当有一个新的任务提交时，线程池中若有空闲线程，则立即执行，若没有，则新的任务会被暂存在一个任务队列（默认无界队列 int 最大数）中，待有线程空闲时，便处理在任务队列中的任务。
2. 单例线程池 ExecutorService service3 = Executors.newSingleThreadExecutor(); 该方法返回一个只有一个线程的线程池。
若多余一个任务被提交到该线程池，任务会被保存在一个任务队列（默认无界队列 int 最大数）中，待线程空闲，按先入先出的顺序执行队列中的任务。
3. 缓存线程池 ExecutorService service2 = Executors.newCachedThreadPool(); 该方法返回一个可根据实际情况调整线程数量的线程池，线程池的线程数量不确定。
但若有空闲线程可以复用，则会优先使用可复用的线程，所有线程均在工作，如果有新的任务提交，则会创建新的线程处理任务。所有线程在当前任务执行完毕后，将返回线程池进行复用。
4. 任务调用线程池 ExecutorService service4 = Executors.newScheduledThreadPool(2); 该方法也返回一个 ScheduledThreadPoolExecutor 对象，该线程池可以指定线程数量。
前3个线程的用法没什么差异，关键是第四个，虽然线程任务调度框架很多，但是我们仍然可以学习该线程池。如何使用呢？下面来个例子：
class A {
public static void main(String[] args) {
ScheduledThreadPoolExecutor service4 = (ScheduledThreadPoolExecutor) Executors
.newScheduledThreadPool(2);
// 如果前面的任务没有完成，则调度也不会启动
service4.scheduleAtFixedRate(new Runnable() {
@Override
public void run() {
try {
// 如果任务执行时间大于间隔时间，那么就以执行时间为准（防止任务出现堆叠）。
Thread.sleep(10000);
System.out.println(System.currentTimeMillis() / 1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}// initialDelay（初始延迟） 表示第一次延时时间 ; period 表示间隔时间
}, 0, 2, TimeUnit.SECONDS);
service4.scheduleWithFixedDelay(new Runnable() {
@Override
public void run() {
try {
Thread.sleep(5000);
System.out.println(System.currentTimeMillis() / 1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}// initialDelay（初始延迟） 表示延时时间；delay + 任务执行时间 = 等于间隔时间 period
}, 0, 2, TimeUnit.SECONDS);
// 在给定时间，对任务进行一次调度
service4.schedule(new Runnable() {
@Override
public void run() {
System.out.println("5 秒之后执行 schedule");
}
}, 5, TimeUnit.SECONDS);
}
}}上面的代码创建了一个 ScheduledThreadPoolExecutor 任务调度线程池，分别调用了3个方法，需要着重解释 scheduleAtFixedRate 和 scheduleWithFixedDelay 方法。
这两个方法的作用很相似，唯一的区别就是他们执行人物的间隔时间的计算方式，前者时间间隔算法是根据指定的 period 时间和任务执行时间中取时间长的，后者取的是指定的 delay 时间 + 任务执行时间。
好了，JDK 给我们封装了创建线程池的 4 个方法，但是，请注意，由于这些方法高度封装，因此，如果使用不当，出了问题将无从排查，因此，我建议，程序员应到自己手动创建线程池，而手动创建的前提就是高度了解线程池的参数设置。
2. 如何手动创建线程池
下面是一个手动创建线程池的范本：
/**
* 默认5条线程（默认数量，即最少数量），
* 最大20线程（指定了线程池中的最大线程数量），
* 空闲时间0秒（当线程池梳理超过核心数量时，多余的空闲时间的存活时间，即超过核心线程数量的空闲线程，在多长时间内，会被销毁），
* 等待队列长度1024，
* 线程名称[MXR-Task-%d],方便回溯，
* 拒绝策略：当任务队列已满，抛出RejectedExecutionException
* 异常。
*/
private static ThreadPoolExecutor threadPool = new ThreadPoolExecutor(5, 20, 0L,
TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(1024)
, new ThreadFactoryBuilder().setNameFormat("My-Task-%d").build()
, new AbortPolicy()
);ThreadPoolExecutor 也就是线程池有 7 个参数如下：
-
corePoolSize 线程池中核心线程数量
-
maximumPoolSize 最大线程数量
-
keepAliveTime 空闲时间（当线程池梳理超过核心数量时，多余的空闲时间的存活时间，即超过核心线程数量的空闲线程，在多长时间内，会被销毁）
-
unit 时间单位
-
workQueue 当核心线程工作已满，需要存储任务的队列
-
threadFactory 创建线程的工厂
-
handler 当队列满了之后的拒绝策略
前面几个参数我们就不讲了，很简单，主要是后面几个参数，队列，线程工厂，拒绝策略。
我们先看看队列，线程池默认提供了 4 个队列：
-
无界队列： 默认大小 int 最大值，因此可能会耗尽系统内存，引起OOM，非常危险。
-
直接提交的队列 ： 没有容量，不会保存，直接创建新的线程，因此需要设置很大的线程池数。否则容易执行拒绝策略，也很危险。
-
有界队列：如果core满了，则存储在队列中，如果core满了且队列满了，则创建线程，直到maximumPoolSize 到了，如果队列满了且最大线程数已经到了，则执行拒绝策略。
-
优先级队列：按照优先级执行任务。也可以设置大小。
在自己的项目中使用了无界队列，但是设置了任务大小，1024。如果你的任务很多，建议分为多个线程池。不要把鸡蛋放在一个篮子里。
再看看拒绝策略，什么是拒绝策略呢？当队列满了，如何处理那些仍然提交的任务。JDK 默认有4种策略。
-
AbortPolicy ：直接抛出异常，阻止系统正常工作.
-
CallerRunsPolicy : 只要线程池未关闭，该策略直接在调用者线程中，运行当前被丢弃的任务。显然这样做不会真的丢弃任务，但是，任务提交线程的性能极有可能会急剧下降。
-
DiscardOldestPolicy: 该策略将丢弃最老的一个请求，也就是即将被执行的一个任务，并尝试再次提交当前任务.
-
DiscardPolicy: 该策略默默地丢弃无法处理的任务，不予任何处理，如果允许任务丢失，我觉得这是最好的方案..
当然，如果你不满意JDK提供的拒绝策略，只需要实现 RejectedExecutionHandler 接口，并重写 rejectedExecution 方法即可。
最后，线程工厂，线程池的所有线程都由线程工厂来创建，而默认的线程工厂太过单一
代码：
/**
* The default thread factory
*/
static class DefaultThreadFactory implements ThreadFactory {
private static final AtomicInteger poolNumber = new AtomicInteger(1);
private final ThreadGroup group;
private final AtomicInteger threadNumber = new AtomicInteger(1);
private final String namePrefix;
DefaultThreadFactory() {
SecurityManager s = System.getSecurityManager();
group = (s != null) ? s.getThreadGroup() :
Thread.currentThread().getThreadGroup();
namePrefix = "pool-" +
poolNumber.getAndIncrement() +
"-thread-";
}
public Thread newThread(Runnable r) {
Thread t = new Thread(group, r,
namePrefix + threadNumber.getAndIncrement(),
0);
if (t.isDaemon())
t.setDaemon(false);
if (t.getPriority() != Thread.NORM_PRIORITY)
t.setPriority(Thread.NORM_PRIORITY);
return t;
}
}可以看到，线程名称为 pool- + 线程池编号 + -thread- + 线程编号 。设置为非守护线程。优先级为默认。
如果我们想修改名称呢？对，实现 ThreadFactory 接口，重写 newThread 方法即可。
但是已经有人造好轮子了， 比如我们的例子中使用的 google 的 guaua 提供的 ThreadFactoryBuilder 工厂。可以自定义线程名称，是否守护，优先级，异常处理等等，功能强大。
3. 如何扩展线程池
那么我们能扩展线程池的功能吗？比如记录线程任务的执行时间。实际上，JDK 的线程池已经为我们预留的接口，在线程池核心方法中，有2 个方法是空的，就是给我们预留的。还有一个线程池退出时会调用的方法。
代码：
/**
* 如何扩展线程池，重写 beforeExecute, afterExecute, terminated 方法，这三个方法默认是空的。
*
* 可以监控每个线程任务执行的开始和结束时间，或者自定义一些增强。
*
* 在 Worker 的 runWork 方法中，会调用这些方法
*/public class ExtendThreadPoolDemo {
static class MyTask implements Runnable {
String name;
public MyTask(String name) {
this.name = name;
}
@Override
public void run() {
System.out
.println("正在执行：Thread ID:" + Thread.currentThread().getId() + ", Task Name = " + name);
try {
Thread.sleep(100);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
public static void main(String[] args) throws InterruptedException {
ExecutorService es = new ThreadPoolExecutor(5, 5, 0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<>()) {
@Override
protected void beforeExecute(Thread t, Runnable r) {
System.out.println("准备执行：" + ((MyTask) r).name);
}
@Override
protected void afterExecute(Runnable r, Throwable t) {
System.out.println("执行完成： " + ((MyTask) r).name);
}
@Override
protected void terminated() {
System.out.println("线程池退出");
}
};
for (int i = 0; i < 5; i++) {
MyTask myTask = new MyTask("TASK-GEYM-" + i);
es.execute(myTask);
Thread.sleep(10);
}
es.shutdown();
}}我们重写了 beforeExecute 方法，也就是执行任务之前会调用该方法，而 afterExecute 方法则是在任务执行完毕后会调用该方法。
还有一个 terminated 方法，在线程池退出时会调用该方法。执行结果是什么呢？
可以看到，每个任务执行前后都会调用 before 和 after 方法。相当于执行了一个切面。而在调用 shutdown 方法后则会调用 terminated 方法。
4. 如何优化线程池的异常信息
如何优化线程池的异常信息？ 在说这个问题之前，我们先说一个不容易发现的bug：
看代码：
public static void main(String[] args) throws ExecutionException, InterruptedException {
ThreadPoolExecutor executor = new ThreadPoolExecutor(0, Integer.MAX_VALUE, 0L,
TimeUnit.MILLISECONDS, new SynchronousQueue<>());
for (int i = 0; i < 5; i++) {
executor.submit(new DivTask(100, i));
}
}
static class DivTask implements Runnable {
int a, b;
public DivTask(int a, int b) {
this.a = a;
this.b = b;
}
@Override
public void run() {
double re = a / b;
System.out.println(re);
}
}执行结果：
注意：只有4个结果，其中一个结果被吞没了，并且没有任何信息。为什么呢？如果仔细看代码，会发现，在进行 100 / 0 的时候肯定会报错的，但是却没有报错信息。
实际上，如果你使用 execute 方法则会打印错误信息，当你使用 submit 方法却没有调用它的get 方法，异常将会被吞没，因为，如果发生了异常，异常是作为返回值返回的。
怎么办呢？我们当然可以使用 execute 方法，但是我们可以有另一种方式：重写 submit 方法，本人写了一个例子，大家看一下：
static class TraceThreadPoolExecutor extends ThreadPoolExecutor {
public TraceThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime,
TimeUnit unit, BlockingQueue<Runnable> workQueue) {
super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
}
@Override
public void execute(Runnable command) {// super.execute(command);
super.execute(wrap(command, clientTrace(), Thread.currentThread().getName()));
}
@Override
public Future<?> submit(Runnable task) {// return super.submit(task);
return super.submit(wrap(task, clientTrace(), Thread.currentThread().getName()));
}
private Exception clientTrace() {
return new Exception("Client stack trace");
}
private Runnable wrap(final Runnable task, final Exception clientStack,
String clientThreaName) {
return new Runnable() {
@Override
public void run() {
try {
task.run();
} catch (Exception e) {
e.printStackTrace();
clientStack.printStackTrace();
throw e;
}
}
};
}
}我们重写了 submit 方法，封装了异常信息，如果发生了异常，将会打印堆栈信息。我们看看使用重写后的线程池后的结果是什么？
从结果中，我们清楚的看到了错误信息的原因：by zero！并且堆栈信息明确，方便排错。优化了默认线程池的策略。
5. 如何设计线程池中的线程数量
线程池的大小对系统的性能有一定的影响，过大或者过小的线程数量都无法发挥最优的系统性能，但是线程池大小的确定也不需要做的非常精确。
因为只要避免极大和极小两种情况，线程池的大小对性能的影响都不会影响太大，一般来说，确定线程池的大小需要考虑CPU数量，内存大小等因素，在《Java Concurrency in Practice》 书中给出了一个估算线程池大小的经验公式：
在服务器性能IO优化 中发现一个估算公式：
最佳线程数目 = （（线程等待时间+线程CPU时间）/线程CPU时间 ）* CPU数目
比如平均每个线程CPU运行时间为0.5s，而线程等待时间（非CPU运行时间，比如IO）为1.5s，CPU核心数为8，那么根据上面这个公式估算得到：((0.5+1.5)/0.5)*8=32。这个公式进一步转化为：
最佳线程数目 = （线程等待时间与线程CPU时间之比 + 1）* CPU数目
可以得出一个结论：
线程等待时间所占比例越高，需要越多线程。线程CPU时间所占比例越高，需要越少线程。
公式还是有点复杂的，简单来说，就是如果你是CPU密集型运算，那么线程数量和CPU核心数相同就好，避免了大量无用的切换线程上下文，如果你是IO密集型的话，需要大量等待，那么线程数可以设置的多一些，比如CPU核心乘以2。
至于如何获取 CPU 核心数，Java 提供了一个方法：
Runtime.getRuntime().availableProcessors()；
返回了CPU的核心数量。
拓展推荐：
https://blog.****.net/zhongxiangbo/article/details/70882309?utm_source=blogxgwz1
https://blog.****.net/cby1516/article/details/80582822?utm_source=blogxgwz48