netty入门06

本文来源：http://blog.csdn.net/zxhoo/article/details/17709765

感谢博主。本文仅供参考学习。

 

目录(?)[+]

前一篇文章画了一张巨大的类图，但是只重点分析了NioEventLoopGroup的继承层次，这篇文章来看看NioEventLoop的继承层次。


从Executor接口说起

[java] view plain copy 在CODE上查看代码片派生到我的代码片
  1. public interface Executor {  
  2.     void execute(Runnable command);  
  3. }  
Executor接口只有一个方法，用来执行一个命令。但是什么时候执行，在哪个线程里执行，得看具体的实现。


ExecutorService接口

ExecutorService接口扩展了Executor接口，添加了两组方法，一组用来终止Executor的执行：

[java] view plain copy 在CODE上查看代码片派生到我的代码片
  1. void shutdown();  
  2. List<Runnable> shutdownNow();  
  3. boolean isShutdown();  
  4. boolean isTerminated();  
  5. boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException;  
另一组扩展Executor的执行能力，让它可以提交和执行Callable任务：
[java] view plain copy 在CODE上查看代码片派生到我的代码片
  1. <T> Future<T> submit(Callable<T> task);  
  2. <T> Future<T> submit(Runnable task, T result);  
  3. Future<?> submit(Runnable task);  
  4. <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks) throws InterruptedException;  
  5. <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit) throws InterruptedException;  
  6. <T> T invokeAny(Collection<? extends Callable<T>> tasks) throws InterruptedException, ExecutionException;  
  7. <T> T invokeAny(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException;  


ScheduledExecutorService接口

ScheduledExecutorService接口扩展ExecutorService接口，添加了调度能力：

[java] view plain copy 在CODE上查看代码片派生到我的代码片
  1. public ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit);  
  2. public <V> ScheduledFuture<V> schedule(Callable<V> callable, long delay, TimeUnit unit);  
  3. public ScheduledFuture<?> scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit);  
  4. public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command, long initialDelay, long delay, TimeUnit unit);  

AbstractExecutorService抽象类

AbstractExecutorService实现了ExecutorService接口，实现了ExecutorService接口里定义的任务执行方法。


EventExecutorGroup接口

前面几个是java.util.concurrent包里头的类或接口，从EventExecutorGroup开始，进入Netty的世界。

[java] view plain copy 在CODE上查看代码片派生到我的代码片
  1. public interface EventExecutorGroup extends ScheduledExecutorService, Iterable<EventExecutor> {  
  2.     boolean isShuttingDown();  
  3.     Future<?> shutdownGracefully();  
  4.     Future<?> shutdownGracefully(long quietPeriod, long timeout, TimeUnit unit);  
  5.     Future<?> terminationFuture();  
  6.   
  7.     EventExecutor next();  
  8. }  
EventExecutorGroup把从ExecutorService接口里继承来的shutdown()和shutdownNow()方法给废弃（@Deprecated）了，然后提供了shutdownGracefully()方法。shutdownGracefully()方法我没有彻底理解，从Javadoc来看，shutdownGracefully()方法通知EventExecutorGroup准备结束，然后进入一个“安静期（the quiet period）”，如果这个期间内有任务提交进来，这个任务仍然会被接收，然后安静期重新开始。只有安静期顺利度过，EventExecutorGroup才会真正的关闭自己。

EventExecutorGroup还实现了Iterable接口，并其提供了next()方法返回其中的一个EventExecutor。这样来看，EventExecutorGroup这个类名起的真是太恰当了：它逻辑上实际上就是一组EventExecutor：



EventExecutor接口

EventExecutor给EventExecutorGroup添加了下面这些方法：

[java] view plain copy 在CODE上查看代码片派生到我的代码片
  1. public interface EventExecutor extends EventExecutorGroup {  
  2.     EventExecutorGroup parent();  
  3.   
  4.     boolean inEventLoop();  
  5.     boolean inEventLoop(Thread thread);  
  6.   
  7.     <V> Promise<V> newPromise();  
  8.     <V> ProgressivePromise<V> newProgressivePromise();  
  9.     <V> Future<V> newSucceededFuture(V result);  
  10.     <V> Future<V> newFailedFuture(Throwable cause);  
  11. }  
parent()方法比较好理解，返回自己所属的group。EventExecutor需要一个线程来执行任务，inEventLoop()方法用来判断某个线程是不是EventExecutor所使用的线程。但剩下的四个方法不太好理解，暂时先不管了。


EventLoopGroup接口

EventLoopGroup接口比较简单，只定义了两个方法，用来注册Channel：

[java] view plain copy 在CODE上查看代码片派生到我的代码片
  1. public interface EventLoopGroup extends EventExecutorGroup {  
  2.     ChannelFuture register(Channel channel);  
  3.     ChannelFuture register(Channel channel, ChannelPromise promise);  
  4. }  

AbstractEventExecutor

AbstractEventExecutor为EventExecutor接口里的一些方法提供了默认实现，这些默认实现大概可以分为下面几组：

  • 无参数版方法默认实现（调用有参数版）
    • inEventLoop()
    • shutdownGracefully()
  • 转发给AbstractExecutorService
    • submit(Runnable task)
    • submit(Runnable task, T result)
    • submit(Callable<T> task)
  • 直接抛出UnsupportedOperationException
    • schedule(Callable<V> callable, long delay, TimeUnit unit)
    • scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit)
    • scheduleWithFixedDelay(Runnable command, long initialDelay, long delay, TimeUnit unit)
  • 覆盖（override）AbstractExecutorService中的方法实现
    • newTaskFor(Runnable runnable, T value)
    • newTaskFor(Callable<T> callable)

最后next()方法只是返回自己：

[java] view plain copy 在CODE上查看代码片派生到我的代码片
  1. @Override  
  2. public EventExecutor next() {  
  3.     return this;  
  4. }  
下面是类图：


SingleThreadEventExecutor

再往下看是SingleThreadEventExecutor，从名字上看，SingleThreadEventExecutor肯定是在单线程里执行task。SingleThreadEventExecutor类的代码有800多行，但是只要大概看一下就可以了解它是如何工作的：

  • 它内部有两个task队列，taskQueue和delayedTaskQueue，普通任务会进入taskQueue，调度任务进入delayedTaskQueue
  • 默认的taskQueue是一个LinkedBlockingQueue
  • 实现了很多任务执行相关的方法，但核心的run()方法留给子类来实现，子类需要在run()方法里取出并执行task

下面是类图：


SingleThreadEventLoop

SingleThreadEventLoop比较简单，下面是它的全部代码：

[java] view plain copy 在CODE上查看代码片派生到我的代码片
  1. public abstract class SingleThreadEventLoop extends SingleThreadEventExecutor implements EventLoop {  
  2.   
  3.     protected SingleThreadEventLoop(EventLoopGroup parent, ThreadFactory threadFactory, boolean addTaskWakesUp) {  
  4.         super(parent, threadFactory, addTaskWakesUp);  
  5.     }  
  6.   
  7.     @Override  
  8.     public EventLoopGroup parent() {  
  9.         return (EventLoopGroup) super.parent();  
  10.     }  
  11.   
  12.     @Override  
  13.     public EventLoop next() {  
  14.         return (EventLoop) super.next();  
  15.     }  
  16.   
  17.     @Override  
  18.     public ChannelFuture register(Channel channel) {  
  19.         return register(channel, channel.newPromise());  
  20.     }  
  21.   
  22.     @Override  
  23.     public ChannelFuture register(final Channel channel, final ChannelPromise promise) {  
  24.         if (channel == null) {  
  25.             throw new NullPointerException("channel");  
  26.         }  
  27.         if (promise == null) {  
  28.             throw new NullPointerException("promise");  
  29.         }  
  30.   
  31.         channel.unsafe().register(this, promise);  
  32.         return promise;  
  33.     }  
  34. }  

NioEventLoop

最后终于轮到NioEventLoop了，不过遗憾的是，NioEventLoop太复杂了，我还没有完全看明白。下面是完整的继承层次结构图：


NioEventLoop的run()方法

最后我想简单分析一下NioEventLoop的run()方法，下面是run()方法的部分代码：
[java] view plain copy 在CODE上查看代码片派生到我的代码片
  1. @Override  
  2. protected void run() {  
  3.     for (;;) {  
  4.         // process io events  
  5.         ...  
  6.         final long ioTime = System.nanoTime() - ioStartTime;  
  7.         final int ioRatio = this.ioRatio;  
  8.         runAllTasks(ioTime * (100 - ioRatio) / ioRatio);  
  9.         ...  
  10.     }  
  11. }  

也就是说，NioEventLoop在单线程里同时处理IO事件和其他任务，NioEventLoop尽量（但不能保 证）按照给定的比率（默认为50%）来分配花在这两种事情上的时间。换句话说，我们不应该在NioEventLoop里执行耗时的操作（比如数据库操 作），这样会卡死NioEventLoop，降低程序的响应性。
 
顶
1
踩
0
【推广】 免费学中医，健康全家人
原文地址：https://www.cnblogs.com/xdrlczj/p/5846088.html