å¸¦ä½ å¦ä¼åºå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çåºå«åç®å示ä¾ãï¼ä½è ï¼å°æ²³ã
java线程池(一):java线程池基本使用及Executors
@[toc] 在前面学习线程组的时候就提到过线程池。实际上线程组在我们的解读日常工作中已经不太会用到,但是源码线程池恰恰相反,是解读我们日常工作中必不可少的工具之一。现在开始对线程池的源码使用,以及底层ThreadPoolExecutor的解读宁夏苹果源码源码进行分析。1.为什么需要线程池我们在前面对线程基础以及线程的源码生命周期有过详细介绍。一个基本的解读常识就是,线程是源码一个特殊的对象,其底层是解读依赖于JVM的native方法,在jvm虚拟机内部实现的源码。线程与普通对象不一样的解读地方在于,除了需要在堆上分配对象之外,源码还需要给每个线程分配一个线程栈、解读以及本地方法栈、源码程序计数器等线程的私有空间。线程的初始化工作相对于线程执行的大多数任务而言,都是一个耗时比较长的工作。这与数据库使用一样。有时候我们连接数据库,仅仅只是为了执行一条很小的sql语句。但是在我们日常的开发工作中,我们的绝大部分工作内容,都会分解为一个个短小的执行任务来执行。这样才能更加合理的复用资源。这种思想就与我们之前提到的协程一样。任务要尽可能的小。但是在java中,任务不可能像协程那样拆分得那么细。那么试想,如果说,有一个已经初始化好的dokan源码下载很多线程,在随时待命,那么当我们有任务提交的时候,这些线程就可以立即工作,无缝接管我们的任务请求。那么效率就会大大增加。这些个线程可以处理任何任务。这样一来我们就把实际的任务与线程本身进行了解耦。从而将这些线程实现了复用。 这种复用的一次创建,可以重复使用的池化的线程对象就被成为线程池。 在线程池中,我们的线程是可以复用的,不用每次都创建一个新的线程。减少了创建和销毁线程的时间开销。 同时,线程池还具有队列缓冲策略,拒绝机制和动态线程管理。可以实现线程环境的隔离。当一个线程有问题的时候,也不会对其他的线程造成影响。 以上就是我们使用线程池的原因。一句话来概括就是资源复用,降低开销。
2.java中线程池的实现在java中,线程池的主要接口是Executor和ExecutorService在这两个接口中分别对线程池的行为进行了约束,最主要的是在ExecutorService。之后,线程池的实际实现类是AbstractExecutorService类。这个类有三个主要的实现类,ThreadpoolExecutorService、ForkJoinPool以及DelegatedExecutorService。源码安装glut
后面我们将对这三种最主要的实现类的源码以及实现机制进行分析。
3.创建线程的工厂方法Executors在java中, 已经给我们提供了创建线程池的工厂方法类Executors。通过这个类以静态方法的模式可以为我们创建大多数线程池。Executors提供了5种创建线程池的方式,我们先来看看这个类提供的工厂方法。
3.1 newFixedThreadPool/** * Creates a thread pool that reuses a fixed number of threads * operating off a shared unbounded queue.At any point, at most * { @code nThreads} threads will be active processing tasks. * If additional tasks are submitted when all threads are active, * they will wait in the queue until a thread is available. * If any thread terminates due to a failure during execution * prior to shutdown, a new one will take its place if needed to * execute subsequent tasks.The threads in the pool will exist * until it is explicitly { @link ExecutorService#shutdown shutdown}. * * @param nThreads the number of threads in the pool * @return the newly created thread pool * @throws IllegalArgumentException if { @code nThreads <= 0} */public static ExecutorService newFixedThreadPool(int nThreads) { return new ThreadPoolExecutor(nThreads, nThreads,0L, TimeUnit.MILLISECONDS,new LinkedBlockingQueue<Runnable>());}这个方法能够创建一个固定线程数量的无界队列的线程池。参数nthreads是最多可同时处理的活动的线程数。如果在所有线程都在处理任务的情况下,提交了其他的任务,那么这些任务将处于等待队列中。直到有一个线程可用为止。如果任何线程在关闭之前的执行过程中,由于失败而终止,则需要在执行后续任务的时候,创建一个新的线程来替换。线程池中的所有线程都将一直存在,直到显示的调用了shutdown方法。 上述方法能创建一个固定线程数量的线程池。内部默认的是使用LinkedBlockingQueue。但是需要注意的是,这个LinkedBlockingQueue底层是链表结构,其允许的最大队列长度为Integer.MAX_VALUE。
public LinkedBlockingQueue() { this(Integer.MAX_VALUE);}这样在使用的过程中如果我们没有很好的控制,那么就可能导致内存溢出,出现OOM异常。因此这种方式实际上已经不被提倡。我们在使用的过程中应该谨慎使用。 newFixedThreadPool(int nThreads, ThreadFactory threadFactory)方法:
/** * Creates a thread pool that reuses a fixed number of threads * operating off a shared unbounded queue, using the provided * ThreadFactory to create new threads when needed.At any point, * at most { @code nThreads} threads will be active processing * tasks.If additional tasks are submitted when all threads are * active, they will wait in the queue until a thread is * available.If any thread terminates due to a failure during * execution prior to shutdown, a new one will take its place if * needed to execute subsequent tasks.The threads in the pool will * exist until it is explicitly { @link ExecutorService#shutdown * shutdown}. * * @param nThreads the number of threads in the pool * @param threadFactory the factory to use when creating new threads * @return the newly created thread pool * @throws NullPointerException if threadFactory is null * @throws IllegalArgumentException if { @code nThreads <= 0} */public static ExecutorService newFixedThreadPool(int nThreads, ThreadFactory threadFactory) { return new ThreadPoolExecutor(nThreads, nThreads,0L, TimeUnit.MILLISECONDS,new LinkedBlockingQueue<Runnable>(),threadFactory);}这个方法与3.1中newFixedThreadPool(int nThreads)的方法的唯一区别就是,增加了threadFactory参数。在前面方法中,介绍html源码对于线程的创建是采用的默认实现Executors.defaultThreadFactory()。而在此方法中,可以根据需要自行定制。
3.2 newSingleThreadExecutor/** * Creates an Executor that uses a single worker thread operating * off an unbounded queue. (Note however that if this single * thread terminates due to a failure during execution prior to * shutdown, a new one will take its place if needed to execute * subsequent tasks.)Tasks are guaranteed to execute * sequentially, and no more than one task will be active at any * given time. Unlike the otherwise equivalent * { @code newFixedThreadPool(1)} the returned executor is * guaranteed not to be reconfigurable to use additional threads. * * @return the newly created single-threaded Executor */public static ExecutorService newSingleThreadExecutor() { return new FinalizableDelegatedExecutorService(new ThreadPoolExecutor(1, 1,0L, TimeUnit.MILLISECONDS,new LinkedBlockingQueue<Runnable>()));}此方法将会创建指有一个线程和一个无届队列的线程池。需要注意的是,如果这个执行线程在执行过程中由于失败而终止,那么需要在执行后续任务的时候,用一个新的线程来替换。 那么这样一来,上述线程池就能确保任务的顺序性,并且在任何时间都不会有多个线程处于活动状态。与newFixedThreadPool(1)不同的是,使用newSingleThreadExecutor返回的ExecutorService不能被重新分配线程数量。而使用newFixExecutor(1)返回的ExecutorService,其活动的线程的数量可以重新分配。后面专门对这个问题进行详细分析。 newSingleThreadExecutor(ThreadFactory threadFactory) 方法:
/** * Creates an Executor that uses a single worker thread operating * off an unbounded queue, and uses the provided ThreadFactory to * create a new thread when needed. Unlike the otherwise * equivalent { @code newFixedThreadPool(1, threadFactory)} the * returned executor is guaranteed not to be reconfigurable to use * additional threads. * * @param threadFactory the factory to use when creating new * threads * * @return the newly created single-threaded Executor * @throws NullPointerException if threadFactory is null */public static ExecutorService newSingleThreadExecutor(ThreadFactory threadFactory) { return new FinalizableDelegatedExecutorService(new ThreadPoolExecutor(1, 1,0L, TimeUnit.MILLISECONDS,new LinkedBlockingQueue<Runnable>(),threadFactory));}这个方法与3.3中newSingleThreadExecutor的区别就在于增加了一个threadFactory。可以自定义创建线程的方法。
3.3 newCachedThreadPool/** * Creates a thread pool that creates new threads as needed, but * will reuse previously constructed threads when they are * available.These pools will typically improve the performance * of programs that execute many short-lived asynchronous tasks. * Calls to { @code execute} will reuse previously constructed * threads if available. If no existing thread is available, a new * thread will be created and added to the pool. Threads that have * not been used for sixty seconds are terminated and removed from * the cache. Thus, a pool that remains idle for long enough will * not consume any resources. Note that pools with similar * properties but different details (for example, timeout parameters) * may be created using { @link ThreadPoolExecutor} constructors. * * @return the newly created thread pool */public static ExecutorService newCachedThreadPool() { return new ThreadPoolExecutor(0, Integer.MAX_VALUE,L, TimeUnit.SECONDS,new SynchronousQueue<Runnable>());}这个方法用来创建一个线程池,该线程池可以根据需要自动增加线程。以前的线程也可以复用。这个线程池通常可以提高很多执行周期短的异步任务的性能。对于execute将重用以前的构造线程。如果没有可用的线程,就创建一个 新的线程添加到pool中。秒内,如果该线程没有被使用,则该线程将会终止,并从缓存中删除。因此,在足够长的物流仓储源码时间内,这个线程池不会消耗任何资源。可以使用ThreadPoolExecutor构造函数创建具有类似属性但是详细信息不同的线程池。 ?需要注意的是,这个方法创建的线程池,虽然队列的长度可控,但是线程的数量的范围是Integer.MAX_VALUE。这样的话,如果使用不当,同样存在OOM的风险。比如说,我们使用的每个任务的耗时比较长,任务的请求又非常快,那么这样势必会造成在单位时间内创建了大量的线程。从而造成内存溢出。 newCachedThreadPool(ThreadFactory threadFactory)方法:
/** * Creates a thread pool that creates new threads as needed, but * will reuse previously constructed threads when they are * available, and uses the provided * ThreadFactory to create new threads when needed. * @param threadFactory the factory to use when creating new threads * @return the newly created thread pool * @throws NullPointerException if threadFactory is null */public static ExecutorService newCachedThreadPool(ThreadFactory threadFactory) { return new ThreadPoolExecutor(0, Integer.MAX_VALUE,L, TimeUnit.SECONDS,new SynchronousQueue<Runnable>(),threadFactory);}这个方法区别同样也是在于,增加了threadFactory可以自行指定线程的创建方式。
2.4 newScheduledThreadPool/** * Creates a thread pool that can schedule commands to run after a * given delay, or to execute periodically. * @param corePoolSize the number of threads to keep in the pool, * even if they are idle * @return a newly created scheduled thread pool * @throws IllegalArgumentException if { @code corePoolSize < 0} */public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) { return new ScheduledThreadPoolExecutor(corePoolSize);}创建一个线程池,该线程池可以将任务在指定的延迟时间之后运行。或者定期运行。这个方法返回的是ScheduledThreadPoolExecutor。这个类是ThreadPoolExecutor的子类。在原有线程池的的基础之上,增加了延迟和定时功能。我们在后面分析了ThreadPoolExecutor源码之后,再来分析这个类的源码。 与之类似的方法:
/** * Creates a thread pool that can schedule commands to run after a * given delay, or to execute periodically. * @param corePoolSize the number of threads to keep in the pool, * even if they are idle * @param threadFactory the factory to use when the executor * creates a new thread * @return a newly created scheduled thread pool * @throws IllegalArgumentException if { @code corePoolSize < 0} * @throws NullPointerException if threadFactory is null */public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize, ThreadFactory threadFactory) { return new ScheduledThreadPoolExecutor(corePoolSize, threadFactory);}通过这个方法,我们可以指定threadFactory。自定义线程创建的方式。 同样,我们还可以只指定一个线程:
public static ScheduledExecutorService newSingleThreadScheduledExecutor() { return new DelegatedScheduledExecutorService(new ScheduledThreadPoolExecutor(1));}public static ScheduledExecutorService newSingleThreadScheduledExecutor(ThreadFactory threadFactory) { return new DelegatedScheduledExecutorService(new ScheduledThreadPoolExecutor(1, threadFactory));}上述两个方法都可以实现这个功能,但是需要注意的是,这两个方法的返回在外层包裹了一个包装类。
3.5 newWorkStealingPool这种方式是在jdk1.8之后新增的。我们先来看看其源码:
public LinkedBlockingQueue() { this(Integer.MAX_VALUE);}0这个方法实际上返回的是ForkJoinPool。该方法创建了一
硬核干货: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开发者在实现定时与周期性任务时的首选工具。
ThreadPoolExecutor简介&源码解析
线程池是通过池化管理线程的高效工具,尤其在多核CPU时代,利用线程池进行并行处理任务有助于提升服务器性能。ThreadPoolExecutor是线程池的具体实现,它负责线程管理和任务管理,以及处理任务拒绝策略。这个类提供了多种功能,如通过Executors工厂方法配置,执行Runnable和Callable任务,维护任务队列,统计任务完成情况等。
创建线程池需要考虑关键参数,如核心线程数(任务开始执行时立即创建),最大线程数(任务过多时限制新线程生成),线程存活时间,任务队列大小,线程工厂以及拒绝策略。JDK提供了四种拒绝策略,如默认的AbortPolicy,当资源饱和时抛出异常。此外,线程池还提供了beforeExecute和afterExecute钩子函数,用于在任务执行前后执行自定义操作。
当任务提交到线程池时,会经历一系列处理流程,包括任务的执行和线程池状态的管理。例如,如果任务队列和线程池满,会根据拒绝策略处理新任务。使用线程池时,需注意线程池容量与状态的计算,以及线程池工作线程的动态调整。
示例中,自定义线程池并重写钩子函数,创建任务后向线程池提交,可以看到线程池如何根据配置动态调整资源。但要注意,如果任务过多且无法处理,可能会抛出异常。源码分析中,submit方法实际上是调用execute,而execute内部包含Worker类和runWorker方法的逻辑,包括任务的获取和执行。
线程池的容量上限并非Integer.MAX_VALUE,而是由ctl变量的低位决定。 Doug Lea的工具函数简化了ctl的操作,使得计算线程池状态和工作线程数更加便捷。通过深入了解ThreadPoolExecutor,开发者可以更有效地利用线程池提高应用性能。
2024-12-28 23:22
2024-12-28 23:13
2024-12-28 22:38
2024-12-28 22:30
2024-12-28 22:23