【网页编辑常用源码】【api自适应源码】【如何精简net源码】java executor 源码-皮皮网

【网页编辑常用源码】【api自适应源码】【如何精简net源码】java executor 源码

2024-11-23 02:25:08 来源：php 任务源码分类：焦点

1.Java的源码并行世界-3.0 线程池与拒绝策略
2.创建线程推荐使用哪些方法
3.java并发中ExecutorService的使用
4.Java 中的 4 种线程池
5.Java多线程——singleThreadExecutor
6.java线程池之ScheduledThreadPoolExecutor实现原理

java executor 源码

Java的并行世界-3.0 线程池与拒绝策略

Jdk并发相关核心类：java.util.concurrent

java.util.concurrent.Executors提供了一些静态工厂方法，用于创建不同类型的源码线程池，例如：

ExecutorService fixedThreadPool = Executors.newFixedThreadPool(5);

ExecutorService cachedThreadPool = Executors.newCachedThreadPool();

ExecutorService singleThreadExecutor = Executors.newSingleThreadExecutor();

ScheduledExecutorService scheduledThreadPool = Executors.newScheduledThreadPool();

可以通过new ThreadPoolExecutor()方法手动创建线程池，源码该方法需要传入四个参数，源码分别是源码核心线程数、最大线程数、源码网页编辑常用源码线程保活时间和任务队列。源码其中，源码核心线程数和最大线程数是源码必填参数，线程保活时间和任务队列是源码可选参数。

Java中的源码Executors共有四种创建方式，这些方式包括使用newFixedThreadPool、源码newCachedThreadPool、源码newSingleThreadExecutor和newScheduledThreadPool。源码在使用这些方法时，源码可以根据实际需求选择最适合的方式来创建线程池。无论哪种方式，线程池都可以有效地管理和控制线程，提高程序的执行效率。

新FixedThreadPool创建一个固定大小的线程池。

以下是一个Java中创建newFixedThreadPool的代码例子：

新CachedThreadPool创建一个根据需要自动扩展的线程池，线程数根据任务数量动态调整。

以下是一个newCachedThreadPool的Java代码示例：

新SingleThreadExecutor创建一个只有一个线程的线程池。

新ScheduledThreadPool创建一个支持定时任务的线程池。

ForkJoinPool是一个用于执行分而治之任务的线程池，特别适用于递归分解的问题，例如并行归并排序、并行求和等。

ForkJoinPool forkJoinPool = new ForkJoinPool();

ForkJoinPool在java.util.concurrent包下，从Java 7开始引入，专门用于处理需要递归分解的任务。它利用工作窃取（Work-Stealing）算法来实现高效的任务调度和线程利用，能够充分利用多核处理器的优势。

ForkJoinPool的主要特点包括：

下面是一个简单的使用ForkJoinPool的示例，计算斐波那契数列的值：

虽然上面5个线程池看上去功能特点不同，但是其内部实现原理都调用了JDK的ThreadPoolExecutor线程池类。

ThreadPoolExecutor的构造函数有多个参数，允许根据实际需求配置线程池的行为，主要包括：

corePoolSize：线程池的核心线程数，即线程池中保持的api自适应源码最小线程数量。即使线程处于空闲状态，核心线程也不会被销毁。

maximumPoolSize：线程池的最大线程数，即线程池中允许的最大线程数量。当任务数量超过核心线程数时，线程池会根据实际情况动态地创建新的线程，但不会超过最大线程数。

keepAliveTime：非核心线程的空闲时间超过这个时间后，会被销毁，从而控制线程池的大小。时间单位可以通过指定的TimeUnit来定义。

workQueue：任务队列，用于存放等待执行的任务。ThreadPoolExecutor支持多种类型的任务队列，如ArrayBlockingQueue、LinkedBlockingQueue等。

threadFactory：用于创建线程的工厂，可以自定义线程的创建方式。

handler：拒绝策略，当线程池和队列都满了，无法继续接受新的任务时，会触发拒绝策略来处理新的任务。常见的拒绝策略包括AbortPolicy（默认策略，直接抛出异常）、CallerRunsPolicy（由调用线程来执行被拒绝的任务）、DiscardPolicy（丢弃被拒绝的任务）、DiscardOldestPolicy（丢弃队列中最老的任务）。

ThreadPoolExecutor提供了submit()和execute()等方法来向线程池提交任务，其中：

submit()方法可以接受Callable和Runnable类型的任务，并返回一个Future对象，通过这个对象可以获取任务的执行结果或者取消任务。

execute()方法只能接受Runnable类型的任务，无法获取任务的返回结果。

ThreadPoolExecutor还提供了一些方法来管理和监控线程池的状态，如getActiveCount()、getCompletedTaskCount()、getTaskCount()等。

workQueue：任务队列，其中的如何精简net源码任务是被提交但尚未执行的任务。其类型是：BlockingQueue接口，只能用于存放Runable对象。

有界队列可以通过ArrayBlockingQueue实现，当有新任务到来时，如果线程池的实际线程数量小于corePoolSize，则会优先创建新的线程；如果大于corePoolSize，则会将新任务加入等待队列；若队列已满，无法加入，则在总线程数量不大于maximumPoolSize的前提下，创建新的线程执行任务；若大于maximumPoolSize,则执行拒绝策略。

无界队列可以通过LinkedBlockingQueue实现，与有界队列相比，除非系统资源耗尽，不然当有新的任务到来，并且线程数量小于corePoolSize时，线程池就会创建新的线程执行任务。但当线程数量大于corePoolSize后，就不会继续创建了。若还有任务进来，系统CPU没那么忙，还有线程资源，则任务直接进入队列等待。

优先任务队列可以通过PriorityBlockingQueue实现，可以根据任务的优先级执行任务，这是一种特殊的无界队列。

有界队列和无界队列需要做demo测试。新CachedThreadPool内部使用的是SynchronousQueue队列，这是一个直接提交队列，系统会增加新的线程执行任务，当任务执行完毕后，线程会被回收；如果开启大量任务提交，每个任务执行慢，系统会开启等量的线程处理，直到系统资源耗尽。

ThreadPoolExecutor的核心调度代码包括workerCountOf(c)来获取当前工作线程池的总数，addWorker（command）用于提交任务或创建线程池，以及reject（command）来执行拒绝策略。

Handler是RejectedExecutionHandler接口类型，代表了不同的股票axes指标源码拒绝策略。常见的拒绝策略包括：

CallerRunsPolicy：如果线程池未关闭，会直接在调用者当前线程执行等待队列放不下的任务。

AbortPolicy：会直接抛出异常，阻止系统正常运行。

DiscardPolicy：直接丢失无法放入等待队列的任务，不做异常抛出。

DiscardOldestPolicy：丢弃最老的一个请求，然后尝试把当前请求任务加入到等待队列。

注意：在创建ThreadPoolExecutor时需要指定拒绝策略，如果以上拒绝策略无法满足，可以继承RejectedExecutionHandler接口来实现自定义的拒绝策略。

最常用的是升级DiscardPolicy策略，但需要在放弃前记录请求；示例如下：

以下是RejectedExecutionHandler接口代码，可以重新实现该方法以满足自定义需求。

熟悉拒绝策略后，在线程池中还有重要参数ThreadFactory，用于控制线程的创建。通过ThreadFactory，可以实现以下功能：

命名线程：通过为线程指定有意义的名称，便于跟踪日志和调试信息。

设置线程属性：根据需要设置线程的优先级、守护状态、异常处理器等。

定制化线程创建逻辑：添加自定义逻辑来创建线程，如记录线程的创建次数、设置线程组等。

以下是简单的ThreadFactory示例：

创建线程推荐使用哪些方法

推荐使用的方法来创建线程是使用线程库中的高级接口，比如Java中的java.util.concurrent包中的ExecutorService和FutureTask，或者Python的concurrent.futures模块。

Java中的ExecutorService

在Java中，ExecutorService提供了一个高级别的线程池接口，使得线程的管理和控制更为方便。你可以创建固定大小的线程池，这样你就可以控制并发的线程数量。ExecutorService提供了submit和invokeAll等方法来提交任务，获取执行结果。

例如：

java

ExecutorService executor = Executors.newFixedThreadPool(5);

for(int i = 0; i < ; i++) {

executor.submit(() -> {

// 你的任务代码

});

}

executor.shutdown(); // 关闭线程池

Java中的FutureTask

FutureTask是Java中的一种可取消的异步计算。它提供了Future的基本实现，具有启动和取消计算、asp网页源码翻译查询计算是否完成以及检索计算的结果等方法。只有当计算完成时才能检索结果，如果计算尚未完成，则阻塞get方法。

例如：

java

FutureTask futureTask = new FutureTask>(() -> {

// 你的任务代码

return "任务结果";

});

new Thread(futureTask).start();

try {

System.out.println(futureTask.get()); // 输出："任务结果"

} catch (InterruptedException | ExecutionException e) {

e.printStackTrace();

}

Python中的concurrent.futures

在Python中，concurrent.futures模块实现了一个高级的接口，用于异步执行可调用的对象。线程池执行器ThreadPoolExecutor和进程池执行器ProcessPoolExecutor都实现了这个接口。两者都提供了map方法来异步执行函数的调用，并阻塞直到结果就绪。

例如：

python

from concurrent.futures import ThreadPoolExecutor

def task(n):

# 你的任务代码

return n * 2

with ThreadPoolExecutor(max_workers=5) as executor:

results = executor.map(task, range())

print(results) # 输出：[0, 2, 4, 6, 8, , , , , ]

以上这些高级线程接口，可以帮助我们更方便地管理和控制线程，提高程序的效率和稳定性。当然，实际的使用情况会根据你的应用程序的具体需求和上下文环境进行调整。

java并发中ExecutorService的使用

ExecutorService是Java中用于异步执行的框架，通过它能方便地创建多线程执行环境。本文将详细介绍ExecutorService的使用方法。

创建ExecutorService通常有两种方式。一种是使用Executors中的工厂类方法，如newFixedThreadPool。此外，Executors提供了多种创建ExecutorService的方法。另一种方式是直接创建一个ExecutorService实例，因为ExecutorService是一个接口，需要实例化它的实现类，如ThreadPoolExecutor。

在为ExecutorService分配任务时，可以使用execute()、submit()、invokeAny()和invokeAll()方法。execute()方法用于提交Runnable任务，返回值为void。submit()方法可以提交Runnable或Callable任务，根据提交的对象类型返回Future或null。对于Callable任务，可以使用invokeAny()获取第一个成功返回的结果，使用invokeAll()获取所有任务成功执行的结果。

关闭ExecutorService需要调用shutdown()或shutdownNow()方法。shutdown()方法会立即销毁ExecutorService，停止接收新任务，并等待现有任务全部执行完毕再销毁。shutdownNow()方法不保证所有任务执行完毕，返回未执行任务列表。最佳关闭方法是shutdown()后使用awaitTermination等待所有任务执行完毕，若超时则调用shutdownNow()强制结束。

Future对象由submit()和invokeAll()方法返回，用于处理Callable任务的结果。获取结果时，需先调用get()方法，捕获可能的InterruptedException和ExecutionException异常。

ScheduledExecutorService提供定时执行任务的机制。使用方式与创建ExecutorService类似，通过schedule方法传入Runnable或Callable。还有两个相关方法，它们的区别在于计算周期的时间点不同。

ExecutorService与Fork/Join框架相比，允许用户更细粒度地控制线程生成，提供对线程的更多控制。而Fork/Join框架则旨在更高效地执行任务。更多相关代码和教程，请参阅指定链接和飞 dean的博客。

Java 中的 4 种线程池

Java中的线程池，其核心接口为ExecutorService，而非Executor。Executor负责执行线程，而实际线程池接口为ExecutorService。其设计旨在高效处理大量短期异步任务，提高程序性能。

创建一个可重用、固定线程数的线程池，共享无界队列方式运作。此池在大多数nThreads状态活动，提交额外任务需等待，直至可用线程出现。关闭前，任务始终执行。新线程在任何因失败导致的终止情况后补位。

另一个实例是可安排在指定延迟后执行或定期执行的任务线程池。这为复杂调度提供强大支持，简化定时任务管理。

使用Executors.newSingleThreadExecutor()构建线程池，实现单线程循环功能。单线程池在线程死亡或发生异常后能重新启动，持续执行任务。

Java多线程——singleThreadExecutor

singleThreadExecutor，Java中Executors类的一个静态方法，创建了一个线程池，该线程池仅包含一个核心线程。这意味着所有任务将由这一单一线程执行，形成单线程执行模式。若核心线程因异常停止，则将启动新的线程替代，确保服务不中断。此线程池特别设计确保任务执行顺序与提交顺序一致，提升程序执行流程的可预测性与稳定性。

创建singleThreadExecutor的代码示例如下：

在这个例子中，ThreadPoolExecutor的corePoolSize和maximumPoolSize的值均为1，明确指出线程池仅包含一个核心线程，且最大线程数同样为1，保证了线程的高效利用。缓冲队列采用的是LinkedBlockingQueue，这是一个无边界队列，用于存储等待执行的任务。

java线程池之ScheduledThreadPoolExecutor实现原理

java中异步周期任务调度有Timer，ScheduledThreadPoolExecutor等实现，目前单机版的定时调度都是使用ScheduledThreadPoolExecutor去实现，那么它是如何实现周期执行任务的呢？其实它还是利用ThreadPoolExecutor线程池去执行任务，这一点从它是继承自ThreadPoolExecutor救可以看的出来，其实关键在于如何实现任务的周期性调度，

ScheduledThreadPoolExecutor类以及核心函数

首先ScheduledThreadPoolExecutor是实现ScheduledExecutorService接口，它主要定义了四个方法:

周期调度一个Runnable的对象

周期调度一个Callable的对象

固定周期调度Runnable对象（不管上一次Runnable执行结束的时间，总是以固定延迟时间执行即上一个Runnable执行开始时候 + 延时时间 = 下一个Runnable执行的时间点）

以固定延迟调度unnable对象(当上一个Runnable执行结束后+固定延迟 = 下一个Runnable执行的时间点)

publicinterfaceScheduledExecutorServiceextendsExecutorService{ publicScheduledFuture<?>schedule(Runnablecommand,longdelay,TimeUnitunit);public<V>ScheduledFuture<V>schedule(Callable<V>callable,longdelay,TimeUnitunit);publicScheduledFuture<?>scheduleAtFixedRate(Runnablecommand,longinitialDelay,longperiod,TimeUnitunit);publicScheduledFuture<?>scheduleWithFixedDelay(Runnablecommand,longinitialDelay,longdelay,TimeUnitunit);}

其次,ScheduledThreadPoolExecutor是继承ThreadPoolExecutor,所以它是借助线程池的能力去执行任务，然后自身去实现周期性调度。从构造方法调用父类的线程池的构造方法，核心线程数是构造方法传入，这里可以看到最大线程数是Integer的最大值即, 还有等待队列是DelayedWorkQueue，它是实现延时的关键.

/***Createsanew{ @codeScheduledThreadPoolExecutor}withthe*givencorepoolsize.**@paramcorePoolSizethenumberofthreadstokeepinthepool,even*iftheyareidle,unless{ @codeallowCoreThreadTimeOut}isset*@throwsIllegalArgumentExceptionif{ @codecorePoolSize<0}*/publicScheduledThreadPoolExecutor(intcorePoolSize){ super(corePoolSize,Integer.MAX_VALUE,0,NANOSECONDS,newDelayedWorkQueue());}

scheduleAtFixedRate是实现周期性调度的方法，调度任务就是实现Runnable对象, 以及系统的开始延时时间，周期的调度的间隔时间。

计算初始触发时间和执行周期，并和传入的Runnable对象作为参数封装成 ScheduledFutureTask,然后调用decorateTask装饰Tas(默认实现为空)。

设置ScheduledFutureTask对象outerTask为t(默认就是它自己)。

调用delayedExecute延迟执行任务。

publicScheduledFuture<?>scheduleAtFixedRate(Runnablecommand,longinitialDelay,longperiod,**TimeUnitunit){ if(command==null||unit==null)thrownewNullPointerException();if(period<=0)thrownewIllegalArgumentException();ScheduledFutureTask<Void>sft=newScheduledFutureTask<Void>(command,null,triggerTime(initialDelay,unit),unit.toNanos(period));RunnableScheduledFuture<Void>t=decorateTask(command,sft);sft.outerTask=t;delayedExecute(t);returnt;}

判断线程池状态，如果不是处于running状态，则拒绝该任务。

将该任务加入父类的延迟队列(实际为初始化的DelayedWorkQueue对象)

再次判断线程池不是处于running状态，并且，判断是否是处于shutdown状态并且continueExistingPeriodicTasksAfterShutdown标志是否是true(默认是false，表示是否线程次处于shutdown状态下是否继续执行周期性任务),若果为true，则从队列删除任务，false，则确保启动线程来执行周期性任务

privatevoiddelayedExecute(RunnableScheduledFuture<?>task){ if(isShutdown())reject(task);else{ super.getQueue().add(task);if(isShutdown()&&!canRunInCurrentRunState(task.isPeriodic())&&remove(task))task.cancel(false);elseensurePrestart();}}

获取线程池数量

如果小于核心线程数，则启动核心线程执行任务，如果线程数为空,则启动非核心线程

voidensurePrestart(){ intwc=workerCountOf(ctl.get());if(wc<corePoolSize)addWorker(null,true);elseif(wc==0)addWorker(null,false);}ScheduledFutureTask的run函数

获取是否是周期性任务

判断是否线程池状态是否可以执行任务，如果为true，则取消任务 3 如果是非周期性任务，则直接调用父类FutureTask的run方法， 4 如果是周期性任务，则调用FutureTask的runAndReset函数，如果该函数返回为true，则调用setNextRunTime设置下一次运行的时间，并且还行reExecutePeriodic再次执行周期性任务。

publicvoidrun(){ booleanperiodic=isPeriodic();if(!canRunInCurrentRunState(periodic))cancel(false);elseif(!periodic)ScheduledFutureTask.super.run();elseif(ScheduledFutureTask.super.runAndReset()){ setNextRunTime();reExecutePeriodic(outerTask);}}

判断线程池是否处于可执行任务的状态，如果为true，则重新将设置下一次运行时间的任务加入父类的等待队列，

如果线程池处于不可运行任务的状态，则并且从等待队列中移除成功，调用任务的取消操作，否则调用ensurePrestart确保启动线程执行任务

voidreExecutePeriodic(RunnableScheduledFuture<?>task){ if(canRunInCurrentRunState(true)){ super.getQueue().add(task);if(!canRunInCurrentRunState(true)&&remove(task))task.cancel(false);elseensurePrestart();}}DelayedWorkQueue类核心函数

DelayedWorkQueue是继承AbstractQueue，并实现BlockingQueue接口

staticclassDelayedWorkQueueextendsAbstractQueue<Runnable>implementsBlockingQueue<Runnable>{

核心字段

//初始容量为privatestaticfinalintINITIAL_CAPACITY=;//等待队列，只能保存RunnableScheduledFuture对象privateRunnableScheduledFuture<?>[]queue=newRunnableScheduledFuture<?>[INITIAL_CAPACITY];//锁privatefinalReentrantLocklock=newReentrantLock();//对俄大小privateintsize=0;//leader线程，表示最近需要执行的任务的线程。privateThreadleader=null;//条件锁privatefinalConditionavailable=lock.newCondition();

offer函数:

将添加的参数转换成RunnableScheduledFuture对象。

加全局锁。

获取当前队列的size，如果等于队列的长度，则嗲用grow扩容，增加%的数组长度。

size加1。

如果数组为0，则将加入的对象放在索引为0的位置，然后设置ScheduledFutureTask的heapIndex的索引(便于后续快速删除)。

调用siftUp做堆的上浮操作，这里是小根堆的操作。

如果队列中第一个元素是传入的对象，则将laader设置null

释放锁

返回true

publicbooleanoffer(Runnablex){ if(x==null)thrownewNullPointerException();RunnableScheduledFuture<?>e=(RunnableScheduledFuture<?>)x;finalReentrantLocklock=this.lock;lock.lock();try{ inti=size;if(i>=queue.length)grow();size=i+1;if(i==0){ queue[0]=e;setIndex(e,0);}else{ siftUp(i,e);}if(queue[0]==e){ leader=null;available.signal();}}finally{ lock.unlock();}returntrue;}

siftUp主要就是做小根堆的上移操作，从if (key.compareTo(e) >= 0) 看出，如果key大于parent索引的元素，则停止。

poll函数

加锁

获取队列中索引为0的云元素，若果为null或者第一个元素的执行时间戳时间大于当前时间则直接返回null,否则调用finishPoll将第一个元素返回.

释放锁

将队列size 减 1

获取队列中队列中最后一个元素，并且设置队列最后一个为null

最后一个元素不为null，则调用sfitdown进行，将最后一个元素设置到索引为0的位置，将下移操作，重新调整小根堆。

ScheduledFutureTask的heapIndex为-1

ScheduledFutureTask实现compareTo方法逻辑

首先比较是否是同一个对象

若果是ScheduledFutureTask对象，则比较time的大小，time是下一次执行的任务的时间戳，如果不是，则比较 getDelay的时间大小

ScheduledThreadPoolExecutor的take函数就是ThreadPoolExecutor的从任务队列中获取任务,没有任务则一直等待(这里是线程数小于核心线程数的情况)

加可中断锁

获取队列中第一个元素的任务，从前面可以知道此任务执行的时间戳最小的任务

如果第一个任务为空，则再全局的锁的条件锁上等待，

如果第一个任务不为空，则获取延迟时间，如果延时时间小于0，说明第一个任务已经到时间了，则返回第一个任务。

如果leader线程不为空，则让线程在全局锁的条件锁上等待

如果leader为空，则将获取第一个任务的当前线程赋值为leader变量。

在全局锁的条件锁上等待delay纳秒, 等待结束后，如果当前线程还是等于leader线程，则重置leader为空

最后判断 leader为空并且第一个任务不为空，则唤醒全局锁上条件锁的等待的线程。

释放全局锁。

总结\ 综合前面所述，线程池从DelayedWorkQueue每次取出的任务就是延迟时间最小的任务, 若果到达时间的任务,则执行任务，否则则用条件锁Conditon的wait进行等待,执行完后，则用signal进行唤醒下一个任务的执行。

java中executor是什么意思

Executor框架是Java中用于统一创建与运行的接口，它基于Executor接口和其子接口构建，实现线程池功能，将工作单元与执行机制分离。Executor框架通过ExecutorService接口提供强大的功能，包括关闭线程池、提交线程获取执行结果和控制线程执行。Executor框架提供了两个主要线程池实现：ThreadPoolExecutor和ScheduledThreadPoolExecutor，分别适用于普通任务和定时任务的处理。

Executor框架通过ExecutorService接口实现线程池功能，提供丰富方法如关闭线程池、提交任务获取执行结果和控制任务执行。Executor框架还提供灵活的配置线程池实现，如通过Executors工厂方法创建指定配置的线程池，并通过实用方法简化使用。

总之，了解Executor框架的基本内容，特别是它在处理线程池应用中的作用，将有助于进一步掌握Executor框架的使用方法。掌握这些知识后，用户可以更高效地利用Executor框架提供的功能，实现复杂的并发任务处理。

文章所属分类：焦点频道，点击进入>>

【网页编辑常用源码】【api自适应源码】【如何精简net源码】java executor 源码

重点关注