如何扩展和优化线程池？

那么，怎么才能最大限度的利用CPU的性能，又能保持系统的稳定性呢？其中有一个方法就是使用线程池。

简而言之，在使用线程池后，创建线程便处理从线程池获得空闲线程，关闭线程变成了向池子归还线程。也就是说，提高了线程的复用。

而 JDK 在 1.5 之后为我提供了现成的线程池工具，我们今天就来学习看看如何使用他们。

Executors 线程池工厂能创建哪些线程池

如何手动创建线程池

如何扩展线程池

如何优化线程池的异常信息

如何设计线程池中的线程数量

1. Executors 线程池工厂能创建哪些线程池

先来一个最简单的线程池使用例子：

1. static class MyTask implements Runnable {

2. 
   

3. @Override

4. public void run() {

5. System.out

6. .println(System.currentTimeMillis() + ": Thread ID :" + Thread.currentThread().getId());

7. try {

8. Thread.sleep(1000);

9. } catch (InterruptedException e) {

10. e.printStackTrace();

11. }

12. }

13. }

14. 
    

15. public static void main(String[] args) {

16. MyTask myTask = new MyTask();

17. ExecutorService service1 = Executors.newFixedThreadPool(5);

18. for (int i = 0; i < 10; i++) {

19. service1.submit(myTask);

20. }

21. service1.shutdown();

22. }

运行结果：

我们创建了一个线程池实例，并设置默认线程数量为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个线程的用法没什么差异，关键是第四个，虽然线程任务调度框架很多，但是我们仍然可以学习该线程池。如何使用呢？下面来个例子：

1. class A {

2. 
   

3. public static void main(String[] args) {

4. ScheduledThreadPoolExecutor service4 = (ScheduledThreadPoolExecutor) Executors

5. .newScheduledThreadPool(2);

6. 
   

7. // 如果前面的任务没有完成，则调度也不会启动

8. service4.scheduleAtFixedRate(new Runnable() {

9. @Override

10. public void run() {

11. try {

12. // 如果任务执行时间大于间隔时间，那么就以执行时间为准（防止任务出现堆叠）。

13. Thread.sleep(10000);

14. System.out.println(System.currentTimeMillis() / 1000);

15. } catch (InterruptedException e) {

16. e.printStackTrace();

17. }

18. }// initialDelay（初始延迟） 表示第一次延时时间 ; period 表示间隔时间

19. }, 0, 2, TimeUnit.SECONDS);

20. 
    

21. service4.scheduleWithFixedDelay(new Runnable() {

22. @Override

23. public void run() {

24. try {

25. Thread.sleep(5000);

26. System.out.println(System.currentTimeMillis() / 1000);

27. } catch (InterruptedException e) {

28. e.printStackTrace();

29. }

30. }// initialDelay（初始延迟） 表示延时时间；delay + 任务执行时间 = 等于间隔时间 period

31. }, 0, 2, TimeUnit.SECONDS);

32. 
    

33. // 在给定时间，对任务进行一次调度

34. service4.schedule(new Runnable() {

35. @Override

36. public void run() {

37. System.out.println("5 秒之后执行 schedule");

38. }

39. }, 5, TimeUnit.SECONDS);

40. }

41. }

42. 
    

43. }

上面的代码创建了一个 ScheduledThreadPoolExecutor 任务调度线程池，分别调用了3个方法，需要着重解释 scheduleAtFixedRate 和 scheduleWithFixedDelay 方法，这两个方法的作用很相似，唯一的区别就是他们执行人物的间隔时间的计算方式，前者时间间隔算法是根据指定的 period 时间和任务执行时间中取时间长的，后者取的是指定的 delay 时间 + 任务执行时间。如果同学们有兴趣，可以将上面的代码跑跑看。一样便能看出端倪。

好了，JDK 给我们封装了创建线程池的 4 个方法，但是，请注意，由于这些方法高度封装，因此，如果使用不当，出了问题将无从排查，因此，我建议，程序员应到自己手动创建线程池，而手动创建的前提就是高度了解线程池的参数设置。那么我们就来看看如何手动创建线程池。

2. 如何手动创建线程池

下面是一个手动创建线程池的范本：

1. /**

2. * 默认5条线程（默认数量，即最少数量），

3. * 最大20线程（指定了线程池中的最大线程数量），

4. * 空闲时间0秒（当线程池梳理超过核心数量时，多余的空闲时间的存活时间，即超过核心线程数量的空闲线程，在多长时间内，会被销毁），

5. * 等待队列长度1024，

6. * 线程名称[MXR-Task-%d],方便回溯，

7. * 拒绝策略：当任务队列已满，抛出RejectedExecutionException

8. * 异常。

9. */

10. private static ThreadPoolExecutor threadPool = new ThreadPoolExecutor(5, 20, 0L,

11. TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(1024)

12. , new ThreadFactoryBuilder().setNameFormat("My-Task-%d").build()

13. , new AbortPolicy()

14. );

我们看到，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 方法即可。

最后，线程工厂，线程池的所有线程都由线程工厂来创建，而默认的线程工厂太过单一，我们看看默认的线程工厂是如何创建线程的：

1. /**

2. * The default thread factory

3. */

4. static class DefaultThreadFactory implements ThreadFactory {

5. private static final AtomicInteger poolNumber = new AtomicInteger(1);

6. private final ThreadGroup group;

7. private final AtomicInteger threadNumber = new AtomicInteger(1);

8. private final String namePrefix;

9. 
   

10. DefaultThreadFactory() {

11. SecurityManager s = System.getSecurityManager();

12. group = (s != null) ? s.getThreadGroup() :

13. Thread.currentThread().getThreadGroup();

14. namePrefix = "pool-" +

15. poolNumber.getAndIncrement() +

16. "-thread-";

17. }

18. 
    

19. public Thread newThread(Runnable r) {

20. Thread t = new Thread(group, r,

21. namePrefix + threadNumber.getAndIncrement(),

22. 0);

23. if (t.isDaemon())

24. t.setDaemon(false);

25. if (t.getPriority() != Thread.NORM_PRIORITY)

26. t.setPriority(Thread.NORM_PRIORITY);

27. return t;

28. }

29. }

可以看到，线程名称为 pool- + 线程池编号 + -thread- + 线程编号 。设置为非守护线程。优先级为默认。

如果我们想修改名称呢？对，实现 ThreadFactory 接口，重写 newThread 方法即可。但是已经有人造好轮子了， 比如我们的例子中使用的 google 的 guaua 提供的 ThreadFactoryBuilder 工厂。可以自定义线程名称，是否守护，优先级，异常处理等等，功能强大。

3. 如何扩展线程池

那么我们能扩展线程池的功能吗？比如记录线程任务的执行时间。实际上，JDK 的线程池已经为我们预留的接口，在线程池核心方法中，有2 个方法是空的，就是给我们预留的。还有一个线程池退出时会调用的方法。我们看看例子：

1. /**

2. * 如何扩展线程池，重写 beforeExecute, afterExecute, terminated 方法，这三个方法默认是空的。

3. *

4. * 可以监控每个线程任务执行的开始和结束时间，或者自定义一些增强。

5. *

6. * 在 Worker 的 runWork 方法中，会调用这些方法

7. */

8. public class ExtendThreadPoolDemo {

9. 
   

10. static class MyTask implements Runnable {

11. 
    

12. String name;

13. 
    

14. public MyTask(String name) {

15. this.name = name;

16. }

17. 
    

18. @Override

19. public void run() {

20. System.out

21. .println("正在执行：Thread ID:" + Thread.currentThread().getId() + ", Task Name = " + name);

22. try {

23. Thread.sleep(100);

24. } catch (InterruptedException e) {

25. e.printStackTrace();

26. }

27. }

28. }

29. 
    

30. public static void main(String[] args) throws InterruptedException {

31. ExecutorService es = new ThreadPoolExecutor(5, 5, 0L, TimeUnit.MILLISECONDS,

32. new LinkedBlockingQueue<>()) {

33. @Override

34. protected void beforeExecute(Thread t, Runnable r) {

35. System.out.println("准备执行：" + ((MyTask) r).name);

36. }

37. 
    

38. @Override

39. protected void afterExecute(Runnable r, Throwable t) {

40. System.out.println("执行完成： " + ((MyTask) r).name);

41. }

42. 
    

43. @Override

44. protected void terminated() {

45. System.out.println("线程池退出");

46. }

47. };

48. 
    

49. for (int i = 0; i < 5; i++) {

50. MyTask myTask = new MyTask("TASK-GEYM-" + i);

51. es.execute(myTask);

52. Thread.sleep(10);

53. 
    

54. }

55. 
    

56. es.shutdown();

57. }

58. 
    

59. }

我们重写了 beforeExecute 方法，也就是执行任务之前会调用该方法，而 afterExecute 方法则是在任务执行完毕后会调用该方法。还有一个 terminated 方法，在线程池退出时会调用该方法。执行结果是什么呢？

可以看到，每个任务执行前后都会调用 before 和 after 方法。相当于执行了一个切面。而在调用 shutdown 方法后则会调用 terminated 方法。

4. 如何优化线程池的异常信息

如何优化线程池的异常信息？ 在说这个问题之前，我们先说一个不容易发现的bug：

看代码：

1. public static void main(String[] args) throws ExecutionException, InterruptedException {

2. 
   

3. ThreadPoolExecutor executor = new ThreadPoolExecutor(0, Integer.MAX_VALUE, 0L,

4. TimeUnit.MILLISECONDS, new SynchronousQueue<>());

5. 
   

6. for (int i = 0; i < 5; i++) {

7. executor.submit(new DivTask(100, i));

8. }

9. 
   

10. }

11. 
    

12. static class DivTask implements Runnable {

13. int a, b;

14. 
    

15. public DivTask(int a, int b) {

16. this.a = a;

17. this.b = b;

18. }

19. 
    

20. @Override

21. public void run() {

22. double re = a / b;

23. System.out.println(re);

24. }

25. }

执行结果：

注意：只有4个结果，其中一个结果被吞没了，并且没有任何信息。为什么呢？如果仔细看代码，会发现，在进行 100 / 0 的时候肯定会报错的，但是却没有报错信息，令人头痛，为什么呢？实际上，如果你使用 execute 方法则会打印错误信息，当你使用 submit 方法却没有调用它的get 方法，异常将会被吞没，因为，如果发生了异常，异常是作为返回值返回的。

怎么办呢？我们当然可以使用 execute 方法，但是我们可以有另一种方式：重写 submit 方法，楼主写了一个例子，大家看一下：

1. static class TraceThreadPoolExecutor extends ThreadPoolExecutor {

2. 
   

3. public TraceThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime,

4. TimeUnit unit, BlockingQueue<Runnable> workQueue) {

5. super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);

6. }

7. 
   

8. @Override

9. public void execute(Runnable command) {

10. // super.execute(command);

11. super.execute(wrap(command, clientTrace(), Thread.currentThread().getName()));

12. }

13. 
    

14. @Override

15. public Future<?> submit(Runnable task) {

16. // return super.submit(task);

17. return super.submit(wrap(task, clientTrace(), Thread.currentThread().getName()));

18. }

19. 
    

20. private Exception clientTrace() {

21. return new Exception("Client stack trace");

22. }

23. 
    

24. private Runnable wrap(final Runnable task, final Exception clientStack,

25. String clientThreaName) {

26. return new Runnable() {

27. @Override

28. public void run() {

29. try {

30. task.run();

31. } catch (Exception e) {

32. e.printStackTrace();

33. clientStack.printStackTrace();

34. throw e;

35. }

36. }

37. };

38. }

39. }

我们重写了 submit 方法，封装了异常信息，如果发生了异常，将会打印堆栈信息。我们看看使用重写后的线程池后的结果是什么？

从结果中，我们清楚的看到了错误信息的原因：by zero！并且堆栈信息明确，方便排错。优化了默认线程池的策略。

5. 如何设计线程池中的线程数量

线程池的大小对系统的性能有一定的影响，过大或者过小的线程数量都无法发挥最优的系统性能，但是线程池大小的确定也不需要做的非常精确。因为只要避免极大和极小两种情况，线程池的大小对性能的影响都不会影响太大，一般来说，确定线程池的大小需要考虑CPU数量，内存大小等因素，在《Java Concurrency in Practice》 书中给出了一个估算线程池大小的经验公式：

公式还是有点复杂的，简单来说，就是如果你是CPU密集型运算，那么线程数量和CPU核心数相同就好，避免了大量无用的切换线程上下文，如果你是IO密集型的话，需要大量等待，那么线程数可以设置的多一些，比如CPU核心乘以2.

至于如何获取 CPU 核心数，Java 提供了一个方法：

Runtime.getRuntime().availableProcessors()；

返回了CPU的核心数量。

总结

好了，到这里，我们已经对如何使用线程池有了一个认识，这里，楼主建议大家手动创建线程池，这样对线程池中的各个参数可以有精准的了解，在对系统进行排错或者调优的时候有好处。比如设置核心线程数多少合适，最大线程数，拒绝策略，线程工厂，队列的大小和类型等等，也可以是G家的线程工厂自定义线程。

原文发布时间为：2018-10-29
本文作者：Java技术驿站
本文来自云栖社区合作伙伴“ Java技术驿站”，了解相关信息可以关注“Java技术驿站”。