Java线程池架构(一)原理和源码解析

本文涉及的产品
公共DNS(含HTTPDNS解析),每月1000万次HTTP解析
全局流量管理 GTM,标准版 1个月
云解析 DNS,旗舰版 1个月
简介: 文章中其实说明了外部的使用方式,但是没有说内部是如何实现的,为了加深对实现的理解,在使用中可以放心,我们这里将做源码解析以及反馈到原理上,Executors工具可以创建普通的线程池以及schedule调度任务的调度池,其实两者实现上还是有一些区别,但是理解了ThreadPoolExecutor,在看ScheduledThreadPoolExecutor就非常轻松了,后面的文章中也会专门介绍这块,但是需要先看这篇文章。

在前面介绍JUC的文章中,提到了关于线程池Execotors的创建介绍,在文章:《java之JUC系列-外部Tools》中第一部分有详细的说明,请参阅;


文章中其实说明了外部的使用方式,但是没有说内部是如何实现的,为了加深对实现的理解,在使用中可以放心,我们这里将做源码解析以及反馈到原理上,Executors工具可以创建普通的线程池以及schedule调度任务的调度池,其实两者实现上还是有一些区别,但是理解了ThreadPoolExecutor,在看ScheduledThreadPoolExecutor就非常轻松了,后面的文章中也会专门介绍这块,但是需要先看这篇文章。


使用Executors最常用的莫过于是使用:Executors.newFixedThreadPool(int)这个方法,因为它既可以限制数量,而且线程用完后不会一直被cache住;那么就通过它来看看源码,回过头来再看其他构造方法的区别:


在《java之JUC系列-外部Tools》文章中提到了构造方法,为了和本文对接,再贴下代码。

publicstatic ExecutorService <strong>newFixedThreadPool</strong>(int nThreads) {

       returnnewThreadPoolExecutor(nThreads, nThreads,

                                     0L, TimeUnit.MILLISECONDS,

                                     newLinkedBlockingQueue());

}

其实你可以自己new一个ThreadPoolExecutor,来达到自己的参数可控的程度,例如,可以将LinkedBlockingQueue换成其它的(如:SynchronousQueue),只是可读性会降低,这里只是使用了一种设计模式。


我们现在来看看ThreadPoolExecutor的源码是怎么样的,也许你刚开始看他的源码会很痛苦,因为你不知道作者为什么是这样设计的,所以本文就我看到的思想会给你做一个介绍,此时也许你通过知道了一些作者的思想,你也许就知道应该该如何去操作了。


这里来看下构造方法中对那些属性做了赋值:

源码段1:

publicThreadPoolExecutor(int corePoolSize,

                             int maximumPoolSize,

                             long keepAliveTime,

                             TimeUnit unit,

                             BlockingQueue workQueue,

                             ThreadFactory threadFactory,

                             RejectedExecutionHandler handler) {

       if (corePoolSize < 0 ||

           maximumPoolSize <= 0 ||

           maximumPoolSize < corePoolSize ||

           keepAliveTime < 0)

           thrownew IllegalArgumentException();

      if (workQueue == null || threadFactory == null || handler == null)

            thrownew NullPointerException();

        this.corePoolSize = corePoolSize;

        this.maximumPoolSize = maximumPoolSize;

        this.workQueue = workQueue;

        this.keepAliveTime = unit.toNanos(keepAliveTime);

        this.threadFactory = threadFactory;

        this.handler = handler;

    }

这里你可以看到最终赋值的过程,可以先大概知道下参数的意思:

corePoolSize:核心运行的poolSize,也就是当超过这个范围的时候,就需要将新的Runnable放入到等待队列workQueue中了,我们把这些Runnable就叫做要去执行的任务吧。


maximumPoolSize:一般你用不到,当大于了这个值就会将任务由一个丢弃处理机制来处理,但是当你发生:newFixedThreadPool的时候,corePoolSizemaximumPoolSize是一样的,而corePoolSize是先执行的,所以他会先被放入等待队列,而不会执行到下面的丢弃处理中,看了后面的代码你就知道了。


workQueue:等待队列,当达到corePoolSize的时候,就向该等待队列放入线程信息(默认为一个LinkedBlockingQueue),运行中的线程属性为:workers,为一个HashSet;我们的Runnable内部被包装了一层,后面会看到这部分代码;这个队列默认是一个无界队列(你也可以设定一个有界队列),所以在生产者疯狂生产的时候,考虑如何控制的问题。


keepAliveTime:默认都是0,当线程没有任务处理后,保持多长时间,当你使用:newCachedThreadPool(),它将是60s的时间。这个参数在运行中的线程从workQueue获取任务时,当(poolSize >corePoolSize || allowCoreThreadTimeOut)会用到,当然allowCoreThreadTimeOut要设置为true,也会先判定keepAliveTime是大于0的,不过由于它在corePoolSize上采用了Integer.MAX_VALUE,当遇到系统遇到瞬间冲击,workers就会迅速膨胀,所以这个地方就不要去设置allowCoreThreadTimeOut=true,否则结果是这些运行中的线程会持续60s以上;另外,如果corePoolSize的值还没到Integer.MAX_VALUE,当超过那个值以后,这些运行中的线程,也是

threadFactory:是构造Thread的方法,你可以自己去包装和传递,主要实现newThread方法即可;


handler:也就是参数maximumPoolSize达到后丢弃处理的方法,java提供了5种丢弃处理的方法,当然你也可以自己根据实际情况去重写,主要是要实现接口:RejectedExecutionHandler中的方法: public void rejectedExecution(Runnabler, ThreadPoolExecutor e) java默认的是使用:AbortPolicy,他的作用是当出现这中情况的时候会抛出一个异常;

其余的还包含:


1、CallerRunsPolicy:如果发现线程池还在运行,就直接运行这个线程

2、DiscardOldestPolicy:在线程池的等待队列中,将头取出一个抛弃,然后将当前线程放进去。

3、DiscardPolicy:什么也不做

4、AbortPolicy:java默认,抛出一个异常:RejectedExecutionException

你可以自己写一个,例如我们想在这个处理中,既不是完全丢弃,也不是完全启动,也不是抛异常,而是控制生产者的线程,那么你就可以尝试某种方式将生产者的线程blocking住,其实就有点类似提到的Semaphor的功能了。

通常你得到线程池后,会调用其中的:submit方法或execute方法去操作;其实你会发现,submit方法最终会调用execute方法来进行操作,只是他提供了一个Future来托管返回值的处理而已,当你调用需要有返回值的信息时,你用它来处理是比较好的;这个Future会包装对Callable信息,并定义一个Sync对象(),当你发生读取返回值的操作的时候,会通过Sync对象进入锁,直到有返回值的数据通知,具体细节先不要看太多。


继续向下,来看看execute最为核心的方法吧: 源码段2:

    public void execute(Runnable command) {

        if (command == null)

            throw newNullPointerException();

        if (poolSize >= corePoolSize || !addIfUnderCorePoolSize(command)) {

           if (runState == RUNNING && workQueue.offer(command)) {

               if (runState != RUNNING || poolSize == 0)

                   ensureQueuedTaskHandled(command);

           }

           elseif (!addIfUnderMaximumPoolSize(command))

               reject(command); // is shutdown or saturated

       }

   }

这段代码看似简单,其实有点难懂,很多人也是这里没看懂,没事,我一个if一个if说:

首先第一个判定空操作就不用说了,下面判定的poolSize >= corePoolSize成立时候会进入if的区域,当然它不成立也有可能会进入,他会判定addIfUnderCorePoolSize是否返回false,如果返回false就会进去;

我们先来看下addIfUnderCorePoolSize方法的源码是什么:

源码段3:

privatebooleanaddIfUnderCorePoolSize(Runnable firstTask) {

       Threadt=null;

       finalReentrantLockmainLock=this.mainLock;

       mainLock.lock();

       try {

           if (poolSize < corePoolSize && runState == RUNNING)

               t = addThread(firstTask);

       } finally {

           mainLock.unlock();

       }

       if (t == null)

           returnfalse;

       t.start();

       returntrue;

   }

可以发现,这段源码是如果发现小雨corePoolSize就会创建一个新的线程,并且调用线程的start()方法将线程运行起来:这个addThread()方法,我们先不考虑细节,因为我们还要先看到前面是怎么进去的,这里可以发信啊,只有没有创建成功Thread才会返回false,也就是当当前的poolSize > corePoolSize的时候,或线程池已经不是在running状态的时候才会出现;

注意:这里在外部判定一次poolSize和corePoolSize只是初步判定,内部是加锁后判定的,以得到更为准确的结果,而外部初步判定如果是大于了,就没有必要进入这段有锁的代码了。

此时我们知道了,当前线程数量大于corePoolSize的时候,就会进入【代码段2】的第一个if语句中,回到【源码段2】,继续看if语句中的内容:

这里标记为

源码段4:

if (runState == RUNNING && workQueue.offer(command)) {

               if (runState != RUNNING || poolSize == 0)

                   ensureQueuedTaskHandled(command);

           }

           elseif (!addIfUnderMaximumPoolSize(command))

               reject(command); // is shutdown or saturated

第一个if,也就是当当前状态为running的时候,就会去执行workQueue.offer(command),这个workQueue其实就是一个BlockingQueue,offer()操作就是在队列的尾部写入一个对象,此时写入的对象为线程的对象而已;所以你可以认为只有线程池在RUNNING状态,才会在队列尾部插入数据,否则就执行else if,其实else if可以看出是要做一个是否大于MaximumPoolSize的判定,如果大于这个值,就会做reject的操作,关于reject的说明,我们在【源码段1】的解释中已经非常明确的说明,这里可以简单看下源码,以应征结果:

源码段5:

   private boolean addIfUnderMaximumPoolSize(Runnable firstTask) {

       Thread t = null;

       final ReentrantLock mainLock = this.mainLock;

       mainLock.lock();

       try {

           if (poolSize < maximumPoolSize && runState == RUNNING)                 //在corePoolSize = maximumPoolSize下,该代码几乎不可能运行                 t = addThread(firstTask);         } finally {             mainLock.unlock();         }         if (t == null)             returnfalse;         t.start();         returntrue; } void reject(Runnable command) {         handler.rejectedExecution(command, this);     }

也就是如果线程池满了,而且线程池调用了shutdown后,还在调用execute方法时,就会抛出上面说明的异常:RejectedExecutionException 再回头来看下【代码段4】中进入到等待队列后的操作:

if (runState != RUNNING || poolSize == 0)     ensureQueuedTaskHandled(command);

这段代码是要在线程池运行状态不是RUNNING或poolSize == 0才会调用,他是干啥呢? 他为什么会不等于RUNNING呢?外面那一层不是判定了他== RUNNING了么,其实有时间差就是了,如果是poolSize == 0也会执行这段代码,但是里面的判定条件是如果不是RUNNING,就做reject操作,在第一个线程进去的时候,会将第一个线程直接启动起来;很多人也是看这段代码很绕,因为不断的循环判定类似的判定条件,你主要记住他们之间有时间差,要取最新的就好了。 此时貌似代码看完了?咦,此时有问题了: 1、 等待中的线程在后来是如何跑起来的呢?线程池是不是有类似Timer一样的守护进程不断扫描线程队列和等待队列?还是利用某种锁机制,实现类似wait和notify实现的? 2、 线程池的运行队列和等待队列是如何管理的呢?这里还没看出影子呢! NO,NO,NO! Java在实现这部分的时候,使用了怪异的手段,神马手段呢,还要再看一部分代码才晓得。 在前面【源码段3】中,我们看到了一个方法叫:addThread(),也许很少有人会想到关键在这里,其实关键就是在这里: 我们看看addThread()方法到底做了什么。 源码段6:

private Thread addThread(Runnable firstTask) {         Worker w = newWorker(firstTask);         Thread t = threadFactory.newThread(w);         if (t != null) {             w.thread = t;             workers.add(w);             int nt = ++poolSize;             if (nt > largestPoolSize)

               largestPoolSize = nt;

       }

       return t;

   }

这里创建了一个Worker,其余的操作,就是将poolSize++的操作,然后将将其放入workers的运行的HashSet中等操作;

我们主要关心Worker是干什么的,因为这个threadFactory对我们用途不大,只是做了Thread的命名处理;而Worker你会发现它的定义也是一个Runnable,外部开始在代码段中发现了调用哪个这个Worker的start()方法,也就是线程的启动方法,其实也就是调用了Worker的run()方法,那么我们重点要关心run方法是如何处理的

源码段7:

publicvoid run() {

           try {

               Runnable task = firstTask;

               firstTask = null;

               while (task != null || (task = getTask()) != null) {

                   runTask(task);

                   task = null;

               }

           } finally {

               workerDone(this);

           }

       }

FirstTask其实就是开始在创建work的时候,由外部传入的Runnable对象,也就是你自己的Thread,你会发现它如果发现task为空,就会调用getTask()方法再判定,直到两者为空,并且是一个while循环体。

那么看看getTask()方法的实现为:

源码段8:

    Runnable getTask() {

       for (;;) {

           try {

               int state = runState;

               if (state > SHUTDOWN)

                   returnnull;

               Runnable r;

               if (state == SHUTDOWN)  // Help drain queue

                   r = workQueue.poll();

               elseif (poolSize > corePoolSize || allowCoreThreadTimeOut)

                   r = workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS);

               else

                   r = workQueue.take();

               if (r != null)

                   return r;

               if (workerCanExit()) {

                   if (runState >= SHUTDOWN) // Wake up others

                       interruptIdleWorkers();

                   returnnull;

               }

               // Else retry

           } catch (InterruptedException ie) {

               // On interruption, re-check runState

           }

       }

   }

你会发现它是从workQueue队列中,也就是等待队列中获取一个元素出来并返回!

回过头来根据代码段6理解下:

当前线程运行完后,在到workQueue中去获取一个task出来,继续运行,这样就保证了线程池中有一定的线程一直在运行;此时若跳出了while循环,只有workQueue队列为空才会出现或出现了类似于shutdown的操作,自然运行队列会减少1,当再有新的线程进来的时候,就又开始向worker里面放数据了,这样以此类推,实现了线程池的功能。

这里可以看下run方法的finally中调用的workerDone方法为:

源码段9:

voidworkerDone(Worker w) {

       final ReentrantLock mainLock = this.mainLock;

       mainLock.lock();

       try {

           completedTaskCount += w.completedTasks;

           workers.remove(w);

           if (--poolSize == 0)

               tryTerminate();

       } finally {

           mainLock.unlock();

       }

   }

注意这里将workers.remove(w)掉,并且调用了—poolSize来做操作。

至于tryTerminate是做了更多关于回收方面的操作。

最后我们还要看一段代码就是在【源码段6】中出现的代码调用为:runTask(task);这个方法也是运行的关键。

源码段10:

privatevoid runTask(Runnable task) {

           final ReentrantLock runLock = this.runLock;

           runLock.lock();

           try {

               if (runState < STOP &&                     Thread.interrupted() &&                     runState >= STOP)

                   thread.interrupt();

               boolean ran = false;

               beforeExecute(thread, task);

               try {

                   task.run();

                   ran = true;

                   afterExecute(task, null);

                   ++completedTasks;

               } catch (RuntimeException ex) {

                   if (!ran)

                       afterExecute(task, ex);

                   throw ex;

               }

           } finally {

               runLock.unlock();

           }

       }

你可以看到,这里面的task为传入的task信息,调用的不是start方法,而是run方法,因为run方法直接调用不会启动新的线程,也是因为这样,导致了你无法获取到你自己的线程的状态,因为线程池是直接调用的run方法,而不是start方法来运行。

这里有个beforeExecuteafterExecute方法,分别代表在执行前和执行后,你可以做一段操作,在这个类中,这两个方法都是【空body】的,因为普通线程池无需做更多的操作。

如果你要实现类似暂停等待通知的或其他的操作,可以自己extends后进行重写构造;

本文没有介绍关于ScheduledThreadPoolExecutor调用的细节,下一篇文章会详细说明,因为大部分代码和本文一致,区别在于一些细节,在介绍:ScheduledThreadPoolExecutor的时候,会明确的介绍它与Timer和TimerTask的巨大区别,区别不在于使用,而是在于本身内在的处理细节。


相关文章
|
8天前
|
监控 网络协议 Java
Tomcat源码解析】整体架构组成及核心组件
Tomcat,原名Catalina,是一款优雅轻盈的Web服务器,自4.x版本起扩展了JSP、EL等功能,超越了单纯的Servlet容器范畴。Servlet是Sun公司为Java编程Web应用制定的规范,Tomcat作为Servlet容器,负责构建Request与Response对象,并执行业务逻辑。
Tomcat源码解析】整体架构组成及核心组件
|
5天前
|
存储 JSON 数据库
Elasticsearch 分布式架构解析
【9月更文第2天】Elasticsearch 是一个分布式的搜索和分析引擎,以其高可扩展性和实时性著称。它基于 Lucene 开发,但提供了更高级别的抽象,使得开发者能够轻松地构建复杂的搜索应用。本文将深入探讨 Elasticsearch 的分布式存储和检索机制,解释其背后的原理及其优势。
28 5
|
7天前
|
存储 容灾 关系型数据库
OceanBase 高可用性架构解析
【8月更文第31天】在大数据和云计算蓬勃发展的今天,数据库作为数据存储的核心组件,其稳定性和可靠性直接影响到整个系统的性能。OceanBase 是由阿里巴巴集团自主研发的一款分布式关系型数据库系统,旨在为大规模在线交易处理(OLTP)场景提供高性能、高可用性的解决方案。本文将深入探讨 OceanBase 是如何通过其独特的架构设计来确保数据的高可用性和容灾能力。
46 0
|
7天前
|
消息中间件 测试技术 API
深入解析微服务架构的设计与实践
在软件工程领域,"分而治之"的策略一直是解决复杂问题的有效方法。微服务架构作为这一策略的现代体现,它通过将大型应用程序分解为一组小的、独立的服务来简化开发与部署。本文将带你了解微服务的核心概念,探讨设计时的关键考虑因素,并分享实践中的一些经验教训,旨在帮助开发者更好地构建和维护可扩展的系统。
|
8天前
|
Java API
Java 8新特性:Lambda表达式与Stream API的深度解析
【7月更文挑战第61天】本文将深入探讨Java 8中的两个重要特性:Lambda表达式和Stream API。我们将首先介绍Lambda表达式的基本概念和语法,然后详细解析Stream API的使用和优势。最后,我们将通过实例代码演示如何结合使用Lambda表达式和Stream API,以提高Java编程的效率和可读性。
|
6天前
|
设计模式 存储 人工智能
深度解析Unity游戏开发:从零构建可扩展与可维护的游戏架构,让你的游戏项目在模块化设计、脚本对象运用及状态模式处理中焕发新生,实现高效迭代与团队协作的完美平衡之路
【9月更文挑战第1天】游戏开发中的架构设计是项目成功的关键。良好的架构能提升开发效率并确保项目的长期可维护性和可扩展性。在使用Unity引擎时,合理的架构尤为重要。本文探讨了如何在Unity中实现可扩展且易维护的游戏架构,包括模块化设计、使用脚本对象管理数据、应用设计模式(如状态模式)及采用MVC/MVVM架构模式。通过这些方法,可以显著提高开发效率和游戏质量。例如,模块化设计将游戏拆分为独立模块。
28 3
|
6天前
|
存储 SQL Cloud Native
Hologres 的架构设计与工作原理
【9月更文第1天】随着大数据时代的到来,实时分析和处理数据的需求日益增长。传统的数据仓库在处理大规模实时数据分析时逐渐显露出性能瓶颈。为了解决这些问题,阿里巴巴集团研发了一款名为 Hologres 的新型云原生交互式分析数据库。Hologres 能够支持 SQL 查询,并且能够实现实时的数据写入和查询,这使得它成为处理大规模实时数据的理想选择。
25 2
|
8天前
|
Java 开发者 UED
“Java开发者必看:异步编程实战解析,掌握这些技巧,让你的代码跑得更快!
【8月更文挑战第30天】随着互联网技术的发展,系统性能和用户体验成为关注焦点。异步编程作为提高应用响应速度和吞吐量的技术,在Java中广泛采用。本文详细介绍了Java异步编程的概念与优势,并通过实战示例展示了如何利用Future、Callable及CompletableFuture在实际项目中实施异步编程,帮助开发者更好地理解和应用这一技术。
22 2
|
8天前
|
存储 分布式计算 Hadoop
ChunkServer 原理与架构详解
【8月更文第30天】在分布式文件系统中,ChunkServer 是一个重要的组件,负责存储文件系统中的数据块(chunks)。ChunkServer 的设计和实现对于确保数据的高可用性、一致性和持久性至关重要。本文将深入探讨 ChunkServer 的核心原理和内部架构设计,并通过代码示例来说明其实现细节。
10 1
|
4天前
|
安全 Java API
Java线程池原理与锁机制分析
综上所述,Java线程池和锁机制是并发编程中极其重要的两个部分。线程池主要用于管理线程的生命周期和执行并发任务,而锁机制则用于保障线程安全和防止数据的并发错误。它们深入地结合在一起,成为Java高效并发编程实践中的关键要素。
4 0

推荐镜像

更多
下一篇
DDNS