ThreadPerTaskExecutor
- ThreadPerTaskExecutor每一个任务的执行器(代理和命令模式)线程解耦(执行线程和创建线程)
public final class ThreadPerTaskExecutor implements Executor { //定义一个私有的线程工厂 private final ThreadFactory threadFactory; //定义一个ThreadPerTaskExecutor构造方法 public ThreadPerTaskExecutor(ThreadFactory threadFactory) { if (threadFactory == null) { throw new NullPointerException("threadFactory"); } this.threadFactory = threadFactory; } //这个是一个命令模式,这个execute方法的作用是:在将来的某个时间执行给定的命令。 该命令可以在Executor实现的判断下在新线程,池化线程或调用线程中执行 @Override public void execute(Runnable command) { threadFactory.newThread(command).start(); } }
ExecutorService
- ExecutorService 继承了Executor接口,增加了对自身生命周期管理的方法,同时提供了一个Future给命令者去获取命令的执行结果
public interface ExecutorService extends Executor { /** * 启动一个有序关闭,其中先前提交的任务将被执行,但不会接受任何新任务。 如果已经关闭,调用没有其他影响。此方法不会等待先前提交的任务完成执行。 使用awaitTermination来做到这一点。 */ void shutdown(); /** * 尝试停止所有正在执行的任务,停止等待任务的处理,并返回等待执行的任务列表。此方法不等待主动执行任务终止。 使用awaitTermination来做到这一点。除尽力尝试停止处理主动执行任务之外,没有任何保证。 例如,典型的实现将通过Thread.interrupt取消,因此任何无法响应中断的任务都可能永远不会终止。 */ List<Runnable> shutdownNow(); /** * 如果此执行程序已关闭,则返回true。 */ boolean isShutdown(); /** * 如果关闭后所有任务都已完成,则返回true。 请注意,除非先调用shutdown或shutdownNow,否则isTerminated永远不会为真。 */ boolean isTerminated(); /** * 阻止所有任务在关闭请求之后完成执行,或者发生超时,或者当前线程被中断,以先发生者为准。 */ boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException; /** * 提交值返回任务以执行并返回表示任务的挂起结果的Future。 Future的get方法将在成功完成后返回任务的结果。 如果您想立即阻止等待任务,可以使用结构形式为result = exec.submit(aCallable).get(); 注意:Executors类包含一组方法,这些方法可以将一些其他常见的类似闭包的对象(例如,java.security.PrivilegedAction)转换为Callable形式,以便可以提交它们。 */ <T> Future<T> submit(Callable<T> task); /** * 提交Runnable任务以执行并返回表示该任务的Future。 Future的get方法将在成功完成后返回给定的结果。 */ <T> Future<T> submit(Runnable task, T result); /** * 提交Runnable任务以执行并返回表示该任务的Future。 Future的get方法将在成功完成后返回null。 */ Future<?> submit(Runnable task); /* 执行给定的任务,返回完成所有状态和结果的Futures列表。 对于返回列表的每个元素,Future.isDone都为true。 请注意,已完成的任务可能正常终止或通过抛出异常终止。 如果在此操作正在进行时修改了给定集合,则此方法的结果是不确定的。 */ <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks) throws InterruptedException; /** *执行给定的任务,返回一个Futures列表,其中包含所有完成或超时到期时的状态和结果,以先发生者为准。 对于返回列表的每个元素,Future.isDone都为true。 返回时,未完成的任务将被取消。 请注意,已完成的任务可能正常终止或通过抛出异常终止。 如果在此操作正在进行时修改了给定集合,则此方法的结果是不确定的。 */ <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit) throws InterruptedException; /** * 执行给定的任务,返回已成功完成的任务的结果(即,不抛出异常),如果有的话。 在正常或特殊退货时,未完成的任务将被取消。 如果在此操作正在进行时修改了给定集合,则此方法的结果是不确定的。 */ <T> T invokeAny(Collection<? extends Callable<T>> tasks) throws InterruptedException, ExecutionException; /** * 执行给定的任务,返回已成功完成的任务的结果(即,不抛出异常),如果在给定的超时之前已经执行了任何操作。 在正常或特殊退货时,未完成的任务将被取消。 如果在此操作正在进行时修改了给定集合,则此方法的结果是不确定的。 */ <T> T invokeAny(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException; }
- invokeAny 被哪些类调用。
ScheduledExecutorService
- ScheduledExecutorService 继承了ExecutorService接口,增加了对定时任务的支持。
//创建和执行在给定延迟后启用的一次性操作。 public ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit); //创建并执行一个周期性操作,该操作在给定的初始延迟后首先启用,随后在给定的时间段内启用; 即执行将在initialDelay之后开始,然后是initialDelay + period,然后是initialDelay + 2 * period,依此类推。 如果任务的任何执行遇到异常,则后续执行被禁止。 否则,任务将仅通过取消或终止执行者来终止。 如果此任务的执行时间超过其周期,则后续执行可能会延迟,但不会同时执行。 public ScheduledFuture<?> scheduleAtFixedRate(@org.jetbrains.annotations.NotNull Runnable command, long initialDelay, long period,TimeUnit unit)
EventExecutorGroup
- EventExecutorGroup 继承了ScheduledExecutorService接口,对原来的ExecutorService的关闭接口提供了增强,提供了优雅的关闭接口。从接口名称上可以看出它是对多个EventExecutor的集合,提供了对多个EventExecutor的迭代访问接口。
SingleThreadEventExcutor
- SingleThreadEventExcutor(单线程)实现了ScheduledExecutorService接口,支持执行定时任务。得有个地方存放定时任务信息。类中的实现是delayedTaskQueue,它是一个PriorityQueue,也是一个BlockingQueue。不过它里面的元素不是按照先来后到的顺序存取的,而是按照各个元素的优先级判断的SingleThreadEventExecutor类中有一个实例变量Thread,它引用的就是当前Executor所拥有的那个thread对象
SingleThreadEventExcutor类用到AtomicIntegerFieldUpdater基于反射的实用程序,可以对指定类的指定volatile int字段进行原子更新。 此类设计用于原子数据结构,其中同一节点的多个字段独立地受原子更新的影响。
请注意,此类中compareAndSet方法的保证比其他原子类弱。 因为此类无法确保该字段的所有使用都适用于原子访问的目的,所以它只能保证在compareAndSet的其他调用和同一更新程序上设置的原子性。
@Override public void execute(Runnable task) { if (task == null) { throw new NullPointerException("task"); } // 这行是判断是否在循环事件里面,点进去会跳到AbstractEventExecutor的inEventLoop方法 boolean inEventLoop = inEventLoop(); //如果inEventLoop为true就把任务添加一个任务队列里 if (inEventLoop) { addTask(task); } else { startThread(); //启动线程 addTask(task);//添加任务队列 //判断是否关闭和移除任务 if (isShutdown() && removeTask(task)) { reject();//调用拒绝方法 } } //不是addTaskWakesUp 并且是wakesUpForTask if (!addTaskWakesUp && wakesUpForTask(task)) { wakeup(inEventLoop); //调用wakeup方法 } }
- startThread方法代码
//启动线程方法 private void startThread() { if (state == ST_NOT_STARTED) { //这行代码是AtomicIntegerFieldUpdaterImpl#Unsafe的compareAndSwapInt方法 if (STATE_UPDATER.compareAndSet(this, ST_NOT_STARTED, ST_STARTED)) { doStartThread(); } } } private void doStartThread() { assert thread == null; //断言一个线程变量,初始化一个null //Executor执行者 executor.execute(new Runnable() { @Override public void run() { thread = Thread.currentThread(); //获取当前线程 if (interrupted) { //终止线程(暴力处理) thread.interrupt(); } boolean success = false; //更新内部时间戳,该时间戳指示最近执行提交的任务的时间。 runAllTasks()和runAllTasks(long)自动更新此时间戳,因此通常不需要调用此方法。 但是,如果使用takeTask()或pollTask()手动执行任务,则必须在任务执行循环结束时调用此方法以进行准确的静默期检查。 updateLastExecutionTime(); try { SingleThreadEventExecutor.this.run(); success = true; } catch (Throwable t) { logger.warn("Unexpected exception from an event executor: ", t); } finally { //死循环 for (;;) { int oldState = state; if (oldState >= ST_SHUTTING_DOWN || STATE_UPDATER.compareAndSet( SingleThreadEventExecutor.this, oldState, ST_SHUTTING_DOWN)) { break; } } // 检查在循环结束时是否调用了confirmShutdown()。 if (success && gracefulShutdownStartTime == 0) { logger.error("Buggy " + EventExecutor.class.getSimpleName() + " implementation; " + SingleThreadEventExecutor.class.getSimpleName() + ".confirmShutdown() must be called " + "before run() implementation terminates."); } try { // 运行所有剩余的任务并关闭挂钩。 for (;;) { //确认关闭 if (confirmShutdown()) { break; } } } finally { try { cleanup();//清除 } finally { STATE_UPDATER.set(SingleThreadEventExecutor.this, ST_TERMINATED); threadLock.release();//释放 if (!taskQueue.isEmpty()) { logger.warn( "An event executor terminated with " + "non-empty task queue (" + taskQueue.size() + ')'); } // terminationFuture.setSuccess(null); } } } } }); }
上面Thread内部run方法执行的是
SingleThreadEventExecutor.this.run(),而这个run方法是一个抽象方法,留给了子类去实现了。不过可以肯定的是子类的run方法是不断的去tasksQueue中取出task去执行。现在重点分析下finally块中的代码。
1、首先更改状态为正在关闭状态。
2、如果子类中的run方法中的loop执行成功了,就得先调用confirmShutdown,确认任务队列中的任务是否都已经被执行了。
3、然后还得再次确认下任务队列中是否已被执行完毕,因为在关闭的过程中外部也是能添加任务的。
4、最终执行清理工作,更改状态为已关闭,释放信号量。
5、如果这个时候还是有任务没执行完,那也只能是无奈了,记个log吧
6、更新整个关闭过程为success
- confirmShutdown代码
protected boolean confirmShutdown() { // 如果state状态 state < ST_SHUTTING_DOWN则直接return false if (!isShuttingDown()) { return false; } // 这个方法必须从内部调用,从修饰符 protected也可以看出 if (!inEventLoop()) { throw new IllegalStateException("must be invoked from an event loop"); } // 取消所有的定时任务 cancelDelayedTasks(); if (gracefulShutdownStartTime == 0) { // 标记shutdown处理的开始时间 gracefulShutdownStartTime = ScheduledFutureTask.nanoTime(); } // 运行tasksQueue或者shutdownHooks中的所有Runnable都处理完成 if (runAllTasks() || runShutdownHooks()) { //分析了下源码,isShutdown()这个只能是在外部线程调用了shutdown()接口的时候才会有可能成为true //但是现在这个方法已经@Deprecated,所以这个if块是不会进入的 if (isShutdown()) { // shutdown 成功,没有更多的runnable需要执行 return true; } // There were tasks in the queue. Wait a little bit more until no tasks are queued for the quiet period. wakeup(true); return false; } final long nanoTime = ScheduledFutureTask.nanoTime(); // runAllTasks() 或者runAllTasks() + runShutdownHooks()方法执行时间操作了最大限制 if (isShutdown() || nanoTime - gracefulShutdownStartTime > gracefulShutdownTimeout) { return true; } // 现在时间与上个任务执行完成的时间差小于quietPeriod时间,继续检测 if (nanoTime - lastExecutionTime <= gracefulShutdownQuietPeriod) { // Check if any tasks were added to the queue every 100ms. // TODO: Change the behavior of takeTask() so that it returns on timeout. wakeup(true); try { //内部线程sleep 100ms Thread.sleep(100); } catch (InterruptedException e) { // Ignore } return false; } // No tasks were added for last quiet period - hopefully safe to shut down. // (Hopefully because we really cannot make a guarantee that there will be no execute() calls by a user.) return true; }
这个WAKEUP_TASK什么也不做,为啥取名wakeup呢
private final Semaphore threadLock = new Semaphore(0); //threadLock的内部permits设置为0,也就是说acquire()永远获取不到permit,会一直被阻塞着。 //那有什么用呢?另一种实现wait()/notify()。
为什么需要AtomicInteger原子操作类
- AtomicInteger 原子性类,对于Java中的运算操作,例如:自增或自减,若没有进行额外的同步操作,在多线程环境下就是线程不安全的。num++解析为num=num+1,明显,这个操作不具备原子性,多线程并发共享这个变量时必然会出现问题
- num ++ 的原子性问题,num++的操作实际上分三个步骤"读-改-写"
int num = 10;
num =num++;//10
- 临时变量读-改-写
int temp = num;
num = num +1;
num = temp;
cas