皮皮网

1.深入理解 RxJava2：Scheduler（2）
2.硬核干货：4W字从源码上分析JUC线程池ThreadPoolExecutor的码分实现原理
3.Java原理系列ScheduledThreadPoolExecutor原理用法示例源码详解
4.带你学会区分ScheduledThreadPoolExecutor与Timer

深入理解 RxJava2：Scheduler（2）

       欢迎来到深入理解 RxJava2 系列第二篇，本文基于 RxJava 2.2.0 正式版源码，码分将探讨 Scheduler 与 Worker 的码分概念及其实现原理。

       Scheduler 与 Worker 在 RxJava2 中扮演着至关重要的码分角色，它们是码分线程调度的核心与基石。虽然 Scheduler 的码分易语言蜘蛛纸牌源码作用较为熟悉，但 Worker 的码分概念了解的人可能较少。为何在已有 Scheduler 的码分情况下，还要引入 Worker 的码分概念呢？让我们继续探讨。

       首先，码分Scheduler 的码分核心定义是调度 Runnable，支持立即、码分延时和周期性调用。码分而 Worker 是码分任务的最小单元的载体。在 RxJava2 内部实现中，码分通常一个或多个 Worker 对应一个 ScheduledThreadPoolExecutor 对象，这里暂不深入探讨。

       在 RxJava 1.x 中，Scheduler 没有 scheduleDirect/schedulePeriodicallyDirect 方法，只能先创建 Worker，再通过 Worker 来调度任务。这些方法是对 Worker 调度的简化，可以理解为创建一个只能调度一次任务的 Worker 并立即调度该任务。在 Scheduler 基类的看源码手机软件源码中，默认实现是直接创建 Worker 并创建对应的 Task（虽然在部分 Scheduler 的覆盖实现上并没有创建 Worker，但可以认为存在虚拟的 Worker）。

       一个 Scheduler 可以创建多个 Worker，这两者是一对多的关系，而 Worker 与 Task 也是一对多的关系。Worker 的存在旨在确保两件事：统一调度 Runnable 和统一取消任务。例如，在 observeOn 操作符中，可以通过 Worker 来统一调度和取消一系列的 Runnable。

       RxJava2 默认内置了多种 Scheduler 实现，适用于不同场景，这些 Scheduler 都可以在 Schedulers 类中直接获得。以下是两个常用 Scheduler 的源码分析：computation 和 io。

       NewThreadWorker 在 computation、io 和 newThread 中都有涉及，下面简单了解一下这个类。NewThreadWorker 与 ScheduledThreadPoolExecutor 之间是一对一的关系，在构造函数中通过工厂方法创建一个 corePoolSize 为 1 的 ScheduledThreadPoolExecutor 对象并持有。

       ScheduledThreadPoolExecutor 从 JDK1.5 开始存在，这个类继承于 ThreadPoolExecutor，支持立即、延时和周期性任务。但是万炮捕鱼游戏源码注意，在 ScheduledThreadPoolExecutor 中，maximumPoolSize 参数是无效的，corePoolSize 表示最大线程数，且它的队列是无界的。这里不再深入探讨该类，否则会涉及太多内容。

       有了这个类，RxJava2 在实现 Worker 时就站在了巨人的肩膀上，线程调度可以直接使用该类解决，唯一的麻烦之处就是封装一层 Disposable 的逻辑。

       ComputationScheduler 是计算密集型的 Scheduler，其线程数与 CPU 核心数密切相关。当线程数远超过 CPU 核心数目时，CPU 的时间更多地损耗在了线程的上下文切换。因此，保持最大线程数与 CPU 核心数一致是比较通用的方式。

       FixedSchedulerPool 可以看作是固定数量的真正 Worker 的缓存池。确定了 MAX_THREADS 后，在 ComputationScheduler 的构造函数中会创建 FixedSchedulerPool 对象，FixedSchedulerPool 内部会直接创建一个长度为 MAX_THREADS 的 PoolWorker 数组。PoolWorker 继承自 NewThreadWorker，但没有任何额外的代码。

       PoolWorker 的源码和代码的关系使用方法是从池子里取一个 PoolWorker 并返回。但是需要注意，每个 Worker 是独立的，每个 Worker 内部的任务是绑定在这个 Worker 中的。如果按照上述方法暴露 PoolWorker，会出现两个问题：

       为了解决上述问题，需要在 PoolWorker 外再包一层 EventLoopWorker。EventLoopWorker 是一个代理对象，它会将 Runnable 代理给 FixedSchedulerPool 中取到的 PoolWorker 来调度，并负责管理通过它创建的任务。当自身被取消时，会将创建的任务全部取消。

       与 ComputationScheduler 恰恰相反，IoScheduler 的线程数是无上限的。这是因为 IO 设备的速度远低于 CPU 速度，在等待 IO 操作时，CPU 往往是闲置的。因此，应该创建更多的线程让 CPU 尽可能地利用。当然，并不是线程越多越好，线程数目膨胀到一定程度会影响 CPU 的效率，也会消耗大量的内存。在 IoScheduler 中，主力快速吸筹源码每个 Worker 在空置一段时间后就会被清除以控制线程的数目。

       CachedWorkerPool 是一个变长并定期清理的 ThreadWorker 的缓存池，内部通过一个 ConcurrentLinkedQueue 维护。和 PoolWorker 类似，ThreadWorker 也是继承自 NewThreadWorker。仅仅是增加了一个 expirationTime 字段，用来标识这个 ThreadWorker 的超时时间。

       在 CachedWorkerPool 初始化时，会传入 Worker 的超时时间，目前是写死的 秒。这个超时时间表示 ThreadWorker 闲置后最大存活时间（实际中不保证 秒时被回收）。

       IoScheduler 中也存在一个 EventLoopWorker 类，它和 ComputationScheduler 中的作用类似。因为 CachedWorkerPool 是每隔 秒清理一次队列的，所以 ThreadWorker 的存活时间取决于入队的时机。如果一直没有被再次取出，其被实际清理的延迟在 - 秒之间。

       熟悉线程的读者会发现，ComputationScheduler 与 IoScheduler 很像某些参数下的 ThreadPoolExecutor。它们对线程的控制外在表现很相似，但实际的线程执行对象不一样。这两者的对比有助于我们更深刻地理解 Scheduler 设计的内在逻辑。

       Scheduler 是 RxJava 线程的核心概念，RxJava 基于此屏蔽了 Thread 相关的概念，只与 Scheduler/Worker/Runnable 打交道。

       本来计划继续基于 Scheduler 和大家一起探讨 subscribeOn 与 observeOn，但考虑到篇幅问题，这些留待下篇分享。

       感谢大家的阅读，欢迎关注笔者的公众号，可以第一时间获取更新，同时欢迎留言沟通。

硬核干货：4W字从源码上分析JUC线程池ThreadPoolExecutor的实现原理

       深入剖析JUC线程池ThreadPoolExecutor的执行核心

       早有计划详尽解读ThreadPoolExecutor的源码，因事务繁忙未能及时整理。在之前的文章中，我们曾提及Doug Lea设计的Executor接口，其顶层方法execute()是线程池扩展的基础。本文将重点关注ThreadPoolExecutor#execute()的实现，结合简化示例，逐步解析。

       ThreadPoolExecutor的核心功能包括固定的核心线程、额外的非核心线程、任务队列和拒绝策略。它的设计巧妙地运用了JUC同步器框架AbstractQueuedSynchronizer(AQS)，以及位操作和CAS技术。以核心线程为例，设计上允许它们在任务队列满时阻塞，或者在超时后轮询，而非核心线程则在必要时创建。

       创建ThreadPoolExecutor时，我们需要指定核心线程数、最大线程数、任务队列类型等。当核心线程和任务队列满载时，会尝试添加额外线程处理新任务。线程池的状态控制至关重要，通过整型变量ctl进行管理和状态转换，如RUNNING、SHUTDOWN、STOP等，状态控制机制包括工作线程上限数量的位操作。

       接下来，我们深入剖析execute()方法。首先，方法会检查线程池状态和工作线程数量，确保在需要时添加新线程。这里涉及一个疑惑：为何需要二次检查？这主要是为了处理任务队列变化和线程池状态切换。任务提交流程中，addWorker()方法负责创建工作线程，其内部逻辑复杂，包含线程中断和适配器Worker的创建。

       Worker内部类是线程池核心，它继承自AQS，实现Runnable接口。Worker的构造和run()方法共同确保任务的执行，同时处理线程中断和生命周期的终结。getTask()方法是工作线程获取任务的关键，它会检查任务队列状态和线程池大小，确保资源的有效利用。

       线程池关闭操作通过shutdown()、shutdownNow()和awaitTermination()方法实现，它们涉及线程中断、任务队列清理和状态更新等步骤，以确保线程池的有序退出。在这些方法中，可重入锁mainLock和条件变量termination起到了关键作用，保证了线程安全。

       ThreadPoolExecutor还提供了钩子方法，允许开发者在特定时刻执行自定义操作。除此之外，它还包含了监控统计、任务队列操作等实用功能，每个功能的实现都是对execute()核心逻辑的扩展和优化。

       总的来说，ThreadPoolExecutor的execute()方法是整个线程池的核心，它的实现原理复杂而精细。后续将陆续分析ExecutorService和ScheduledThreadPoolExecutor的源码，深入探讨线程池的扩展和调度机制。敬请关注，期待下文的详细解析。

Java原理系列ScheduledThreadPoolExecutor原理用法示例源码详解

       ScheduledThreadPoolExecutor是Java中实现定时任务与周期性执行任务的高效工具。它继承自ThreadPoolExecutor类，能够提供比常规Timer类更强大的灵活性与功能，特别是在需要多个工作线程或有特殊调度需求的场景下。

       该类主要功能包含但不限于提交在指定延迟后执行的任务，以及按照固定间隔周期执行的任务。它实现了ScheduledExecutorService接口，进而提供了丰富的API以实现任务的调度与管理。其中包括now()、getDelay()、compareTo()等方法，帮助开发者更精确地处理任务调度与延迟。

       在实际应用中，ScheduledThreadPoolExecutor的使用案例广泛。比如，初始化一个ScheduledThreadPoolExecutor实例，设置核心线程数，从而为定时任务提供资源保障。提交延迟任务，例如在5秒后执行特定操作，并输出相关信息。此外，提交周期性任务，如每隔2秒执行一次特定操作，用于实时监控或数据更新。最后，通过调用shutdown()与shutdownNow()方法来关闭执行器并等待所有任务完成，确保系统资源的合理释放与任务的有序结束。

       总的来说，ScheduledThreadPoolExecutor在处理需要精确时间控制的任务时展现出了强大的功能与灵活性，是Java开发者在实现定时与周期性任务时的首选工具。

带你学会区分ScheduledThreadPoolExecutor与Timer

       æ‘˜è¦ï¼šæœ¬æ–‡ç®€å•ä»‹ç»ä¸‹ScheduledThreadPoolExecutor类与Timer类的区别，ScheduledThreadPoolExecutor类相比于Timer类来说，究竟有哪些优势，以及二者分别实现任务调度的简单示例。

       JDK1.5开始提供ScheduledThreadPoolExecutor类，ScheduledThreadPoolExecutor类继承ThreadPoolExecutor类重用线程池实现了任务的周期性调度功能。在JDK1.5之前，实现任务的周期性调度主要使用的是Timer类和TimerTask类。本文，就简单介绍下ScheduledThreadPoolExecutor类与Timer类的区别，ScheduledThreadPoolExecutor类相比于Timer类来说，究竟有哪些优势，以及二者分别实现任务调度的简单示例。

二者的区别线程角度

       Timer是单线程模式，如果某个TimerTask任务的执行时间比较久，会影响到其他任务的调度执行。

       ScheduledThreadPoolExecutor是多线程模式，并且重用线程池，某个ScheduledFutureTask任务执行的时间比较久，不会影响到其他任务的调度执行。

系统时间敏感度

       Timer调度是基于操作系统的绝对时间的，对操作系统的时间敏感，一旦操作系统的时间改变，则Timer的调度不再精确。

       ScheduledThreadPoolExecutor调度是基于相对时间的，不受操作系统时间改变的影响。

是否捕获异常

       Timer不会捕获TimerTask抛出的异常，加上Timer又是单线程的。一旦某个调度任务出现异常，则整个线程就会终止，其他需要调度的任务也不再执行。

       ScheduledThreadPoolExecutor基于线程池来实现调度功能，某个任务抛出异常后，其他任务仍能正常执行。

任务是否具备优先级

       Timer中执行的TimerTask任务整体上没有优先级的概念，只是按照系统的绝对时间来执行任务。

       ScheduledThreadPoolExecutor中执行的ScheduledFutureTask类实现了java.lang.Comparable接口和java.util.concurrent.Delayed接口，这也就说明了ScheduledFutureTask类中实现了两个非常重要的方法，一个是java.lang.Comparable接口的compareTo方法，一个是java.util.concurrent.Delayed接口的getDelay方法。在ScheduledFutureTask类中compareTo方法实现了任务的比较，距离下次执行的时间间隔短的任务会排在前面，也就是说，距离下次执行的时间间隔短的任务的优先级比较高。而getDelay方法则能够返回距离下次任务执行的时间间隔。

是否支持对任务排序

       Timer不支持对任务的排序。

       ScheduledThreadPoolExecutor类中定义了一个静态内部类DelayedWorkQueue，DelayedWorkQueue类本质上是一个有序队列，为需要调度的每个任务按照距离下次执行时间间隔的大小来排序

能否获取返回的结果

       Timer中执行的TimerTask类只是实现了java.lang.Runnable接口，无法从TimerTask中获取返回的结果。

       ScheduledThreadPoolExecutor中执行的ScheduledFutureTask类继承了FutureTask类，能够通过Future来获取返回的结果。

       é€šè¿‡ä»¥ä¸Šå¯¹ScheduledThreadPoolExecutor类和Timer类的分析对比，相信在JDK1.5之后，就没有使用Timer来实现定时任务调度的必要了。

二者简单的示例

       è¿™é‡Œï¼Œç»™å‡ºä½¿ç”¨Timer和ScheduledThreadPoolExecutor实现定时调度的简单示例，为了简便，我这里就直接使用匿名内部类的形式来提交任务。

Timer类简单示例

       æºä»£ç ç¤ºä¾‹å¦‚下所示。

packageio.binghe.concurrent.lab;importjava.util.Timer;importjava.util.TimerTask;/***@authorbinghe*@version1.0.0*@description测试Timer*/publicclassTimerTest{ publicstaticvoidmain(String[]args)throwsInterruptedException{ Timertimer=newTimer();timer.scheduleAtFixedRate(newTimerTask(){ @Overridepublicvoidrun(){ System.out.println("测试Timer类");}},,);Thread.sleep();timer.cancel();}}

       è¿è¡Œç»“果如下所示。

测试Timer类测试Timer类测试Timer类测试Timer类测试Timer类测试Timer类测试Timer类测试Timer类测试Timer类测试Timer类ScheduledThreadPoolExecutor类简单示例

       æºä»£ç ç¤ºä¾‹å¦‚下所示。

packageio.binghe.concurrent.lab;importjava.util.concurrent.*;/***@authorbinghe*@version1.0.0*@description测试ScheduledThreadPoolExecutor*/publicclassScheduledThreadPoolExecutorTest{ publicstaticvoidmain(String[]args)throwsInterruptedException{ ScheduledExecutorServicescheduledExecutorService=Executors.newScheduledThreadPool(3);scheduledExecutorService.scheduleAtFixedRate(newRunnable(){ @Overridepublicvoidrun(){ System.out.println("测试测试ScheduledThreadPoolExecutor");}},1,1,TimeUnit.SECONDS);//主线程休眠秒Thread.sleep();System.out.println("正在关闭线程池...");//关闭线程池scheduledExecutorService.shutdown();booleanisClosed;//等待线程池终止do{ isClosed=scheduledExecutorService.awaitTermination(1,TimeUnit.DAYS);System.out.println("正在等待线程池中的任务执行完成");}while(!isClosed);System.out.println("所有线程执行结束，线程池关闭");}}

       è¿è¡Œç»“果如下所示。

测试测试ScheduledThreadPoolExecutor测试测试ScheduledThreadPoolExecutor测试测试ScheduledThreadPoolExecutor测试测试ScheduledThreadPoolExecutor测试测试ScheduledThreadPoolExecutor测试测试ScheduledThreadPoolExecutor测试测试ScheduledThreadPoolExecutor测试测试ScheduledThreadPoolExecutor测试测试ScheduledThreadPoolExecutor正在关闭线程池...测试测试ScheduledThreadPoolExecutor正在等待线程池中的任务执行完成所有线程执行结束，线程池关闭

       æ³¨æ„ï¼šå…³äºŽTimer和ScheduledThreadPoolExecutor还有其他的使用方法，这里，我就简单列出以上两个使用示例，更多的使用方法大家可以自行实现。

       æœ¬æ–‡åˆ†äº«è‡ªåŽä¸ºäº‘社区《【高并发】ScheduledThreadPoolExecutor与Timer的区别和简单示例》，作者：冰河。