Netty源码分析:ChannelPipeline

2024-05-14 03:48

本文主要是介绍Netty源码分析:ChannelPipeline,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

Netty源码分析:ChannelPipeline

在博文Netty源码分析:服务端启动全过程

我们在知道NioServerSocketChannel这个类的构造函数的调用链如下:

    public NioServerSocketChannel() {this(newSocket(DEFAULT_SELECTOR_PROVIDER));//newSocket的功能为:利用SelectorProvider产生一个SocketChannelImpl对象。}public NioServerSocketChannel(ServerSocketChannel channel) {super(null, channel, SelectionKey.OP_ACCEPT);config = new NioServerSocketChannelConfig(this, javaChannel().socket());} //父类AbstractNioMessageChannel的构造函数protected AbstractNioMessageChannel(Channel parent, SelectableChannel ch, int readInterestOp) {super(parent, ch, readInterestOp);}   //父类 AbstractNioChannel的构造函数protected AbstractNioChannel(Channel parent, SelectableChannel ch, int readInterestOp) {super(parent);this.ch = ch;this.readInterestOp = readInterestOp;//SelectionKey.OP_ACCEPTtry {ch.configureBlocking(false);//设置当前的ServerSocketChannel为非阻塞的} catch (IOException e) {try {ch.close();} catch (IOException e2) {if (logger.isWarnEnabled()) {logger.warn("Failed to close a partially initialized socket.", e2);}}throw new ChannelException("Failed to enter non-blocking mode.", e);}} //父类AbstractChannel的构造函数protected AbstractChannel(Channel parent) {this.parent = parent;unsafe = newUnsafe();pipeline = new DefaultChannelPipeline(this);}

在如上的AbstractChannel构造函数中, 我们看到,使用 DefaultChannelPipeline类的实例初始化了一个 pipeline 属性。

下面首先看下DefaultChannelPipeline的构造函数。

    public DefaultChannelPipeline(AbstractChannel channel) {if (channel == null) {throw new NullPointerException("channel");}this.channel = channel;tail = new TailContext(this);head = new HeadContext(this);head.next = tail;tail.prev = head;}

在如上的构造函数中,首先将与之关联的Channel保存在属性channel中,然后实例化了两个对象:一个是TailContext 实例tail,一个是HeadContext 实例 head。然后将head和tail相互指向,构成了一个双向链表。

HeadContext、TailContext的继承体系结构如下:

从HeadContext和TailContext的继承结构可以看到:这两个类具有Context和Handler的双重属性,可以这么说:head和tail既是一个ChannelHandlerContext也是一个ChannelHandler,原因在于:

1、这两个类均继承的是AbstractChannelHandlerContext这个类

2、均实现了ChannelHandler接口,只是HeadContext实现的是ChannelOutboundHandler,而TailContext实现的是ChannelInboundHandler接口。

而AbstractChannelHandlerContext类有如下两个属性:

    abstract class AbstractChannelHandlerContext extends DefaultAttributeMap implements ChannelHandlerContext {volatile AbstractChannelHandlerContext next;volatile AbstractChannelHandlerContext prev;

因此,可以得到其实在 DefaultChannelPipeline 中, 维护了一个以 AbstractChannelHandlerContext 为节点的双向链表,其中此链表是 以head(HeadContext)作为头,以tail(TailContext)作为尾的双向链表,这个链表是 Netty 实现 Pipeline 机制的关键.

下面看下HeadContext、TailContext这两个类的构造函数

        HeadContext(DefaultChannelPipeline pipeline) {super(pipeline, null, HEAD_NAME, false, true);unsafe = pipeline.channel().unsafe();}TailContext(DefaultChannelPipeline pipeline) {super(pipeline, null, TAIL_NAME, true, false);}
AbstractChannelHandlerContext(DefaultChannelPipeline pipeline, EventExecutorGroup group, String name,boolean inbound, boolean outbound) {if (name == null) {throw new NullPointerException("name");}channel = pipeline.channel;this.pipeline = pipeline;this.name = name;if (group != null) {// Pin one of the child executors once and remember it so that the same child executor// is used to fire events for the same channel.EventExecutor childExecutor = pipeline.childExecutors.get(group);if (childExecutor == null) {childExecutor = group.next();pipeline.childExecutors.put(group, childExecutor);}executor = childExecutor;} else {executor = null;}this.inbound = inbound;this.outbound = outbound;} 

HeadContext、TailContext这两个构造函数都是调用了父类AbstractChannelHandlerContext如上的构造函数,只是参数有所不同。 有两个参数需要额外注意:

1、对于HeadContex,传入的参数:inbound=false,outbound=true;

2、对于TailContext,传入的参数与HeadContext相反:inbound=true,outbound=false;

可以这么来理解:inbound和outbound这两个标志用来区分链表的节点到底是inboundHandler还是outboundHandler。例如:从上面HeadContext和TailContext的继承结构中可以得到HeadContext就是outboundHandler,而TailContext就是InboundHandler

自定义的handler是如何被添加到Pipeline所持有的双向链表中的呢

一般情况,我们在初始化Bootstrap中,经常会看到如下类似的代码,

            b.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class).handler(new SimpleServerHandler()).childHandler(new ChannelInitializer<SocketChannel>() {@Overridepublic void initChannel(SocketChannel ch) throws Exception {}});

其中SimpleServerHandler是我们自定义的Handler,那么这个SimpleServerHandler是如何被添加到上面所介绍的Pipeline的双向链表中去呢?

在博文 Netty源码分析:服务端启动全过程中可以看到在initAndRegister()方法中调用的init(channel)中有如下代码:

    final ChannelFuture initAndRegister() {final Channel channel = channelFactory().newChannel();try {init(channel);}//...}ServerBootstrap.java@Overridevoid init(Channel channel) throws Exception {//...ChannelPipeline p = channel.pipeline();if (handler() != null) {p.addLast(handler());}//...} 

结合前面的分析,我们知道 ChannelPipeline p = channel.pipeline();得到的就是channel所持有的pipeline对象,而handler()方法返回就是通过b.handler(new SimpleServerHandler())设置的我们自定义的 SimpleServerHandler对象。然后将此Handler插入到Pipeline的双向链表中。

下面我们来跟下addLast方法

DefaultChannelPipeline.java

    @Overridepublic ChannelPipeline addLast(ChannelHandler... handlers) {return addLast(null, handlers);}  @Overridepublic ChannelPipeline addLast(EventExecutorGroup executor, ChannelHandler... handlers) {if (handlers == null) {throw new NullPointerException("handlers");}for (ChannelHandler h: handlers) {if (h == null) {break;}addLast(executor, generateName(h), h);}return this;}  @Overridepublic ChannelPipeline addLast(EventExecutorGroup group, final String name, ChannelHandler handler) {synchronized (this) {checkDuplicateName(name);//检查是否有重复的名字AbstractChannelHandlerContext newCtx = new DefaultChannelHandlerContext(this, group, name, handler);addLast0(name, newCtx);}return this;} 

这是addLast的调用链,下面主要分析最后一个addLast这个重载的方法。

基本逻辑为:

1、首先调用 checkDuplicateName方法来检查是否有重复名字的handler,如果有则抛异常。

    private void checkDuplicateName(String name) {if (name2ctx.containsKey(name)) {throw new IllegalArgumentException("Duplicate handler name: " + name);}}

其中:

    private final Map<String, AbstractChannelHandlerContext> name2ctx =new HashMap<String, AbstractChannelHandlerContext>(4);   

在DefaultChannelPipeline中搜索此字段,可以发现在所有添加handler的方法中addFirst0、addLast0等中出现了如下代码语句:

name2ctx.put(name, newCtx);//重点

上面的name2ctx是一个Map,key为handler的名字,而value则是handler本身。

即name2ctx中保存的是Pipeline中存在的所有handler,因此就可以在checkDuplicateName方法中利用 name2ctx.containsKey(name)来判断是否重明。

既然介绍到了这里,就有必要说下handler的name是如何产生的?从上面的第2个addLast重载方法中可以看到是通过调用generateName(h)方法产生的,下面来看下这个方法:

    private String generateName(ChannelHandler handler) {WeakHashMap<Class<?>, String> cache = nameCaches[(int) (Thread.currentThread().getId() % nameCaches.length)];Class<?> handlerType = handler.getClass();String name;synchronized (cache) {name = cache.get(handlerType);if (name == null) {name = generateName0(handlerType);cache.put(handlerType, name);}}synchronized (this) {// It's not very likely for a user to put more than one handler of the same type, but make sure to avoid// any name conflicts.  Note that we don't cache the names generated here.if (name2ctx.containsKey(name)) {String baseName = name.substring(0, name.length() - 1); // Strip the trailing '0'.for (int i = 1;; i ++) {String newName = baseName + i;if (!name2ctx.containsKey(newName)) {name = newName;break;}}}}return name;}private static String generateName0(Class<?> handlerType) {return StringUtil.simpleClassName(handlerType) + "#0";} 

该函数的功能为:如果nameCache中没有该handler类的名字,则调用generatName0方法产生,产生的名字为:类名+“#0”,例如在本例中SimpleServerHandler这个handler所产生的名字为:SimpleServerHandler#0;如果nameCache中有该handler类的名字且name2ctx Map中有包括此名字的handler(添加了多个同类型的Handler到Pipeline中就会出现这种情况),则递增后面的那个数字作为名字直至不重复。

2、将handler包装为Context

为什么要利用如下的语句奖handler包装成Context呢?

 AbstractChannelHandlerContext newCtx = new DefaultChannelHandlerContext(this, group, name, handler);

原因在于:Pipeline中是以AbstractChannelHandlerContext为节点的双向链表。

下面我们来看下DefaultChannelHandlerContext这个类如下的构造函数

    DefaultChannelHandlerContext(DefaultChannelPipeline pipeline, EventExecutorGroup group, String name, ChannelHandler handler) {super(pipeline, group, name, isInbound(handler), isOutbound(handler));if (handler == null) {throw new NullPointerException("handler");}this.handler = handler;}

从DefaultChannelHandlerContext的继承关系图中可以得到该类继承了AbstractChannelHandlerContext,与前面所介绍的HeadContext、TailContext一样。还记不记得如下的一段话:

HeadContext、TailContext这两个构造函数都是调用了父类AbstractChannelHandlerContext如上的构造函数,只是参数有所不同。 有两个参数需要额外注意:

1)、对于HeadContex,传入的参数:inbound=false,outbound=true;

2)、对于TailContext,传入的参数与HeadContext相反:inbound=true,outbound=false;

类似,这里的DefaultChannelHandlerContext构造函数也是调用了父类AbstractChannelHandlerContext的构造函数,只是,inbound、outbound这两个参数是通过函数isInbound(handler)和isOutbound(handler)来得到,当handler实现了ChannelInboundHandler接口,则isInbound方法返回true,当handler实现了ChannelOutboundHandler接口,则isOunbound方法返回true。

    private static boolean isInbound(ChannelHandler handler) {return handler instanceof ChannelInboundHandler;}private static boolean isOutbound(ChannelHandler handler) {return handler instanceof ChannelOutboundHandler;}

就如前面所说:inbound和outbound这两个标志用来区分链表的节点到底是inboundHandler还是outboundHandler。

在我们的例子中SimpleServerHandler实现的是ChannelInboundHandlerAdapter类,该类的继承结构如下,即可得我们自定义的SimpleServerHandler所对应的 DefaultChannelHandlerContext 的 inbound = true, outbound = false 。记住这一点,后面有用

    private static class SimpleServerHandler extends ChannelInboundHandlerAdapter {@Overridepublic void channelActive(ChannelHandlerContext ctx) throws Exception {System.out.println("channelActive");}@Overridepublic void channelRegistered(ChannelHandlerContext ctx) throws Exception {System.out.println("channelRegistered");}@Overridepublic void handlerAdded(ChannelHandlerContext ctx) throws Exception {System.out.println("handlerAdded");}}

3、调用addLast0方法将handler加入到Pipeline的双向链表的末尾

    private void addLast0(final String name, AbstractChannelHandlerContext newCtx) {checkMultiplicity(newCtx);AbstractChannelHandlerContext prev = tail.prev;newCtx.prev = prev;newCtx.next = tail;prev.next = newCtx;tail.prev = newCtx;name2ctx.put(name, newCtx);callHandlerAdded(newCtx);} 

addLast0方法中就是典型的在链表中插入节点的代码。即当调用了 addLast 方法后, 此 handler 就被添加到双向链表中 tail 元素之前的位置了。

注意:上面addLast0方法中的最后一行代码:callHandlerAdded(newCtx);就是第一次使用我们自定义的handler(SimpleServerHandler)的地方,稍后将进行分析。

以上就分析了我们自定义的handler是如何被添加的Pipeline中去的。

那么问题来了,那我们自定义的handler在哪里被使用了呢?

自定义的handler在哪使用了呢

一般服务器端的代码如下所示:

    public final class SimpleServer {public static void main(String[] args) throws Exception {EventLoopGroup bossGroup = new NioEventLoopGroup(1);EventLoopGroup workerGroup = new NioEventLoopGroup();try {ServerBootstrap b = new ServerBootstrap();b.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class).handler(new SimpleServerHandler()).childHandler(new ChannelInitializer<SocketChannel>() {@Overridepublic void initChannel(SocketChannel ch) throws Exception {}});ChannelFuture f = b.bind(8887).sync();f.channel().closeFuture().sync();} finally {bossGroup.shutdownGracefully();workerGroup.shutdownGracefully();}}private static class SimpleServerHandler extends ChannelInboundHandlerAdapter {@Overridepublic void channelActive(ChannelHandlerContext ctx) throws Exception {System.out.println("channelActive");}@Overridepublic void channelRegistered(ChannelHandlerContext ctx) throws Exception {System.out.println("channelRegistered");}@Overridepublic void handlerAdded(ChannelHandlerContext ctx) throws Exception {System.out.println("handlerAdded");}}}

在运行此代码时,在控制台会输出如下的内容:

    handlerAddedchannelRegisteredchannelActive   

既然这三行的内容被打印出来了,就说明我们自定义的SimpleServerHandler的这些方法在某处依次被调用了,是吧,下面就主要看下。

1、SimpleServerHandler 的 handlerAdded 是在哪里被调用了呢

在本博文稍前面一点,我说过如下的一段话:

上面addLast0方法中的最后一行代码:callHandlerAdded(newCtx);就是第一次使用我们自定义的handler(SimpleServerHandler)的地方,稍后将进行分析。

下面将来分析下callHandlerAdded(newCtx)方法

DefaultChannelPipiline.javaprivate void callHandlerAdded(final ChannelHandlerContext ctx) {if (ctx.channel().isRegistered() && !ctx.executor().inEventLoop()) {ctx.executor().execute(new Runnable() {@Overridepublic void run() {callHandlerAdded0(ctx);}});return;}callHandlerAdded0(ctx);}

上面方法主要是调用了callHandlerAdded0(ctx);方法,代码如下:

    private void callHandlerAdded0(final ChannelHandlerContext ctx) {ctx.handler().handlerAdded(ctx);//...}

如前面所介绍我们知道:ctx就是封装了我们自定义的handler(SimpleServerHandler)的DefaultChannelHandlerContext实例,ctx.handler()返回的就是SimpleServeHandler实例,进行调用的此类中如下的 handlerAdded方法,进而在控制台输出:handlerAdded

            @Overridepublic void handlerAdded(ChannelHandlerContext ctx) throws Exception {System.out.println("handlerAdded");}

2、SimpleServerHandler 的 channelRegistered 是在哪里被调用了呢

跟踪b.bind(8887)代码逻辑,发现在如下的initAndRegister()方法的注册阶段调用的register0()方法中出现了如下的代码片段:

    final ChannelFuture initAndRegister() {final Channel channel = channelFactory().newChannel();init(channel); ChannelFuture regFuture = group().register(channel);//...省略了一些此时不关注的代码逻辑}AbstractChannel.javaprivate void register0(ChannelPromise promise) {//...doRegister();registered = true;safeSetSuccess(promise);pipeline.fireChannelRegistered();if (isActive()) {pipeline.fireChannelActive();}//...} 

先看下:pipeline.fireChannelRegistered();

DefaultChannelPipeline.java

    @Overridepublic ChannelPipeline fireChannelRegistered() {head.fireChannelRegistered();return this;} 

这里直接调用了head(HeadContext实例)的fireChannelRegistered()方法,如下:

AbstractChannelHandlerContext.java@Overridepublic ChannelHandlerContext fireChannelRegistered() {final AbstractChannelHandlerContext next = findContextInbound();//分析EventExecutor executor = next.executor();if (executor.inEventLoop()) {next.invokeChannelRegistered();} else {executor.execute(new OneTimeTask() {@Overridepublic void run() {next.invokeChannelRegistered();}});}return this;}private AbstractChannelHandlerContext findContextInbound() {AbstractChannelHandlerContext ctx = this;do {ctx = ctx.next;} while (!ctx.inbound);return ctx;} 

该方法首先是调用如上的findContextInbound()方法从链头head开始寻找第一个inbound=true的节点,看到没,这就是前面特别强调inbound、outbound这两个变量。很明显,这里找到的就是封装我们自定义的SimpleServerHandler的DefaultChannelHandlerContext实例。

下面将看下 next.invokeChannelRegistered();方法

    private void invokeChannelRegistered() {try {((ChannelInboundHandler) handler()).channelRegistered(this);} catch (Throwable t) {notifyHandlerException(t);}}  

显然,封装我们自定义的SimpleServerHandler的DefaultChannelHandlerContext实例对象调用如上方法中的handler()方法的就是我们自定义的 SimpleServerHandler 实例。然后调用此类的如下代码的channelRegistered方法,这样就在控制台输出了: channelRegistered。

            @Overridepublic void channelRegistered(ChannelHandlerContext ctx) throws Exception {System.out.println("channelRegistered");} 

3、SimpleServerHandler 的 channelActive方法 是在哪里被调用了呢

接下来,看下:pipeline.fireChannelActive();要说明的是在register0()方法中isActive()方法返回的false,因此在register0()方法中是不会执行此语句的,至于在哪里执行了该语句,可以参考博文 Netty源码分析:服务端启动全过程

DefaultChannelPipeline.java@Overridepublic ChannelPipeline fireChannelActive() {head.fireChannelActive();if (channel.config().isAutoRead()) {channel.read();}return this;}   @Overridepublic ChannelHandlerContext fireChannelActive() {final AbstractChannelHandlerContext next = findContextInbound();EventExecutor executor = next.executor();if (executor.inEventLoop()) {next.invokeChannelActive();} else {executor.execute(new OneTimeTask() {@Overridepublic void run() {next.invokeChannelActive();}});}return this;}//AbstractChannelHandlerContext    private void invokeChannelActive() {try {((ChannelInboundHandler) handler()).channelActive(this);} catch (Throwable t) {notifyHandlerException(t);}}  

pipeline.fireChannelActive()这个方法与上面分析的fireChannelRegistered()方法类似,也是通过调用head.fireChannelActive();语句来从Pipeline的头节点开始找到第一个Inbound=true的节点:包装了我们自定义handler的DefaultChannelHandlerContext实例。然后调用此实例的invokeChannelActive()方法进而调用了我们自定的SimpleServerHandler的 channelActive方法,进而在控制台输出:channelActive。

        @Overridepublic void channelActive(ChannelHandlerContext ctx) throws Exception {System.out.println("channelActive");} 

小结

本博文从源码的角度分析了ChannelPipeline这个类。了解了我们自定义的handler是被加入到Pipeline所持有的双向链表中的,了解了我们自定义的handler中重载的几种方法在哪里被调用的。

需要记住的几点如下:

1、在 Netty 中每个 Channel 都有且仅有一个 ChannelPipeline 与之对应。

2、ChannelPipeline是一个维护了一个以 AbstractChannelHandlerContext 为节点的双向链表,其中此链表是 以head(HeadContext)作为头,以tail(TailContext)作为尾的双向链表.

如上两点关系用图表示如下:(注:图片截图于参考资料)

在最后,感谢参考资料所列出的博文的作者,写作思路相当清晰,自己看的很爽,然后自己也对照的源码过了这整个过程,理解的更深刻了一些。

参考资料

1、https://segmentfault.com/a/1190000007308934

这篇关于Netty源码分析:ChannelPipeline的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/987664

相关文章

怎样通过分析GC日志来定位Java进程的内存问题

《怎样通过分析GC日志来定位Java进程的内存问题》:本文主要介绍怎样通过分析GC日志来定位Java进程的内存问题,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教... 目录一、GC 日志基础配置1. 启用详细 GC 日志2. 不同收集器的日志格式二、关键指标与分析维度1.

MySQL中的表连接原理分析

《MySQL中的表连接原理分析》:本文主要介绍MySQL中的表连接原理分析,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教... 目录1、背景2、环境3、表连接原理【1】驱动表和被驱动表【2】内连接【3】外连接【4编程】嵌套循环连接【5】join buffer4、总结1、背景

python中Hash使用场景分析

《python中Hash使用场景分析》Python的hash()函数用于获取对象哈希值,常用于字典和集合,不可变类型可哈希,可变类型不可,常见算法包括除法、乘法、平方取中和随机数哈希,各有优缺点,需根... 目录python中的 Hash除法哈希算法乘法哈希算法平方取中法随机数哈希算法小结在Python中,

Java Stream的distinct去重原理分析

《JavaStream的distinct去重原理分析》Javastream中的distinct方法用于去除流中的重复元素,它返回一个包含过滤后唯一元素的新流,该方法会根据元素的hashcode和eq... 目录一、distinct 的基础用法与核心特性二、distinct 的底层实现原理1. 顺序流中的去重

关于MyISAM和InnoDB对比分析

《关于MyISAM和InnoDB对比分析》:本文主要介绍关于MyISAM和InnoDB对比分析,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教... 目录开篇:从交通规则看存储引擎选择理解存储引擎的基本概念技术原理对比1. 事务支持:ACID的守护者2. 锁机制:并发控制的艺

MyBatis Plus 中 update_time 字段自动填充失效的原因分析及解决方案(最新整理)

《MyBatisPlus中update_time字段自动填充失效的原因分析及解决方案(最新整理)》在使用MyBatisPlus时,通常我们会在数据库表中设置create_time和update... 目录前言一、问题现象二、原因分析三、总结:常见原因与解决方法对照表四、推荐写法前言在使用 MyBATis

Python主动抛出异常的各种用法和场景分析

《Python主动抛出异常的各种用法和场景分析》在Python中,我们不仅可以捕获和处理异常,还可以主动抛出异常,也就是以类的方式自定义错误的类型和提示信息,这在编程中非常有用,下面我将详细解释主动抛... 目录一、为什么要主动抛出异常?二、基本语法:raise关键字基本示例三、raise的多种用法1. 抛

github打不开的问题分析及解决

《github打不开的问题分析及解决》:本文主要介绍github打不开的问题分析及解决,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教... 目录一、找到github.com域名解析的ip地址二、找到github.global.ssl.fastly.net网址解析的ip地址三

Mysql的主从同步/复制的原理分析

《Mysql的主从同步/复制的原理分析》:本文主要介绍Mysql的主从同步/复制的原理分析,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教... 目录为什么要主从同步?mysql主从同步架构有哪些?Mysql主从复制的原理/整体流程级联复制架构为什么好?Mysql主从复制注意

java -jar命令运行 jar包时运行外部依赖jar包的场景分析

《java-jar命令运行jar包时运行外部依赖jar包的场景分析》:本文主要介绍java-jar命令运行jar包时运行外部依赖jar包的场景分析,本文给大家介绍的非常详细,对大家的学习或工作... 目录Java -jar命令运行 jar包时如何运行外部依赖jar包场景:解决:方法一、启动参数添加: -Xb