专栏首页刘望舒Java高并发之线程池详解

Java高并发之线程池详解

作者:大道方圆 https://www.cnblogs.com/xdecode

线程池优势

在业务场景中, 如果一个对象创建销毁开销比较大, 那么此时建议池化对象进行管理。例如线程, jdbc连接等等, 在高并发场景中, 如果可以复用之前销毁的对象, 那么系统效率将大大提升。另外一个好处是可以设定池化对象的上限, 例如预防创建线程数量过多导致系统崩溃的场景.

JDK中的线程池

下文主要从以下几个角度讲解:

创建线程池

我们可以通过自定义ThreadPoolExecutor或者jdk内置的Executors来创建一系列的线程池

  • newFixedThreadPool: 创建固定线程数量的线程池
  • newSingleThreadExecutor: 创建单一线程的池
  • newCachedThreadPool: 创建线程数量自动扩容, 自动销毁的线程池
  • newScheduledThreadPool: 创建支持计划任务的线程池

上述几种都是通过new ThreadPoolExecutor()来实现的, 构造函数源码如下:

 1     /**
 2      * @param corePoolSize 池内核心线程数量, 超出数量的线程会进入阻塞队列
 3      * @param maximumPoolSize 最大可创建线程数量
 4      * @param keepAliveTime 线程存活时间
 5      * @param unit 存活时间的单位
 6      * @param workQueue 线程溢出后的阻塞队列
 7      */
 8     public ThreadPoolExecutor(int corePoolSize,
 9                               int maximumPoolSize,
10                               long keepAliveTime,
11                               TimeUnit unit,
12                               BlockingQueue<Runnable> workQueue) {
13         this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, Executors.defaultThreadFactory(), defaultHandler);
14     }
15 
16     public static ExecutorService newFixedThreadPool(int nThreads) {
17         return new ThreadPoolExecutor(nThreads, nThreads, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>());
18     }
19 
20     public static ExecutorService newSingleThreadExecutor() {
21         return new Executors.FinalizableDelegatedExecutorService
22                 (new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>()));
23     }
24 
25     public static ExecutorService newCachedThreadPool() {
26         return new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS, new SynchronousQueue<Runnable>());
27     }
28 
29     public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
30         return new ScheduledThreadPoolExecutor(corePoolSize);
31     }
32 
33     public ScheduledThreadPoolExecutor(int corePoolSize) {
34         super(corePoolSize, Integer.MAX_VALUE, 0, TimeUnit.NANOSECONDS, new ScheduledThreadPoolExecutor.DelayedWorkQueue());
35     }

提交任务

直接调用executorService.execute(runnable)或者submit(runnable)即可。

execute和submit的区别在于submit会返回Future来获取任何执行的结果.

我们看下newScheduledThreadPool的使用示例。

 1 public class SchedulePoolDemo {
 2 
 3     public static void main(String[] args){
 4         ScheduledExecutorService service = Executors.newScheduledThreadPool(10);
 5         // 如果前面的任务没有完成, 调度也不会启动
 6         service.scheduleAtFixedRate(new Runnable() {
 7             @Override
 8             public void run() {
 9                 try {
10                     Thread.sleep(2000);
11                     // 每两秒打印一次.
12                     System.out.println(System.currentTimeMillis()/1000);
13                 } catch (InterruptedException e) {
14                     e.printStackTrace();
15                 }
16             }
17         }, 0, 2, TimeUnit.SECONDS);
18     }
19 }

潜在宕机风险

使用Executors来创建要注意潜在宕机风险.其返回的线程池对象的弊端如下:

  • FixedThreadPool和SingleThreadPoolPool : 允许的请求队列长度为 Integer.MAX_VALUE,可能会堆积大量的请求,从而导致 OOM.
  • CachedThreadPool和ScheduledThreadPool : 允许的创建线程数量为 Integer.MAX_VALUE,可能会创建大量的线程,从而导致 OOM.

综上所述, 在可能有大量请求的线程池场景中, 更推荐自定义ThreadPoolExecutor来创建线程池, 具体构造函数配置见下文.

线程池大小配置

一般根据任务类型进行区分, 假设CPU为N核

  • CPU密集型任务需要减少线程数量, 降低线程之间切换造成的开销, 可配置线程池大小为N + 1.
  • IO密集型任务则可以加大线程数量, 可配置线程池大小为 N * 2.
  • 混合型任务则可以拆分为CPU密集型与IO密集型, 独立配置.

自定义阻塞队列BlockingQueue

主要存放等待执行的线程, ThreadPoolExecutor中支持自定义该队列来实现不同的排队队列.

  • ArrayBlockingQueue:先进先出队列,创建时指定大小, 有界;
  • LinkedBlockingQueue:使用链表实现的先进先出队列,默认大小为- Integer.MAX_VALUE;
  • SynchronousQueue:不保存提交的任务, 数据也不会缓存到队列中, 用于生产者和消费者互等对方, 一起离开.
  • PriorityBlockingQueue: 支持优先级的队列

回调接口

线程池提供了一些回调方法, 具体使用如下所示。

 1         ExecutorService service = new ThreadPoolExecutor(5, 5, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingDeque<Runnable>()) {
 2 
 3             @Override
 4             protected void beforeExecute(Thread t, Runnable r) {
 5                 System.out.println("准备执行任务: " + r.toString());
 6             }
 7 
 8             @Override
 9             protected void afterExecute(Runnable r, Throwable t) {
10                 System.out.println("结束任务: " + r.toString());
11             }
12 
13             @Override
14             protected void terminated() {
15                 System.out.println("线程池退出");
16             }
17         };

可以在回调接口中, 对线程池的状态进行监控, 例如任务执行的最长时间, 平均时间, 最短时间等等, 还有一些其他的属性如下:

  • taskCount:线程池需要执行的任务数量.
  • completedTaskCount:线程池在运行过程中已完成的任务数量.小于或等于taskCount。
  • largestPoolSize:线程池曾经创建过的最大线程数量.通过这个数据可以知道线程池是否满过.如等于线程池的最大大小,则表示线程池曾经满了。
  • getPoolSize:线程池的线程数量.如果线程池不销毁的话,池里的线程不会自动销毁,所以这个大小只增不减。
  • getActiveCount:获取活动的线程数。

自定义拒绝策略

线程池满负荷运转后, 因为时间空间的问题, 可能需要拒绝掉部分任务的执行。

jdk提供了RejectedExecutionHandler接口, 并内置了几种线程拒绝策略

  • AbortPolicy: 直接拒绝策略, 抛出异常.
  • CallerRunsPolicy: 调用者自己执行任务策略.
  • DiscardOldestPolicy: 舍弃最老的未执行任务策略.

使用方式也很简单, 直接传参给ThreadPool

1         ExecutorService service = new ThreadPoolExecutor(5, 5, 0L, TimeUnit.MILLISECONDS, 
2                 new SynchronousQueue<Runnable>(),
3                 Executors.defaultThreadFactory(),
4                 new RejectedExecutionHandler() {
5                     @Override
6                     public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
7                         System.out.println("reject task: " + r.toString());
8                     }
9                 });

自定义ThreadFactory

线程工厂用于创建池里的线程.。例如在工厂中都给线程setDaemon(true),,这样程序退出的时候,线程自动退出。

或者统一指定线程优先级, 设置名称等等。

 1 class NamedThreadFactory implements ThreadFactory {
 2     private static final AtomicInteger threadIndex = new AtomicInteger(0);
 3     private final String baseName;
 4     private final boolean daemon;
 5 
 6     public NamedThreadFactory(String baseName) {
 7         this(baseName, true);
 8     }
 9 
10     public NamedThreadFactory(String baseName, boolean daemon) {
11         this.baseName = baseName;
12         this.daemon = daemon;
13     }
14 
15     public Thread newThread(Runnable runnable) {
16         Thread thread = new Thread(runnable, this.baseName + "-" + threadIndex.getAndIncrement());
17         thread.setDaemon(this.daemon);
18         return thread;
19     }
20 }

关闭线程池

跟直接new Thread不一样, 局部变量的线程池,,需要手动关闭,,不然会导致线程泄漏问题。

默认提供两种方式关闭线程池.

  • shutdown:等所有任务,包括阻塞队列中的执行完,才会终止,但是不会接受新任务。
  • shutdownNow:立即终止线程池,,打断正在执行的任务, 清空队列。

本文分享自微信公众号 - 刘望舒(liuwangshuAndroid)

原文出处及转载信息见文内详细说明,如有侵权,请联系 yunjia_community@tencent.com 删除。

原始发表时间:2018-07-20

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

我来说两句

0 条评论
登录 后参与评论

相关文章

  • Android中多线程切换的几种方法

    我们知道,多线程是Android开发中必现的场景,很多原生API和开源项目都有多线程的内容,这里简单总结和探讨一下常见的多线程切换方式。 我们先回顾一下Java...

    用户1269200
  • Android解析WindowManagerService(一)WMS的诞生

    前言 此前我用多篇文章介绍了WindowManager,这个系列我们来介绍WindowManager的管理者WMS,首先我们先来学习WMS是如何产生的。本文源码...

    用户1269200
  • Android响应式编程(一)RxJava前篇[入门基础]

    1.RxJava概述 ReactiveX与RxJava 在讲到RxJava之前我们首先要了解什么是ReactiveX,因为RxJava是ReactiveX的一种...

    用户1269200
  • 1.11守护线程

    在Java中有两种线程,一种为用户线程,一种为守护线程。 守护线程是一种特殊的线程,它具有“陪伴”的含义,当进程中不存在非守护线程时,则守护线程自动销毁。 典型...

    用户1134788
  • Java并发编程之线程池必用知识点

    再使用线程池之前,我们应该了解为什么需要使用线程池。进行执行任务(task)的时候我们一般情况是new Thread进行执行,如果进行大量的并发任务的时候呢?

    静默加载
  • 我画了25张图展示线程池工作原理和实现原理,原创干货,建议先收藏再阅读

    有朋友留言提到文中的场景是IO密集型操作,不是CPU密集操作,不需要使用线程池,我猜这位朋友可能想表达的是IO密集且阻塞时间久的不要使用线程池方案解决。IO密集...

    JavaQ
  • Java 线程池ThreadPoolExecutor原理及源码全面解析(基于JDK8)

    1、线程在java中是一个对象,更是操作系统的资源,线程创建、销毁都需要时间。 如果创建时间+销毁时间>执行任务时间就很不合算 2、Java对象占用堆内存,...

    JavaEdge
  • Thread 源码面试

    每个线程都有一个优先级。优先级高的线程优先于优先级低的线程执行。每个线程可能被标记为守护线程,也可能不被标记为守护线程。

    JavaEdge
  • Java线程组ThreadGroup

    一个线程集合。是为了更方便地管理线程。父子结构的,一个线程组可以集成其他线程组,同时也可以拥有其他子线程组。

    JavaEdge
  • Java 线程池讲解——针对 IO 密集型任务

    针对 IO 密集型的任务,我们可以针对原本的线程池做一些改造,从而可以提高任务的处理效率。

    健程之道

扫码关注云+社区

领取腾讯云代金券