UDN-企业互联网技术人气社区

板块导航

浏览  : 1004
回复  : 0

[讨论交流] Java 线程池的创建过程分析

[复制链接]
呵呵燕的头像 楼主
发表于 2016-11-20 22:32:27 | 显示全部楼层 |阅读模式
  最近在改进项目的并发功能,但开发起来磕磕碰碰的。看了好多资料,总算加深了认识。于是打算配合查看源代码,总结并发编程的原理。

  准备从用得最多的线程池开始,围绕创建、执行、关闭认识线程池整个生命周期的实现原理。后续再研究原子变量、并发容器、阻塞队列、同步工具、锁等等主题。java.util.concurrent里的并发工具用起来不难,但不能仅仅会用,我们要read the fucking source code,哈哈。顺便说声,我用的JDK是1.8。

  Executor框架

  Executor是一套线程池管理框架,接口里只有一个方法execute,执行Runnable任务。ExecutorService接口扩展了Executor,添加了线程生命周期的管理,提供任务终止、返回任务结果等方法。AbstractExecutorService实现了ExecutorService,提供例如submit方法的默认实现逻辑。

  然后到今天的主题ThreadPoolExecutor,继承了AbstractExecutorService,提供线程池的具体实现。

  构造方法

  下面是ThreadPoolExecutor最普通的构造函数,最多有七个参数。具体代码不贴了,只是一些参数校验和设置的语句。

  1.   public ThreadPoolExecutor(int corePoolSize,

  2.   int maximumPoolSize,

  3.   long keepAliveTime,

  4.   TimeUnit unit,

  5.   BlockingQueue workQueue,

  6.   ThreadFactory threadFactory,

  7.   RejectedExecutionHandler handler) {

  8.   }
复制代码


  corePoolSize是线程池的目标大小,即是线程池刚刚创建起来,还没有任务要执行时的大小。maximumPoolSize是线程池的最大上限。keepAliveTime是线程的存活时间,当线程池内的线程数量大于corePoolSize,超出存活时间的空闲线程就会被回收。unit就不用说了,剩下的三个参数看后文的分析。

  预设的定制线程池

  ThreadPoolExecutor预设了一些已经定制好的线程池,由Executors里的工厂方法创建。下面分析newSingleThreadExecutor、newFixedThreadPool、newCachedThreadPool的创建参数。

  1.   newFixedThreadPool

  2.   public static ExecutorService newFixedThreadPool(int nThreads) {

  3.   return new ThreadPoolExecutor(nThreads, nThreads,

  4.   0L, TimeUnit.MILLISECONDS,

  5.   new LinkedBlockingQueue());

  6.   }
复制代码


  newFixedThreadPool的corePoolSize和maximumPoolSize都设置为传入的固定数量,keepAliveTim设置为0。线程池创建后,线程数量将会固定不变,适合需要线程很稳定的场合。

  1.   newSingleThreadExecutor

  2.   public static ExecutorService newSingleThreadExecutor() {

  3.   return new FinalizableDelegatedExecutorService

  4.   (new ThreadPoolExecutor(1, 1,

  5.   0L, TimeUnit.MILLISECONDS,

  6.   new LinkedBlockingQueue()));

  7.   }
复制代码


  newSingleThreadExecutor是线程数量固定为1的newFixedThreadPool版本,保证池内的任务串行。注意到返回的是FinalizableDelegatedExecutorService,来看看源码:

  1.   static class FinalizableDelegatedExecutorService

  2.   extends DelegatedExecutorService {

  3.   FinalizableDelegatedExecutorService(ExecutorService executor) {

  4.   super(executor);

  5.   }

  6.   protected void finalize() {

  7.   super.shutdown();

  8.   }

  9.   }
复制代码


  FinalizableDelegatedExecutorService继承了DelegatedExecutorService,仅仅在gc时增加关闭线程池的操作,再来看看DelegatedExecutorService的源码:

  1.   static class DelegatedExecutorService extends AbstractExecutorService {

  2.   private final ExecutorService e;

  3.   DelegatedExecutorService(ExecutorService executor) { e = executor; }

  4.   public void execute(Runnable command) { e.execute(command); }

  5.   public void shutdown() { e.shutdown(); }

  6.   public List shutdownNow() { return e.shutdownNow(); }

  7.   public boolean isShutdown() { return e.isShutdown(); }

  8.   public boolean isTerminated() { return e.isTerminated(); }

  9.   //...

  10.   }
复制代码


  代码很简单,DelegatedExecutorService包装了ExecutorService,使其只暴露出ExecutorService的方法,因此不能再配置线程池的参数。本来,线程池创建的参数是可以调整的,ThreadPoolExecutor提供了set方法。使用newSingleThreadExecutor目的是生成单线程串行的线程池,如果还能配置线程池大小,那就没意思了。

  Executors还提供了unconfigurableExecutorService方法,将普通线程池包装成不可配置的线程池。如果不想线程池被不明所以的后人修改,可以调用这个方法。

  1.   newCachedThreadPool

  2.   public static ExecutorService newCachedThreadPool() {

  3.   return new ThreadPoolExecutor(0, Integer.MAX_VALUE,

  4.   60L, TimeUnit.SECONDS,

  5.   new SynchronousQueue());

  6.   }
复制代码


  newCachedThreadPool生成一个会缓存的线程池,线程数量可以从0到Integer.MAX_VALUE,超时时间为1分钟。线程池用起来的效果是:如果有空闲线程,会复用线程;如果没有空闲线程,会新建线程;如果线程空闲超过1分钟,将会被回收。

  newScheduledThreadPool

  newScheduledThreadPool将会创建一个可定时执行任务的线程池。这个不打算在本文展开,后续会另开文章细讲。

  等待队列

  newCachedThreadPool的线程上限几乎等同于无限,但系统资源是有限的,任务的处理速度总有可能比不上任务的提交速度。因此,可以为ThreadPoolExecutor提供一个阻塞队列来保存因线程不足而等待的Runnable任务,这就是BlockingQueue。

  JDK为BlockingQueue提供了几种实现方式,常用的有:

  ArrayBlockingQueue:数组结构的阻塞队列

  LinkedBlockingQueue:链表结构的阻塞队列

  PriorityBlockingQueue:有优先级的阻塞队列

  SynchronousQueue:不会存储元素的阻塞队列

  newFixedThreadPool和newSingleThreadExecutor在默认情况下使用一个无界的LinkedBlockingQueue。要注意的是,如果任务一直提交,但线程池又不能及时处理,等待队列将会无限制地加长,系统资源总会有消耗殆尽的一刻。所以,推荐使用有界的等待队列,避免资源耗尽。但解决一个问题,又会带来新问题:队列填满之后,再来新任务,这个时候怎么办?后文会介绍如何处理队列饱和。

  newCachedThreadPool使用的SynchronousQueue十分有趣,看名称是个队列,但它却不能存储元素。要将一个任务放进队列,必须有另一个线程去接收这个任务,一个进就有一个出,队列不会存储任何东西。因此,SynchronousQueue是一种移交机制,不能算是队列。newCachedThreadPool生成的是一个没有上限的线程池,理论上提交多少任务都可以,使用SynchronousQueue作为等待队列正合适。

  饱和策略

  当有界的等待队列满了之后,就需要用到饱和策略去处理,ThreadPoolExecutor的饱和策略通过传入RejectedExecutionHandler来实现。如果没有为构造函数传入,将会使用默认的defaultHandler。

  1.   private static final RejectedExecutionHandler defaultHandler = new AbortPolicy();

  2.   public static class AbortPolicy implements RejectedExecutionHandler {

  3.   public AbortPolicy() { }

  4.   public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {

  5.   throw new RejectedExecutionException("Task " + r.toString() + " rejected from " + e.toString());

  6.   }

  7.   }
复制代码


  AbortPolicy是默认的实现,直接抛出一个RejectedExecutionException异常,让调用者自己处理。除此之外,还有几种饱和策略,来看一下:

  1.   public static class DiscardPolicy implements RejectedExecutionHandler {

  2.   public DiscardPolicy() { }

  3.   public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {

  4.   }

  5.   }
复制代码


  DiscardPolicy的rejectedExecution直接是空方法,什么也不干。如果队列满了,后续的任务都抛弃掉。

  1.   public static class DiscardOldestPolicy implements RejectedExecutionHandler {

  2.   public DiscardOldestPolicy() { }

  3.   public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {

  4.   if (!e.isShutdown()) {

  5.   e.getQueue().poll();

  6.   e.execute(r);

  7.   }

  8.   }

  9.   }
复制代码


  DiscardOldestPolicy会将等待队列里最旧的任务踢走,让新任务得以执行。

  1.   public static class CallerRunsPolicy implements RejectedExecutionHandler {

  2.   public CallerRunsPolicy() { }

  3.   public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {

  4.   if (!e.isShutdown()) {

  5.   r.run();

  6.   }

  7.   }

  8.   }
复制代码


  最后一种饱和策略是CallerRunsPolicy,它既不抛弃新任务,也不抛弃旧任务,而是直接在当前线程运行这个任务。当前线程一般就是主线程啊,让主线程运行任务,说不定就阻塞了。如果不是想清楚了整套方案,还是少用这种策略为妙。

  ThreadFactory

  每当线程池需要创建一个新线程,都是通过线程工厂获取。如果不为ThreadPoolExecutor设定一个线程工厂,就会使用默认的defaultThreadFactory:

  1.   public static ThreadFactory defaultThreadFactory() {

  2.   return new DefaultThreadFactory();

  3.   }
复制代码


 
  1.  static class DefaultThreadFactory implements ThreadFactory {

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

  3.   private final ThreadGroup group;

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

  5.   private final String namePrefix;

  6.   DefaultThreadFactory() {

  7.   SecurityManager s = System.getSecurityManager();

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

  9.   Thread.currentThread().getThreadGroup();

  10.   namePrefix = "pool-" +

  11.   poolNumber.getAndIncrement() +

  12.   "-thread-";

  13.   }

  14.   public Thread newThread(Runnable r) {

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

  16.   namePrefix + threadNumber.getAndIncrement(),

  17.   0);

  18.   if (t.isDaemon())

  19.   t.setDaemon(false);

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

  21.   t.setPriority(Thread.NORM_PRIORITY);

  22.   return t;

  23.   }

  24.   }
复制代码


  平时打印线程池里线程的name时,会输出形如pool-1-thread-1之类的名称,就是在这里设置的。这个默认的线程工厂,创建的线程是普通的非守护线程,如果需要定制,实现ThreadFactory后传给ThreadPoolExecutor即可。

  不看代码不总结不会知道,光是线程池的创建就可以引出很多学问。别看平时创建线程池是一句代码的事,其实ThreadPoolExecutor提供了很灵活的定制方法。

原文作者:佚名  来源:开发者头条

相关帖子

您需要登录后才可以回帖 登录 | 立即注册

本版积分规则

关于我们
联系我们
  • 电话:010-86393388
  • 邮件:udn@yonyou.com
  • 地址:北京市海淀区北清路68号
移动客户端下载
关注我们
  • 微信公众号:yonyouudn
  • 扫描右侧二维码关注我们
  • 专注企业互联网的技术社区
版权所有:用友网络科技股份有限公司82041 京ICP备05007539号-11 京公网网备安1101080209224 Powered by Discuz!
快速回复 返回列表 返回顶部