首页    技术博客    PHP教程    数据库技术    前端开发   HTML5    Nginx    php论坛
新用户注册 | 会员登录
加入会员，解锁专属内容
取消
热门标签 | HotTags
当前位置:  开发笔记 > 编程语言 > 正文

jucExecutor框架详解

作者：lin | 来源：互联网 | 2023-10-11 11:26
Executor框架是juc里提供的线程池的实现。前两天看了下Executor框架的一些源码，做个简单的总结。线程池大概的思路是维护一个的线程池用于执行提交的任务。我理解池的技术的主要意

Executor 框架是 juc 里提供的线程池的实现。前两天看了下 Executor 框架的一些源码，做个简单的总结。

线程池大概的思路是维护一个的线程池用于执行提交的任务。我理解池的技术的主要意义有两个：

  1. 资源的控制，如并发量限制。像连接池这种是对数据库资源的保护。
  2. 资源的有效利用，如线程复用，避免频繁创建线程和线程上下文切换。

那么想象中设计一个线程池就需要有线程池大小、线程生命周期管理、等待队列等等功能，下面结合代码看看原理。

Excutor 整体结构如下： 

Executor 接口定义了最基本的 execute 方法，用于接收用户提交任务。 ExecutorService 定义了线程池终止和创建及提交 futureTask 任务支持的方法。

AbstractExecutorService 是抽象类，主要实现了 ExecutorService 和 futureTask 相关的一些任务创建和提交的方法。

ThreadPoolExecutor 是最核心的一个类，是线程池的内部实现。线程池的功能都在这里实现了，平时用的最多的基本就是这个了。其源码很精练，远没当时想象的多。

ScheduledThreadPoolExecutor 在 ThreadPoolExecutor 的基础上提供了支持定时调度的功能。线程任务可以在一定延时时间后才被触发执行。

1.ThreadPoolExecutor 原理

 

1.1 ThreadPoolExecutor内部的几个重要属性

 

 

1.线程池本身的状态

Java代码   
  1. volatile int runState;   
  2. static final int RUNNING = 0;   
  3. static final int SHUTDOWN = 1;   
  4. static final int STOP = 2;   
  5. static final int TERMINATED = 3;   

 

2.等待任务队列和工作集

Java代码   
  1. private final BlockingQueue workQueue; //等待被执行的Runnable任务   
  2. private final HashSet workers = new HashSet(); //正在被执行的Worker任务集   

 

3.线程池的主要状态锁。线程池内部的状态变化 ( 如线程大小 ) 都需要基于此锁。

Java代码   
  1. private final ReentrantLock mainLock = new ReentrantLock();  

 

4.线程的存活时间和大小

Java代码   
  1. private volatile long keepAliveTime;// 线程存活时间   
  2. private volatile boolean allowCoreThreadTimeOut;// 是否允许核心线程存活   
  3. private volatile int corePoolSize;// 核心池大小   
  4. private volatile int maximumPoolSize; // 最大池大小   
  5. private volatile int poolSize; //当前池大小   
  6. private int largestPoolSize; //最大池大小，区别于maximumPoolSize，是用于记录线程池曾经达到过的最大并发,理论上小于等于maximumPoolSize。   

 

5.线程工厂和拒绝策略

Java代码   
  1. private volatile RejectedExecutionHandler handler;// 拒绝策略，用于当线程池无法承载新线程是的处理策略。  
  2.  private volatile ThreadFactory threadFactory;// 线程工厂，用于在线程池需要新创建线程的时候创建线程  

 

6.线程池完成任务数

Java代码   
  1. private long completedTaskCount;//线程池运行到当前完成的任务数总和  

 

 

1.2 ThreadPoolExecutor 的内部工作原理

 

有了以上定义好的数据，下面来看看内部是如何实现的 。 Doug Lea 的整个思路总结起来就是 5 句话：

  1. 如果当前池大小 poolSize 小于 corePoolSize ，则创建新线程执行任务。
  2. 如果当前池大小 poolSize 大于 corePoolSize ，且等待队列未满，则进入等待队列
  3. 如果当前池大小 poolSize 大于 corePoolSize 且小于 maximumPoolSize ，且等待队列已满，则创建新线程执行任务。
  4. 如果当前池大小 poolSize 大于 corePoolSize 且大于 maximumPoolSize ，且等待队列已满，则调用拒绝策略来处理该任务。
  5. 线程池里的每个线程执行完任务后不会立刻退出，而是会去检查下等待队列里是否还有线程任务需要执行，如果在 keepAliveTime 里等不到新的任务了，那么线程就会退出。

 

下面看看代码实现 :

线程池最重要的方法是由 Executor 接口定义的 execute 方法 , 是任务提交的入口。

我们看看 ThreadPoolExecutor.execute(Runnable cmd) 的实现：

 

Java代码   
  1. public void execute(Runnable command) {  
  2.         if (command == null)  
  3.             throw new NullPointerException();  
  4.         if (poolSize >= corePoolSize || !addIfUnderCorePoolSize(command)) {  
  5.             if (runState == RUNNING && workQueue.offer(command)) {  
  6.                 if (runState != RUNNING || poolSize == 0)  
  7.                     ensureQueuedTaskHandled(command);  
  8.             }  
  9.             else if (!addIfUnderMaximumPoolSize(command))  
  10.                 reject(command); // is shutdown or saturated  
  11.         }  
  12. }  

 

解释如下：

当提交一个新的 Runnable 任务：

分支1 ：  如果当前池大小小于 corePoolSize, 执行 addIfUnderCorePoolSize(command) , 如果线程池处于运行状态且 poolSize < corePoolSize addIfUnderCorePoolSize(command) 会做如下事情，将Runnable 任务封装成 Worker 任务 , 创建新的 Thread ，执行 Worker 任务。如果不满足条件，则返回 false。代码如下：

 

Java代码   
  1.  private boolean addIfUnderCorePoolSize(Runnable firstTask) {  
  2.         Thread t = null;  
  3.         final ReentrantLock mainLock = this.mainLock;  
  4.         mainLock.lock();  
  5.         try {  
  6.             if (poolSize < corePoolSize && runState == RUNNING)  
  7.                 t = addThread(firstTask);  
  8.         } finally {  
  9.             mainLock.unlock();  
  10.         }  
  11.         if (t == null)  
  12.             return false;  
  13.         t.start();  
  14.         return true;  
  15. }  

 

 

    分支2 ： 如果大于 corePoolSize 或 1 失败失败，则：

  •   如果等待队列未满，把 Runnable 任务加入到 workQueue 等待队列 

    workQueue .offer(command)

  •   如多等待队列已经满了，调用 addIfUnderMaximumPoolSize(command) ，和addIfUnderCorePoolSize 基本类似，只不过判断条件是 poolSize < maximumPoolSize 。如果大于maximumPoolSize ，则把 Runnable 任务交由 RejectedExecutionHandler 来处理。

 

问题：如何实现线程的复用 ?  

Doug Lea 的实现思路是 线程池里的每个线程执行完任务后不立刻退出，而是去检查下等待队列里是否还有线程任务需要执行，如果在 keepAliveTime 里等不到新的任务了，那么线程就会退出。这个功能的实现 关键在于Worker 。线程池在执行 Runnable 任务的时候，并不单纯把 Runnable 任务交给创建一个 Thread 。而是会把Runnable 任务封装成 Worker 任务。

下面看看 Worker 的实现：

 代码很简单，可以看出， worker 里面包装了 firstTask 属性，在构造worker 的时候传进来的那个 Runnable 任务就是 firstTask 。 同时也实现了Runnable接口，所以是个代理模式，看看代理增加了哪些功能。 关键看 woker 的run 方法：

 

Java代码   
  1. public void run() {  
  2.            try {  
  3.                Runnable task = firstTask;  
  4.                firstTask = null;  
  5.                while (task != null || (task = getTask()) != null) {  
  6.                    runTask(task);  
  7.                    task = null;  
  8.                }  
  9.            } finally {  
  10.                workerDone(this);  
  11.            }  
  12.        }  

 可以看出 worker 的 run 方法是一个循环，第一次循环运行的必然是 firstTask ，在运行完 firstTask 之后，并不会立刻结束，而是会调用 getTask 获取新的任务（ getTask 会从等待队列里获取等待中的任务），如果keepAliveTime 时间内得到新任务则继续执行，得不到新任务则那么线程才会退出。这样就保证了多个任务可以复用一个线程，而不是每次都创建新任务。 keepAliveTime 的逻辑在哪里实现的呢？主要是利用了 BlockingQueue的 poll 方法支持等待。可看 getTask 的代码段：

 

Java代码   
  1. if (state == SHUTDOWN)  // Help drain queue  
  2.     r = workQueue.poll();  
  3. else if (poolSize > corePoolSize || allowCoreThreadTimeOut)  
  4.     r = workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS);  
  5. else  
  6.     r = workQueue.take();  
2.ThreadFactory 和R ejectedExecutionHandler

 

ThreadFactory 和 RejectedExecutionHandler是ThreadPoolExecutor的两个属性，也 可以认为是两个简单的扩展点 . ThreadFactory 是创建线程的工厂。

默认的线程工厂会创建一个带有“ pool-poolNumber-thread-threadNumber ”为名字的线程，如果我们有特别的需要，如线程组命名、优先级等，可以定制自己的 ThreadFactory 。

RejectedExecutionHandler 是拒绝的策略。常见有以下几种：

AbortPolicy ：不执行，会抛出 RejectedExecutionException 异常。

CallerRunsPolicy ：由调用者（调用线程池的主线程）执行。

DiscardOldestPolicy ：抛弃等待队列中最老的。

DiscardPolicy: 不做任何处理，即抛弃当前任务。

 

3.ScheduledThreadPoolExecutor

ScheduleThreadPoolExecutor 是对ThreadPoolExecutor的集成。增加了定时触发线程任务的功能。需要注意  

从内部实现看， ScheduleThreadPoolExecutor 使用的是  corePoolSize  线程和一个无界队列的固定大小的池，所以调整 maximumPoolSize  没有效果。无界队列是一个内部自定义的 DelayedWorkQueue 。

ScheduleThreadPoolExecutor 线程池接收定时任务的方法是 schedule ，看看内部实现：

 

 

Java代码   
  1. public ScheduledFuture schedule(Runnable command,  
  2.                                    long delay,  
  3.                                    TimeUnit unit) {  
  4.     if (command == null || unit == null)  
  5.         throw new NullPointerException();  
  6.     RunnableScheduledFuture t = decorateTask(command,  
  7.         new ScheduledFutureTask(command, null,  
  8.                                       triggerTime(delay, unit)));  
  9.   
  10.     delayedExecute(t);  
  11.     return t;  
  12. }  

 

     以上代码会初始化一个 RunnableScheduledFuture 类型的任务 t, 并交给 delayedExecute 方法。delayedExecute(t) 方法实现如下：

 

   

Java代码   
  1.     private void delayedExecute(Runnable command) {  
  2.         if (isShutdown()) {  
  3.             reject(command);  
  4.             return;  
  5.         }  
  6.         if (getPoolSize() < getCorePoolSize())  
  7.             prestartCoreThread();  
  8.   
  9.         super.getQueue().add(command);  
  10. }  

 

 

如果当前线程池大小 poolSize 小于 CorePoolSize ，则创建一个新的线程，注意这里创建的线程是空的，不会把任务直接交给线程来做，而是把线程任务放到队列里。因为任务是要定时触发的，所以不能直接交给线程去执行。

问题： 那如何做到定时触发呢？

关键在于DelayedWorkQueue,它代理了 DelayQueue 。可以认为 DelayQueue 是这样一个队列（具体可以去看下源码，不详细分析）：

  1. 队列里的元素按照任务的 delay 时间长短升序排序， delay 时间短的在队头， delay 时间长的在队尾。
  2. 从 DelayQueue 里 FIFO 的获取一个元素的时候，不会直接返回 head 。可能会阻塞，等到 head 节点到达delay 时间后才能被获取。可以看下 DelayQueue 的 take 方法实现：

 

Java代码   
  1. public E take() throws InterruptedException {  
  2.     final ReentrantLock lock = this.lock;  
  3.     lock.lockInterruptibly();  
  4.     try {  
  5.         for (;;) {  
  6.             E first = q.peek();  
  7.             if (first == null) {  
  8.                 available.await();  
  9.             } else {  
  10.                 long delay =  first.getDelay(TimeUnit.NANOSECONDS);  
  11.                 if (delay > 0) {  
  12.                     long tl = available.awaitNanos(delay);//等待delay时间  
  13.                 } else {  
  14.                     E x = q.poll();  
  15.                     assert x != null;  
  16.                     if (q.size() != 0)  
  17.                         available.signalAll(); // wake up other takers  
  18.                     return x;  
  19.                 }  
  20.             }  
  21.         }  
  22.     } finally {  
  23.         lock.unlock();  
  24.     }  
  25. }  

 

 

4.线程池使用策略

通过以上的详解基本上能够定制出自己需要的策略了，下面简单介绍下Executors里面提供的一些常见线程池策略：

1.FixedThreadPool

Java代码   
  1. public static ExecutorService newFixedThreadPool(int nThreads) {  
  2.     return new ThreadPoolExecutor(nThreads, nThreads,  
  3.                                   0L, TimeUnit.MILLISECONDS,  
  4.                                   new LinkedBlockingQueue());  
  5. }  

 实际上就是个不支持keepalivetime，且corePoolSize和maximumPoolSize相等的线程池。

2.SingleThreadExecutor

Java代码   
  1. public static ExecutorService newSingleThreadExecutor() {  
  2.     return new FinalizableDelegatedExecutorService  
  3.         (new ThreadPoolExecutor(1, 1,  
  4.                                 0L, TimeUnit.MILLISECONDS,  
  5.                                 new LinkedBlockingQueue()));  
  6. }  

  实际上就是个不支持keepalivetime，且corePoolSize和maximumPoolSize都等1的线程池。

3.CachedThreadPool

Java代码   
  1. public static ExecutorService newCachedThreadPool() {  
  2.     return new ThreadPoolExecutor(0, Integer.MAX_VALUE,  
  3.                                   60L, TimeUnit.SECONDS,  
  4.                                   new SynchronousQueue());  
  5. }  

 实际上就是个支持keepalivetime时间是60秒（线程空闲存活时间），且corePoolSize为0，maximumPoolSize无穷大的线程池。

4.SingleThreadScheduledExecutor

Java代码   
  1. public static ScheduledExecutorService newSingleThreadScheduledExecutor(ThreadFactory threadFactory) {  
  2.     return new DelegatedScheduledExecutorService  
  3.         (new ScheduledThreadPoolExecutor(1, threadFactory));  
  4. }  

 实际上是个corePoolSize为1的ScheduledExecutor。上文说过ScheduledExecutor采用无界等待队列，所以maximumPoolSize没有作用。

5.ScheduledThreadPool

Java代码   
  1. public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {  
  2.     return new ScheduledThreadPoolExecutor(corePoolSize);  
  3. }  

  实际上是corePoolSize课设定的ScheduledExecutor。上文说过ScheduledExecutor采用无界等待队列，所以maximumPoolSize没有作用。

 

以上还不一定满足你的需要，完全可以根据自己需要去定制。

 

 

5.参考资料

juc executor源码


推荐阅读
  • get

    优化Spring Boot项目，大幅提升并发性能

    本文探讨了如何通过一系列技术手段提升Spring Boot项目的并发处理能力，解决生产环境中因慢请求导致的系统性能下降问题。 ... [详细]
  • get

    深入解析Java多线程与并发库的应用：空中网实习生面试题详解

    深入解析Java多线程与并发库的应用：空中网实习生面试题详解
    本文详细探讨了Java多线程与并发库的高级应用，结合空中网在挑选实习生时的面试题目，深入分析了相关技术要点和实现细节。文章通过具体的代码示例展示了如何使用Semaphore和SynchronousQueue来管理线程同步和任务调度。 ... [详细]
  • get

    ListView简单使用

    ListView简单使用
    先上效果：主要实现了Listview的绑定和点击事件。项目资源结构如下：先创建一个动物类，用来装载数据：Animal类如下：packagecom.example.simplelis ... [详细]
  • get

    深入剖析JVM垃圾回收机制

    深入剖析JVM垃圾回收机制
    本文详细探讨了Java虚拟机（JVM）中的垃圾回收机制，包括其意义、对象判定方法、引用类型、常见垃圾收集算法以及各种垃圾收集器的特点和工作原理。通过理解这些内容，开发人员可以更好地优化内存管理和程序性能。 ... [详细]
  • jsp

    计算整数位数与基础编程示例

    本文介绍了如何通过Java代码计算一个整数的位数，并展示了多个基础编程示例，包括求和、平均分计算、条件判断等。 ... [详细]
  • spring

    深入理解Java多线程并发处理：基础与实践

    深入理解Java多线程并发处理：基础与实践
    本文探讨了Java中的多线程并发处理机制，从基本概念到实际应用，帮助读者全面理解并掌握多线程编程技巧。通过实例解析和理论阐述，确保初学者也能轻松入门。 ... [详细]
  • get

    Java异步编程实践

    本文详细介绍了Java中实现异步调用的多种方式，包括线程创建、Future接口、CompletableFuture类以及Spring框架的@Async注解。通过代码示例和深入解析，帮助读者理解并掌握这些技术。 ... [详细]
  • get

    Java 中的不可变集合与同步集合详解

    本文将详细探讨 Java 中提供的不可变集合（如 `Collections.unmodifiableXXX`）和同步集合（如 `Collections.synchronizedXXX`）的实现原理及使用方法，帮助开发者更好地理解和应用这些工具。 ... [详细]
  • get

    软件工程课堂测试2

    软件工程课堂测试2
    要做一个简单的保存网页界面，首先用jsp写出保存界面，本次界面比较简单，首先是三个提示语，后面是三个输入框，然 ... [详细]
  • get

    UNIX进程间通信（IPC）详解

    本文深入探讨了UNIX/Linux系统中的进程间通信（IPC）机制，包括消息传递、同步和共享内存等。详细介绍了管道（Pipe）、有名管道（FIFO）、Posix和System V消息队列、互斥锁与条件变量、读写锁、信号量以及共享内存的使用方法和应用场景。 ... [详细]
  • get

    java文本编辑器,java文本编辑器设计思路

    java文本编辑器,java文本编辑器设计思路
    java文本编辑器,java文本编辑器设计思路 ... [详细]
  • io

    Java期末复习：throw与throws的区别及其他核心知识点

    本文详细解析了Java中throw和throws的关键区别，同时涵盖了JDK的定义、Java虚拟机的关键约定、Java的跨平台性、自动垃圾回收机制、源文件结构、包的概念及作用等多个核心知识点，旨在帮助学生更好地准备Java期末考试。 ... [详细]
  • io

    利用jstack进行死锁检测与线程堆栈分析

    利用jstack进行死锁检测与线程堆栈分析
    本文介绍了如何使用jstack工具进行Java应用中的死锁检测及高CPU使用率线程的堆栈分析，帮助开发者快速定位并解决性能瓶颈。 ... [详细]
  • get

    Web与游戏开发的主要差异

    本文探讨了Web开发与游戏开发之间的主要区别，旨在帮助开发者更好地理解两种开发领域的特性和需求。文章基于作者的实际经验和网络资料整理而成。 ... [详细]
  • get

    《算法》第四章部分程序 part 19

    ▶书中第四章部分程序，包括在加上自己补充的代码，有边权有向图的邻接矩阵，FloydWarshall算法可能含负环的有边权有向图任意两点之间的最短路径●有边权有向图的邻接矩阵1 ... [详细]
author-avatar
lin
这个家伙很懒，什么也没留下！
Tags | 热门标签
RankList | 热门文章
PHP1.CN | 中国最专业的PHP中文社区 | DevBox开发工具箱 | json解析格式化 |PHP资讯 | PHP教程 | 数据库技术 | 服务器技术 | 前端开发技术 | PHP框架 | 开发工具 | 在线工具
Copyright © 1998 - 2020 PHP1.CN. All Rights Reserved | 京公网安备 11010802041100号 | 京ICP备19059560号-4 | PHP1.CN 第一PHP社区 版权所有