Netty 核心原理二 EventExecutorGroup 与 EventExecutor 原理

EventExecutorGroup 原理

该类继承自ScheduledExecutorService，在提供ScheduledExecutorService接口及其父接口的功能外，增加了如下几个接口，用于执行关闭和获取事件执行器组中的下一个执行器EventExecutor。我们可以从源码中了解到：该接口管理了一组EventExecutor，并提供管理他们生命周期的方法和获取EventExecutor的方法。源码描述如下。

public interface EventExecutorGroup extends ScheduledExecutorService, Iterable<EventExecutor> {

 

 // 用于判断当前EventExecutorGroup管理的所有EventExecutor是否正在关闭

 boolean isShuttingDown();



 // 用于安全关闭执行器组

 Future<?> shutdownGracefully(); // 通过默认值调用下面有参的方法完成关闭

 Future<?> shutdownGracefully(long quietPeriod, long timeout, TimeUnit unit); // 通知执行器组当前需要关闭。一旦该方法被调用，那么isShuttingDown方法将返回true，并且执行器将开始执行关闭动作。

 

 // 获取一个Future对象，该对象将在当前EventExecutorGroup管理的所有EventExecutor是否都已经终止后被设置完成

 Future<?> terminationFuture();

 

 // 用于获取一个当前EventExecutorGroup管理的EventExecutor

 EventExecutor next();



}

EventExecutor 原理

直接通过推理来看看该接口的原理：

1. 我们知道EventExecutorGroup接口定义了一直事件执行器组，那么这个组可以表示1个或者多个
2. 而EventExecutor继承自EventExecutorGroup，从名字可以看出，它只代表了一个执行器
3. 而EventExecutorGroup接口定义我们可以在执行器组中可以只定义一个执行器
4. 所以我们让EventExecutor继承自EventExecutorGroup，表示一个特殊的执行器组：里面只有一个执行器也即它本身
5. 这样的话，next方法永远返回他自己，就没问题了

该接口对EventExecutorGroup扩展，更进一步的定义了事件执行器的功能函数。读者可以想想这种接口设计有何好处：比如我定义一个EventExecutorGroup：EEG-A，然后里面包括一组：EventExecutor，然后：EventExecutor内部又包括一组EventExecutorGroup：EEG-B，同理依次向下包括多个子执行器。源码如下。

public interface EventExecutor extends EventExecutorGroup {



 /**

 * 由于当前事件执行器组只有他自己，那么next永远返回他自己

 */

 @Override

 EventExecutor next();



 /**

 * 返回该事件执行器所属的时间执行器组EventExecutorGroup

 */

 EventExecutorGroup parent();



 /**

 * 通过Thread.currentThread()检测当前线程是否为当前EventExecutor的绑定执行线程

 */

 boolean inEventLoop();



 /**

 * 指定线程对象来判断，同inEventLoop方法

 */

 boolean inEventLoop(Thread thread);



 /**

 * 生成一个新的Promise实例

 */

 <V> Promise<V> newPromise();



 /**

 * 生成一个新的ProgressivePromise实例

 */

 <V> ProgressivePromise<V> newProgressivePromise();



 /**

 * 创建一个新的Future实例，但是通过result参数指定Future的结果，并标记已经成功完成

 */

 <V> Future<V> newSucceededFuture(V result);



 /**

 * 创建一个新的Future实例，但是通过cause参数指定Future的异常结果，并标记已经失败完成

 */

 <V> Future<V> newFailedFuture(Throwable cause);

}

AbstractEventExecutorGroup 原理

该方法提供了EventExecutorGroup的服务方法默认实现。通过源码我们可以看到：该类为了保证一切方法皆异步，所以默认实现均是调用next方法获取到执行器组中下一个执行器，然后调用该执行器的同样方法。源码描述如下。

public abstract class AbstractEventExecutorGroup implements EventExecutorGroup {

 

 // 用于向EventExecutorGroup执行器组提交Runnable任务

 @Override

 public Future<?> submit(Runnable task) {

  return next().submit(task);

 }



 // 用于向EventExecutorGroup执行器组提交Runnable任务并指定结果值

 @Override

 public <T> Future<T> submit(Runnable task, T result) {

  return next().submit(task, result);

 }



 // 用于向EventExecutorGroup执行器组提交Callable任务并指定结果值

 @Override

 public <T> Future<T> submit(Callable<T> task) {

  return next().submit(task);

 }



 // 用于向EventExecutorGroup执行器组提交延迟执行的Runnable任务

 @Override

 public ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit) {

  return next().schedule(command, delay, unit);

 }



 // 用于向EventExecutorGroup执行器组提交延迟执行的Callable任务

 @Override

 public <V> ScheduledFuture<V> schedule(Callable<V> callable, long delay, TimeUnit unit) {

  return next().schedule(callable, delay, unit);

 }



 // 用于向EventExecutorGroup执行器组提交周期性执行的Runnable任务

 @Override

 public ScheduledFuture<?> scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit) {

  return next().scheduleAtFixedRate(command, initialDelay, period, unit);

 }



 // 用于向EventExecutorGroup执行器组提交周期性执行的Runnable任务

 @Override

 public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command, long initialDelay, long delay, TimeUnit unit) {

  return next().scheduleWithFixedDelay(command, initialDelay, delay, unit);

 }



 // 通过指定默认参数来调用shutdownGracefully方法完成优雅关闭。DEFAULT_SHUTDOWN_QUIET_PERIOD = 2、DEFAULT_SHUTDOWN_TIMEOUT = 15

 @Override

 public Future<?> shutdownGracefully() {

  return shutdownGracefully(DEFAULT_SHUTDOWN_QUIET_PERIOD, DEFAULT_SHUTDOWN_TIMEOUT, TimeUnit.SECONDS);

 }



 // 废弃不用，推荐使用shutdownGracefully()

 @Override

 @Deprecated

 public abstract void shutdown();



 // 废弃不用，推荐使用shutdownGracefully()

 @Override

 @Deprecated

 public List<Runnable> shutdownNow() {

  shutdown();

  return Collections.emptyList();

 }



 // 执行所有提交的集合中的任务

 @Override

 public <T> List<java.util.concurrent.Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)

  throws InterruptedException {

  return next().invokeAll(tasks);

 }



 // 执行所有提交的集合中的任务,可以指定任务超时时间

 @Override

 public <T> List<java.util.concurrent.Future<T>> invokeAll(

  Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit) throws InterruptedException {

  return next().invokeAll(tasks, timeout, unit);

 }



 // 执行集合中的任务，直到有一个任务被执行完成

 @Override

 public <T> T invokeAny(Collection<? extends Callable<T>> tasks) throws InterruptedException, ExecutionException {

  return next().invokeAny(tasks);

 }



 // 执行集合中的任务，直到有一个任务被执行完成，可以指定等待时间

 @Override

 public <T> T invokeAny(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit)

  throws InterruptedException, ExecutionException, TimeoutException {

  return next().invokeAny(tasks, timeout, unit);

 }



 // 执行传入的任务对象

 @Override

 public void execute(Runnable command) {

  next().execute(command);

 }

}