Netty3架构解析-阿里云开发者社区

开发者社区> 雪地脚印> 正文

Netty3架构解析

简介:
+关注继续查看

前记

很早以前就有读Netty源码的打算了,然而第一次尝试的时候从Netty4开始,一直抓不到核心的框架流程,后来因为其他事情忙着就放下了。这次趁着休假重新捡起这个硬骨头,因为Netty3现在还在被很多项目使用,因而这次决定先从Netty3入手,瞬间发现Netty3的代码比Netty4中规中矩的多,很多概念在代码本身中都有清晰的表达,所以半天就把整个框架的骨架搞清楚了。再读Netty4对Netty3的改进总结,回去读Netty4的源码,反而觉得轻松了,一种豁然开朗的感觉。

记得去年读Jetty源码的时候,因为代码太庞大,并且自己的HTTP Server的了解太少,因而只能自底向上的一个一个模块的叠加,直到最后把所以的模块连接在一起而看清它的真正核心骨架。现在读源码,开始习惯先把骨架理清,然后延伸到不同的器官、血肉而看清整个人体。

本文从Reactor模式在Netty3中的应用,引出Netty3的整体架构以及控制流程;然而除了Reactor模式,Netty3还在ChannelPipeline中使用了Intercepting Filter模式,这个模式也在Servlet的Filter中成功使用,因而本文还会从Intercepting Filter模式出发详细介绍ChannelPipeline的设计理念。本文假设读者已经对Netty有一定的了解,因而不会包含过多入门介绍,以及帮Netty做宣传的文字。

Netty3中的Reactor模式

Reactor模式在Netty中应用非常成功,因而它也是在Netty中受大肆宣传的模式,关于Reactor模式可以详细参考本人的另一篇文章《Reactor模式详解》,对Reactor模式的实现是Netty3的基本骨架,因而本小节会详细介绍Reactor模式如何应用Netty3中。

如果读《Reactor模式详解》,我们知道Reactor模式由Handle、Synchronous Event Demultiplexer、Initiation Dispatcher、Event Handler、Concrete Event Handler构成,在Java的实现版本中,Channel对应Handle,Selector对应Synchronous Event Demultiplexer,并且Netty3还使用了两层Reactor:Main Reactor用于处理Client的连接请求,Sub Reactor用于处理和Client连接后的读写请求(关于这个概念还可以参考Doug Lea的这篇PPT:Scalable IO In Java)。所以我们先要解决Netty3中使用什么类实现所有的上述模块并把他们联系在一起的,以NIO实现方式为例:

模式是一种抽象,但是在实现中,经常会因为语言特性、框架和性能需要而做一些改变,因而Netty3对Reactor模式的实现有一套自己的设计:
1. ChannelEvent:Reactor是基于事件编程的,因而在Netty3中使用ChannelEvent抽象的表达Netty3内部可以产生的各种事件,所有这些事件对象在Channels帮助类中产生,并且由它将事件推入到ChannelPipeline中,ChannelPipeline构建ChannelHandler管道,ChannelEvent流经这个管道实现所有的业务逻辑处理。ChannelEvent对应的事件有:ChannelStateEvent表示Channel状态的变化事件,而如果当前Channel存在Parent Channel,则该事件还会传递到Parent Channel的ChannelPipeline中,如OPEN、BOUND、CONNECTED、INTEREST_OPS等,该事件可以在各种不同实现的Channel、ChannelSink中产生;MessageEvent表示从Socket中读取数据完成、需要向Socket写数据或ChannelHandler对当前Message解析(如Decoder、Encoder)后触发的事件,它由NioWorker、需要对Message做进一步处理的ChannelHandler产生;WriteCompletionEvent表示写完成而触发的事件,它由NioWorker产生;ExceptionEvent表示在处理过程中出现的Exception,它可以发生在各个构件中,如Channel、ChannelSink、NioWorker、ChannelHandler中;IdleStateEvent由IdleStateHandler触发,这也是一个ChannelEvent可以无缝扩展的例子。注:在Netty4后,已经没有ChannelEvent类,所有不同事件都用对应方法表达,这也意味这ChannelEvent不可扩展,Netty4采用在ChannelInboundHandler中加入userEventTriggered()方法来实现这种扩展,具体可以参考这里
2. ChannelHandler:在Netty3中,ChannelHandler用于表示Reactor模式中的EventHandler。ChannelHandler只是一个标记接口,它有两个子接口:ChannelDownstreamHandler和ChannelUpstreamHandler,其中ChannelDownstreamHandler表示从用户应用程序流向Netty3内部直到向Socket写数据的管道,在Netty4中改名为ChannelOutboundHandler;ChannelUpstreamHandler表示数据从Socket进入Netty3内部向用户应用程序做数据处理的管道,在Netty4中改名为ChannelInboundHandler。
3. ChannelPipeline:用于管理ChannelHandler的管道,每个Channel一个ChannelPipeline实例,可以运行过程中动态的向这个管道中添加、删除ChannelHandler(由于实现的限制,在最末端的ChannelHandler向后添加或删除ChannelHandler不一定在当前执行流程中起效,参考这里)。ChannelPipeline内部维护一个ChannelHandler的双向链表,它以Upstream(Inbound)方向为正向,Downstream(Outbound)方向为方向。ChannelPipeline采用Intercepting Filter模式实现,具体可以参考这里,这个模式的实现在后一节中还是详细介绍。
4. NioSelector:Netty3使用NioSelector来存放Selector(Synchronous Event Demultiplexer),每个新产生的NIO Channel都向这个Selector注册自己以让这个Selector监听这个NIO Channel中发生的事件,当事件发生时,调用帮助类Channels中的方法生成ChannelEvent实例,将该事件发送到这个Netty Channel对应的ChannelPipeline中,而交给各级ChannelHandler处理。其中在向Selector注册NIO Channel时,Netty Channel实例以Attachment的形式传入,该Netty Channel在其内部的NIO Channel事件发生时,会以Attachment的形式存在于SelectionKey中,因而每个事件可以直接从这个Attachment中获取相关链的Netty Channel,并从Netty Channel中获取与之相关联的ChannelPipeline,这个实现和Doug Lea的Scalable IO In Java一模一样。另外Netty3还采用了Scalable IO In Java中相同的Main Reactor和Sub Reactor设计,其中NioSelector的两个实现:Boss即为Main Reactor,NioWorker为Sub Reactor。Boss用来处理新连接加入的事件,NioWorker用来处理各个连接对Socket的读写事件,其中Boss通过NioWorkerPool获取NioWorker实例,Netty3模式使用RoundRobin方式放回NioWorker实例。更形象一点的,可以通过Scalable IO In Java的这张图表达:

若与Ractor模式对应,NioSelector中包含了Synchronous Event Demultiplexer,而ChannelPipeline中管理着所有EventHandler,因而NioSelector和ChannelPipeline共同构成了Initiation Dispatcher。
5. ChannelSink:在ChannelHandler处理完成所有逻辑需要向客户端写响应数据时,一般会调用Netty Channel中的write方法,然而在这个write方法实现中,它不是直接向其内部的Socket写数据,而是交给Channels帮助类,内部创建DownstreamMessageEvent,反向从ChannelPipeline的管道中流过去,直到第一个ChannelHandler处理完毕,最后交给ChannelSink处理,以避免阻塞写而影响程序的吞吐量。ChannelSink将这个MessageEvent提交给Netty Channel中的writeBufferQueue,最后NioWorker会等到这个NIO Channel已经可以处理写事件时无阻塞的向这个NIO Channel写数据。这就是上图的send是从SubReactor直接出发的原因。
6. Channel:Netty有自己的Channel抽象,它是一个资源的容器,包含了所有一个连接涉及到的所有资源的饮用,如封装NIO Channel、ChannelPipeline、Boss、NioWorkerPool等。另外它还提供了向内部NIO Channel写响应数据的接口write、连接/绑定到某个地址的connect/bind接口等,个人感觉虽然对Channel本身来说,因为它封装了NIO Channel,因而这些接口定义在这里是合理的,但是如果考虑到Netty的架构,它的Channel只是一个资源容器,有这个Channel实例就可以得到和它相关的基本所有资源,因而这种write、connect、bind动作不应该再由它负责,而是应该由其他类来负责,比如在Netty4中就在ChannelHandlerContext添加了write方法,虽然netty4并没有删除Channel中的write接口。

Netty3中的Intercepting Filter模式

如果说Reactor模式是Netty3的骨架,那么Intercepting Filter模式则是Netty的中枢。Reactor模式主要应用在Netty3的内部实现,它是Netty3具有良好性能的基础,而Intercepting Filter模式则是ChannelHandler组合实现一个应用程序逻辑的基础,只有很好的理解了这个模式才能使用好Netty,甚至能得心应手。

关于Intercepting Filter模式的详细介绍可以参考这里,本节主要介绍Netty3中对Intercepting Filter模式的实现,其实就是DefaultChannelPipeline对Intercepting Filter模式的实现。在上文有提到Netty3的ChannelPipeline是ChannelHandler的容器,用于存储与管理ChannelHandler,同时它在Netty3中也起到桥梁的作用,即它是连接Netty3内部到所有ChannelHandler的桥梁。作为ChannelPipeline的实现者DefaultChannelPipeline,它使用一个ChannelHandler的双向链表来存储,以DefaultChannelPipelineContext作为节点:
public interface ChannelHandlerContext {
    Channel getChannel();

    ChannelPipeline getPipeline();

    String getName();

    ChannelHandler getHandler();

    
boolean canHandleUpstream();
    
boolean canHandleDownstream();
    
void sendUpstream(ChannelEvent e);
    
void sendDownstream(ChannelEvent e);
    Object getAttachment();

    
void setAttachment(Object attachment);
}

private final class DefaultChannelHandlerContext implements ChannelHandlerContext {
    
volatile DefaultChannelHandlerContext next;
    
volatile DefaultChannelHandlerContext prev;
    
private final String name;
    
private final ChannelHandler handler;
    
private final boolean canHandleUpstream;
    
private final boolean canHandleDownstream;
    
private volatile Object attachment;
.....
}
在DefaultChannelPipeline中,它存储了和当前ChannelPipeline相关联的Channel、ChannelSink以及ChannelHandler链表的head、tail,所有ChannelEvent通过sendUpstream、sendDownstream为入口流经整个链表:
public class DefaultChannelPipeline implements ChannelPipeline {
    
private volatile Channel channel;
    
private volatile ChannelSink sink;
    
private volatile DefaultChannelHandlerContext head;
    
private volatile DefaultChannelHandlerContext tail;
......
    
public void sendUpstream(ChannelEvent e) {
        DefaultChannelHandlerContext head 
= getActualUpstreamContext(this.head);
        
if (head == null) {
            
return;
        }
        sendUpstream(head, e);
    }

    
void sendUpstream(DefaultChannelHandlerContext ctx, ChannelEvent e) {
        
try {
            ((ChannelUpstreamHandler) ctx.getHandler()).handleUpstream(ctx, e);
        } 
catch (Throwable t) {
            notifyHandlerException(e, t);
        }
    }

    
public void sendDownstream(ChannelEvent e) {
        DefaultChannelHandlerContext tail 
= getActualDownstreamContext(this.tail);
        
if (tail == null) {
            
try {
                getSink().eventSunk(
this, e);
                
return;
            } 
catch (Throwable t) {
                notifyHandlerException(e, t);
                
return;
            }
        }
        sendDownstream(tail, e);
    }

    
void sendDownstream(DefaultChannelHandlerContext ctx, ChannelEvent e) {
        
if (e instanceof UpstreamMessageEvent) {
            
throw new IllegalArgumentException("cannot send an upstream event to downstream");
        }
        
try {
            ((ChannelDownstreamHandler) ctx.getHandler()).handleDownstream(ctx, e);
        } 
catch (Throwable t) {
            e.getFuture().setFailure(t);
            notifyHandlerException(e, t);
        }
    }
对Upstream事件,向后找到所有实现了ChannelUpstreamHandler接口的ChannelHandler组成链(getActualUpstreamContext()),而对Downstream事件,向前找到所有实现了ChannelDownstreamHandler接口的ChannelHandler组成链(getActualDownstreamContext()):
    private DefaultChannelHandlerContext getActualUpstreamContext(DefaultChannelHandlerContext ctx) {
        
if (ctx == null) {
            
return null;
        }
        DefaultChannelHandlerContext realCtx 
= ctx;
        
while (!realCtx.canHandleUpstream()) {
            realCtx 
= realCtx.next;
            
if (realCtx == null) {
                
return null;
            }
        }
        
return realCtx;
    }
    
private DefaultChannelHandlerContext getActualDownstreamContext(DefaultChannelHandlerContext ctx) {
        
if (ctx == null) {
            
return null;
        }
        DefaultChannelHandlerContext realCtx 
= ctx;
        
while (!realCtx.canHandleDownstream()) {
            realCtx 
= realCtx.prev;
            
if (realCtx == null) {
                
return null;
            }
        }
        
return realCtx;
    }
在实际实现ChannelUpstreamHandler或ChannelDownstreamHandler时,调用 ChannelHandlerContext中的sendUpstream或sendDownstream方法将控制流程交给下一个 ChannelUpstreamHandler或下一个ChannelDownstreamHandler,或调用Channel中的write方法发送 响应消息。
public class MyChannelUpstreamHandler implements ChannelUpstreamHandler {
    
public void handleUpstream(ChannelHandlerContext ctx, ChannelEvent e) throws Exception {
        
// handle current logic, use Channel to write response if needed.
        
// ctx.getChannel().write(message);
        ctx.sendUpstream(e);
    }
}

public class MyChannelDownstreamHandler implements ChannelDownstreamHandler {
    
public void handleDownstream(
            ChannelHandlerContext ctx, ChannelEvent e) 
throws Exception {
        
// handle current logic
        ctx.sendDownstream(e);
    }
}
当ChannelHandler向ChannelPipelineContext发送事件时,其内部从当前ChannelPipelineContext节点出发找到下一个ChannelUpstreamHandler或ChannelDownstreamHandler实例,并向其发送ChannelEvent,对于Downstream链,如果到达链尾,则将ChannelEvent发送给ChannelSink:
public void sendDownstream(ChannelEvent e) {
    DefaultChannelHandlerContext prev 
= getActualDownstreamContext(this.prev);
    
if (prev == null) {
        
try {
            getSink().eventSunk(DefaultChannelPipeline.
this, e);
        } 
catch (Throwable t) {
            notifyHandlerException(e, t);
        }
    } 
else {
        DefaultChannelPipeline.
this.sendDownstream(prev, e);
    }
}

public void sendUpstream(ChannelEvent e) {
    DefaultChannelHandlerContext next 
= getActualUpstreamContext(this.next);
    
if (next != null) {
        DefaultChannelPipeline.
this.sendUpstream(next, e);
    }
}
正是因为这个实现,如果在一个末尾的ChannelUpstreamHandler中先移除自己,在向末尾添加一个新的ChannelUpstreamHandler,它是无效的,因为它的next已经在调用前就固定设置为null了。

ChannelPipeline作为ChannelHandler的容器,它还提供了各种增、删、改ChannelHandler链表中的方法,而且如果某个ChannelHandler还实现了LifeCycleAwareChannelHandler,则该ChannelHandler在被添加进ChannelPipeline或从中删除时都会得到同志:
public interface LifeCycleAwareChannelHandler extends ChannelHandler {
    
void beforeAdd(ChannelHandlerContext ctx) throws Exception;
    
void afterAdd(ChannelHandlerContext ctx) throws Exception;
    
void beforeRemove(ChannelHandlerContext ctx) throws Exception;
    
void afterRemove(ChannelHandlerContext ctx) throws Exception;
}

public interface ChannelPipeline {
    
void addFirst(String name, ChannelHandler handler);
    
void addLast(String name, ChannelHandler handler);
    
void addBefore(String baseName, String name, ChannelHandler handler);
    
void addAfter(String baseName, String name, ChannelHandler handler);
    
void remove(ChannelHandler handler);
    ChannelHandler remove(String name);

    
<extends ChannelHandler> T remove(Class<T> handlerType);
    ChannelHandler removeFirst();

    ChannelHandler removeLast();

    
void replace(ChannelHandler oldHandler, String newName, ChannelHandler newHandler);
    ChannelHandler replace(String oldName, String newName, ChannelHandler newHandler);

    
<extends ChannelHandler> T replace(Class<T> oldHandlerType, String newName, ChannelHandler newHandler);
    ChannelHandler getFirst();

    ChannelHandler getLast();

    ChannelHandler get(String name);

    
<extends ChannelHandler> T get(Class<T> handlerType);
    ChannelHandlerContext getContext(ChannelHandler handler);

    ChannelHandlerContext getContext(String name);

    ChannelHandlerContext getContext(Class
<? extends ChannelHandler> handlerType);
    
void sendUpstream(ChannelEvent e);
    
void sendDownstream(ChannelEvent e);
    ChannelFuture execute(Runnable task);

    Channel getChannel();

    ChannelSink getSink();

    
void attach(Channel channel, ChannelSink sink);
    
boolean isAttached();
    List
<String> getNames();
    Map
<String, ChannelHandler> toMap();
}

在DefaultChannelPipeline的ChannelHandler链条的处理流程为:


版权声明:本文内容由阿里云实名注册用户自发贡献,版权归原作者所有,阿里云开发者社区不拥有其著作权,亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容,填写侵权投诉表单进行举报,一经查实,本社区将立刻删除涉嫌侵权内容。

相关文章
.net core 源码解析-web app是如何启动并接收处理请求(二) kestrel的启动
上篇讲到.net core web app是如何启动并接受请求的,下面接着探索kestrel server是如何完成此任务的。 1.kestrel server的入口KestrelServer.Start(Microsoft.AspNetCore.Hosting.Server.IHttpApplication) FrameFactory创建的frame实例最终会交给libuv的loop回调接收请求。
793 0
阿里云服务器怎么设置密码?怎么停机?怎么重启服务器?
如果在创建实例时没有设置密码,或者密码丢失,您可以在控制台上重新设置实例的登录密码。本文仅描述如何在 ECS 管理控制台上修改实例登录密码。
10089 0
未能解析目标框架“.NETFramework,Version=v4.0”的 mscorlib 错误的解决办法
VS2010有时候莫名出现下面问题: 未能解析目标框架“.NETFramework,Version=v4.0”的 mscorlib 错误 相关的工程出现这个问题,可能是使用同步盘同步的引起的。
2364 0
CFNetwork 框架详细解析
1. CFNetwork框架详细解析(一) —— 基本概览2. CFNetwork框架详细解析(二) —— CFNetwork编程指导之简介(一)3. CFNetwork框架详细解析(三) —— CFNetwork编程指导之CFNetwork概念(二)4.
1076 0
Argo项目入驻CNCF,一文解析Kubernetes原生工作流
近期CNCF宣布欢迎Argo项目进入 CNCF 孵化器,作为一个新加入的项目,Argo主要关注于Kubernetes原生的工作流,持续部署等方面。 Argo项目是一组Kubernetes原生工具集合,用于运行和管理Kubernetes上的作业和应用程序。
1806 0
.net core 源码解析-web app是如何启动并接收处理请求
最近.net core 1.1也发布了,蹒跚学步的小孩又长高了一些,园子里大家也都非常积极的在学习,闲来无事,扒拔源码,涨涨见识。 先来见识一下web站点是如何启动的,如何接受请求,.net core web app最简单的例子,大约长这样 public static void M...
699 0
阿里云服务器如何登录?阿里云服务器的三种登录方法
购买阿里云ECS云服务器后如何登录?场景不同,阿里云优惠总结大概有三种登录方式: 登录到ECS云服务器控制台 在ECS云服务器控制台用户可以更改密码、更换系.
13892 0
.Net利用Newtonsoft进行解析Json的快捷方法
现在异构系统之间的数据交换多采用Json格式 .Net如何快捷地解析Json和将Object转换成json呢? 1.利用Newtonsoft解析Json字符串 在百度上查找资料,利用Newtonsoft解析Json多是把Json转换成某一特定的Object,这样的不便之处是有事需要特意为了解析该json而写一个class,对一些不规范的的Json,难以用Object来转换的,用他们的方法就不叫难行得通了。
2077 0
+关注
116
文章
0
问答
文章排行榜
最热
最新
相关电子书
更多
《2021云上架构与运维峰会演讲合集》
立即下载
《零基础CSS入门教程》
立即下载
《零基础HTML入门教程》
立即下载