本文来自网易云社区
作者:乔安然
1. Chain of Responsiblity
定义:
使多个对象都有机会处理请求,从而避免请求的发送者和接受者之间的耦合关系。将这个对象连成一条链,并沿着这条链传递该请求,直到有一个对象处理他为止。
结构实图:
2. Netty ChannelPipeline 分析
Netty的ChannelPipeline和ChannelHandler机制类似于Servlet和Filter过滤器,这类过滤器其实就是责任链模式的一种变形,方便事件的拦截和用户业务逻辑的定制且相互不必耦合在一起。
Netty将Channel的数据管道抽象为ChannelPipeline,消息在ChannelPipline中流动和传递。ChannelPipeline持有IO事件拦截器ChannelHandler的链表,由ChannelHandler对IO事件进行拦截和处理,可以方便的新增和删除ChannelHandler来实现不同的业务逻辑定制,不必对已有的ChannelHandler进行修改,这个开放闭合原则的很好体现。
下面我们对ChannelPipeline和ChannelHandler,以及相关的ChannelHandlerContext进行详细介绍和源码分析。
先看下ChannlePipeline的事件事件处理流程,如下图
- 底层Socket读取bytebuf触发ChannelRead事件(Inbound 事件),由NioEventLoop调用ChannelPipeline的fireChannelRead方法
- 消息被ChannelPipeline中的ChannelHandlerContext传递,依次被各个ChannelHandler处理
- 当有写出的需求(Outbound 事件),调用ChannelHandlerContext write方法,消息再通过ChannelHandlerContext反向传递通过各个ChannelHandler处理。当然各个ChannelHadler可以通过定制只对自己感兴趣的消息进行处理,其余跳过。
下图是ChannelPipeline相关的类UML图
DefaultChannelPipeline:I/O事件承载的数据管道,由ChannelHandlerContext节点组成双链表结构
ChannelHandler: I/O事件的处理层,分别为Inbound和outbound两种事件类型派生ChannelInboundHandler和ChannelOutboundHandler接口,如上图中的MessageToMessageDecoder和MessageToMessageEncoder类分别对消息的解码和编码处理。用户在实际使用中根据需求处理Inbound还是outbound事件。
DefaultChannelHandlerContext:组成pipeline的节点,执行handler的上下文环境,支持异步模式,如下面read事件处理:
@Override public ChannelHandlerContext fireChannelRead(final Object msg) { // 找到下一个inbound的handler invokeChannelRead(findContextInbound(), msg); return this; } static void invokeChannelRead(final AbstractChannelHandlerContext next, Object msg) { final Object m = next.pipeline.touch(ObjectUtil.checkNotNull(msg, "msg"), next); EventExecutor executor = next.executor(); // 判断是否由内部触发 if (executor.inEventLoop()) { next.invokeChannelRead(m); } else { // 外部触发异步处理 executor.execute(new Runnable() { @Override public void run() { next.invokeChannelRead(m); } }); } } // 触发handler中的channelRead方法,对消息进行处理 private void invokeChannelRead(Object msg) { if (invokeHandler()) { try { ((ChannelInboundHandler) handler()).channelRead(this, msg); } catch (Throwable t) { notifyHandlerException(t); } } else { fireChannelRead(msg); } }
3.Mina IoFilterChain分析
责任链模式在mina中也发挥着重要的作用,其中Filter机制就是基于责任链实现的,来看下mina框架组成
从上图看到消息的接受从IoService层先经过Filter层过滤处理后最后交给IoHander,消息的发送则是反过来从IoHander层经过Filter层再到IoService层。由此可以看到netty和mina对消息处理都是相似的。
从图中看到接收消息和发送消息经过Filter层是相反处理的,那么每个Filter就必须知道前一个和后一个Filter,那么mina中的Filter层和netty的pipeline相同都是使用双向链表实现的,那么让我们来看看Filter层具体是如何实现
mina的filterchain包结构:
Filter层的每个filter都是对上图IoFilter接口的实现,我们将具体讲解IoFilter,IoFilterChain,DefaultIoFilterChain这几个类
IoFilterChainBuilder接口和DefaultIoFilterChainBuilder实现不再细讲,从字面意思就是IoFilterChain的建造者
IoFilterEvent是代表filter事件,IoFilterLifeCycleException是指加入链表异常
下面的图是我们要重点讲解的几个类的关系
IoFilter接口:NextFilter接口是其内部接口
IoFilterAdapter类:对IoFilter接口的实现,是所有Filter的基类
IoFilterChain接口:Entry接口是其内部接口
DefaultIoFilterChain类:是对IoFilterChain接口的实现,有EntryImpl,HeadFilter,TailFilter三个内部类,其中EntryImpl类中又有NextFilter接口的内部实现
还需要说明下:IoFilter还有相关接口就写了两个方法,一个接受消息触发的方法还有一个是发送消息触发的方法,剩下的都是这两类消息处理方法就不表示了,这和netty中的inbound、outbound相同
HeadFilter类只对发送消息处理方法重载,TailFilter类只对接受消息处理方法重载
从上图看到EntryImp类是重点,我们就来看看EntryImpl类的实现
private class EntryImpl implements Entry { private EntryImpl prevEntry ; private EntryImpl nextEntry ; private final String name; private IoFilter filter ; private final NextFilter nextFilter; private EntryImpl(EntryImpl prevEntry, EntryImpl nextEntry, String name, IoFilter filter) { if (filter == null) { throw new IllegalArgumentException("filter"); } if (name == null) { throw new IllegalArgumentException("name"); } this.prevEntry = prevEntry; this.nextEntry = nextEntry; this.name = name; // 业务的fliter处理层 this.filter = filter; // 调度filter对读入和写出消息处理 this.nextFilter = new NextFilter() { // 读入消息调用nextEntry处理 public void sessionOpened(IoSession session) { Entry nextEntry = EntryImpl. this.nextEntry ; callNextSessionOpened(nextEntry, session); } // 写出消息调用preEntry反向处理 public void filterWrite(IoSession session, WriteRequest writeRequest) { Entry nextEntry = EntryImpl. this.prevEntry ; callPreviousFilterWrite(nextEntry, session, writeRequest); } }; }
从EntryImpl类的构造方法看到,EntryImpl中保持对上一个节点和下一个节点引用,双向链表结构,name即过滤层名称,filter即过滤层的具体实现,而nextFilter是在构造方法中的内部实现。
下面我们来看看sessionOpen处理的完整过程,sessionOpen事件属于读入对应netty中的inbound事件类型。首先是IoFilterChain收到这个消息触发fireSessionOpened方法
public void fireSessionOpened() { Entry head = this.head ; callNextSessionOpened(head, session); } private void callNextSessionOpened(Entry entry, IoSession session) { try { IoFilter filter = entry.getFilter(); NextFilter nextFilter = entry.getNextFilter(); filter.sessionOpened(nextFilter, session); } catch (Throwable e) { fireExceptionCaught(e); } }
fireSessionOpened方法获取当前的头节点,然后调用callNextSessionOpened方法,而callNextSessionOpened方法是从entry中获取filter和nextfitler,触发filter的sessionOpened方法,同时将nextfilter作为参数传进去,而filter层如果对这个消息感兴趣可以处理完成后调用nextfilter的sessionOpened方法,不感兴趣的话,可能消息到此就结束了。
由此可看出mina中的Fliter和netty的ChannelHandler功能相同,而NextFilter其实是起到中转和调度的作用,收到Reveceive消息转交给后一节点,收到Send消息转交给前一个消息。这和netty中ChannelHandlerContext功能相似。
mina和netty不相同的一点对异步多线程的使用,netty中ChannelHandlerContext中加入对异步支持,而mina中代之以一个更通用的系统,基于一个过滤器:ExecutorFilter。当Fliter层将消息事件传递到ExecutorFilter中,它包含一个Executor来将消息事件传递给线程池运行处理。
本文来自网易云社区,经作者马宝授权发布
网易云免费体验馆,0成本体验20+款云产品!
更多网易研发、产品、运营经验分享请访问网易云社区。
相关文章:
【推荐】 什么是高防服务器?
原文地址:https://www.cnblogs.com/zyfd/p/9667129.html