Pipeline inbound(netty源码7)

netty源码死磕7

Pipeline 入站流程详解

1. Pipeline的入站流程

在讲解入站处理流程前,先脑补和铺垫一下两个知识点:

(1)如何向Pipeline添加一个Handler节点

(2)Handler的出站和入站的区分方式

1.1. HandlerContext节点的添加

在Pipeline实例创建的同时,Netty为Pipeline创建了一个Head和一个Tail,并且建立好了链接关系。

代码如下:

protected DefaultChannelPipeline(Channel channel) {
    this.channel = ObjectUtil.checkNotNull(channel, "channel");
tail = new TailContext(this); head = new HeadContext(this);
    head.next = tail;
    tail.prev = head;
}

也就是说,在加入业务Handler之前,Pipeline的内部双向链表不是一个空链表。而新加入的Handler,加入的位置是,插入在链表的倒数第二个位置,在Tail的前面。

wpsF713.tmp

加入Handler的代码,在DefaultChannelPipeline类中。

具体的代码如下:

@Override
public final ChannelPipeline addLast(EventExecutorGroup group, String name, ChannelHandler handler) {
    final AbstractChannelHandlerContext newCtx;
    synchronized (this) {

     //检查重复
        checkMultiplicity(handler);
        //创建上下文
        newCtx = newContext(group, filterName(name, handler), handler);
         //加入双向链表
        addLast0(newCtx);
        //…
    }
    callHandlerAdded0(newCtx);
    return this;
}

加入之前,首先进行Handler的重复性检查。非共享类型的Handler,只能被添加一次。如果当前要添加的Handler是非共享的,并且已经添加过,那就抛出异常,否则,标识该handler已经添加。

什么是共享类型,什么是非共享类型呢?先聚焦一下主题,后面会详细解答。

检查完成后,给Handler创建包裹上下文Context,然后将Context加入到双向列表的尾部Tail前面。

代码如下:

private void addLast0(AbstractChannelHandlerContext newCtx) {
    AbstractChannelHandlerContext prev = tail.prev;
    newCtx.prev = prev;
    newCtx.next = tail;
    prev.next = newCtx;
    tail.prev = newCtx;
}

这里主要是通过调整双向链接的指针,完成节点的插入。如果对双向链表不熟悉,可以自己画画指向变化的草图,就明白了。

1.2. Context的出站和入站的类型

对于入站和出站,Pipeline中两种不同类型的Handler处理器,出站Handler和入站Handler。

入站(inBound)事件Handler的基类是 ChannelInboundHandler,出站(outBound)事件Handler的基类是 ChannelOutboundHandler。

处理入站(inBound)事件,最典型的就是处理Channel读就绪事件,还有就是业务处理Handler。处理出站outBound操作,最为典型的处理,是写数据到Channel。

对应于两种Handler处理器的Context 包裹器,更加需要区分入站和出站。对Context的区分方式,又是什么呢?

首先,需要在Context加了一组boolean类型判断属性,判断出站和入站的类型。这组属性就是——inbound、outbound。这组属性,定义在上下文包裹器的基类中——ContextAbstractChannelHandlerContext 定义。它们在构造函数中进行初始化。

1ContextAbstractChannelHandlerContext 的构造器代码如下: 2 3abstract class AbstractChannelHandlerContext extends DefaultAttributeMap implements ChannelHandlerContext 4 5{ 6 7private final boolean inbound; 8private final boolean outbound; 9 10AbstractChannelHandlerContext(DefaultChannelPipeline pipeline, EventExecutor executor, String name, 11 boolean inbound, boolean outbound) { 12 //…. 13 this.pipeline = pipeline; 14 this.executor = executor; 15 this.inbound = inbound; 16 this.outbound = outbound; 17 //… 18} 19 20//… 21 22}

对于通用的默认包裹器,继承了ContextAbstractChannelHandlerContext 基类,并且在自己的构造器中,初始化这两个父类属性的方法,如下:

1final class DefaultChannelHandlerContext extends AbstractChannelHandlerContext { 2 3 //… 4private final ChannelHandler handler; 5 private static boolean isInbound(ChannelHandler handler) { 6 return handler instanceof ChannelInboundHandler; 7} 8 9private static boolean isOutbound(ChannelHandler handler) { 10 return handler instanceof ChannelOutboundHandler; 11 } 12 13DefaultChannelHandlerContext( 14 DefaultChannelPipeline pipeline, EventExecutor executor, String name, ChannelHandler handler) { 15 super(pipeline, executor, name, isInbound(handler), isOutbound(handler)); 16 //…. 17 18this.handler = handler; 19 20} 21 22}

从上面的代码可以看出, 通用的包裹器DefaultChannelHandlerContext ,通过自己的 isInbound ()、 is Out bound () **方法的返回值,**对构造函数参数中的Handler 类型进行判断,来设置分类的boolean类型属性inbound、outbound的值。

再看两个非通用的HandlerContext——head和tail。

在HeadContext,则调用父类构造器的第五个参数(outbound)的值为true,表示Head是一个出站类型的Context。代码如下:

final class HeadContext extends AbstractChannelHandlerContext
        implements ChannelOutboundHandler, ChannelInboundHandler {
    private final Unsafe unsafe;
    HeadContext(DefaultChannelPipeline pipeline) {

//父类构造器
        super(pipeline, null, HEAD_NAME, false, true);
//...

}

}

在TailContext,则调用父类构造器的第四个参数(inbound)的值为true,表示Tail是一个入站类型的Context。代码如下:

final class TailContext extends AbstractChannelHandlerContext implements ChannelInboundHandler {
    TailContext(DefaultChannelPipeline pipeline) {
        super(pipeline, null, TAIL_NAME, true, false);
//...

   }

}

无论是哪种类型的handler,Pipeline没有单独和分开的入站和出站链表,都是统一在一个双向链表中进行管理。

下图中,使用紫色代表入站Context,橙色代表出站Context。

wpsF724.tmp

在上图中,橙色表示出站Context,紫色表示入站Context。

在上图中的流程中,区分一个 ChannelHandlerContext到底是in(入站)还是out(出站) ,使用的是Context的isInbound() 和 isOutbound() 这一组方法。

赘述一下:

Tail是出站执行流程的启动点,但是,它最后一个入站处理器。

Hearder,是入站流程的启动起点,但是,它最后一个出站处理器。

感觉,有点儿饶。容易让人混淆。看完整个的入站流程和出站流程的详细介绍,就清楚了。

1.3. 入站操作的全流程

入站事件前面已经讲过,流向是从Java 底层IO到ChannelHandler。入站事件的类型包括连接建立和断开、读就绪、写就绪等。

基本上,,在处理流程上,大部分的入站事件的处理过程,是一致的。

通用的入站Inbound事件处理过程,大致如下(使用IN_EVT符号代替一个通用事件):

(1)pipeline.fireIN_EVT

(2)AbstractChannelHandlerContext.invokeIN_EVT(head, msg);

(3)context.invokeIN_EVT(msg);

(4)handler.IN_EVT

(5)context.fireIN_EVT(msg);

(6)Connect.findContextInbound()

(7)context.invokeIN_EVT(msg);

上面的流程,如果短时间内看不懂,没有关系。可以先看一个例子,再回来推敲学习这个通用流程。

1.4. 读就绪事件的流程实例

下面以最为常见和最好理解的事件——读就绪的事件为例,将Inbound事件做一个详细的描述。

整个读就绪的入站处理流程图,如下:

inbount 1

1.5. 入站源头的Java底层 NIO封装

入站事件处理的源头,在Channel的底层Java NIO 就绪事件。

Netty对底层Java NIO的操作类,进行了封装,封装成了Unsafe系列的类。比方说,AbstractNioByteChannel 中,就有一个NioByteUnsafe 类,封装了底层的Java NIO的底层Byte字节的读取操作。

为什么叫Unsafe呢?

很简单,就是在外部使用,是不安全的。Unsafe就是只能在Channel内部使用的,在Netty 外部的应用开发中,不建议使用。Unsafe****包装了底层的数据读取工作,包装在Channel中,不需要应用程序关心。应用程序只需要从缓存中,取出缓存数据,完成业务处理即可。

Channel 读取数据到缓存后,下一步就是调用Pipeline的fireChannelRead()方法,从这个点开始,正式开始了Handler的入站处理流程。

从Channel 到Pipeline这一段,Netty的代码如下:

1public abstract class AbstractNioByteChannel extends AbstractNioChannel 2 3{ 4 5 protected class NioByteUnsafe extends AbstractNioUnsafe { 6 7 @Override 8 public final void read() { 9final ChannelPipeline pipeline = pipeline(); 10 …… 11 // 读取结果. 12 13 byteBuf = allocHandle.allocate(allocator); 14 …… 15 int localReadAmount = doReadBytes(byteBuf); 16 ……… 17 18 // 通过pipeline dispatch(分发)结果到Handler 19 20 pipeline.fireChannelRead(byteBuf); 21 …… 22 } 23 24//通过重写newUnsafe() 方法 25 26//取得内部类NioSocketChannelUnsafe的实例 27@Override 28protected AbstractNioUnsafe newUnsafe() { 29 return new NioSocketChannelUnsafe(); 30} 31 3233 34}

channel调用了 pipeline.fireChannelRead(byteBuf)后,进入pipeline 开始处理。这是流程的真正启动的动作。

1.6. Head是入站流程的起点

前面分析到,Pipeline中,入站事件处理流程的处理到的第一个Context是Head。

这一点,从DefaultChannelPipeline 源码可以得到验证,如下所示:

1public class DefaultChannelPipeline implements ChannelPipeline 2 3{ 4 56 7@Override 8public final ChannelPipeline fireChannelRead(Object msg) { 9AbstractChannelHandlerContext.invokeChannelRead(head, msg); 10 return this; 11} 12 1314 15}

Pipeline将内部链表的head头作为参数,传入了invokeChannelRead的静态方法中。

就像开动了流水线的开关,开启了整个的流水线的循环处理。

1.7. 小迭代的五个动作

一个Pipeline上有多个InBound Handler,每一个InBound Handler的处理,可以算做一次迭代,也可以说成小迭代。

每一个迭代,有四个动作。这个invokeIN_EVT方法,是整个四个动作的小迭代的起点。

四个动作,分别如下:

(1)invokeChannelRead(next, msg)

(2)context.invokeIN_EVT(msg);

(3)handler.IN_EVT

(4)context.fireIN_EVT(msg);

(5)Connect.findContextInbound()

局部的流程图如下:

inbount 2

整个五个动作中,只有第三步在Handler中,其他的四步都在Context中完成。

1.8. 流水线小迭代的第一步

invokeChannelRead(next,msg) 静态方法,非常关键,其重要作为是:作为流水线迭代处理的每一轮循环小迭代的第一步。在Context的抽象基类中,源码如下:

1abstract class AbstractChannelHandlerContext extends DefaultAttributeMap implements ChannelHandlerContext 2 3{ 4//... 5static void invokeChannelRead(final AbstractChannelHandlerContext next, final Object msg) { 6 …… 7next.invokeChannelRead(msg); 8 …… 9} 10 11//... 12}

首先,这个是一个静态方法。

其次,这个方法没有啥特别。只是做了一个二转。将处理传递给context实例,调用context实例的invokeChannelRead方法。强调一下,使用了同一个名称哈。但是后边的invokeChannelRead,是一个实例方法,而且只有一个参数。

1.9. context.invokeIN_EVT实例方法

流水线小迭代第二步,触发当前的Context实例的IN_EVT操作。

对于IN_EVT为ChannelRead的时候,第二步方法为invokeChannelRead,其源码如下:

abstract class AbstractChannelHandlerContext extends DefaultAttributeMap implements ChannelHandlerContext

{

private void invokeChannelRead(Object msg) {
    ……
       ((ChannelInboundHandler) handler()).channelRead(this, msg);
……
}

}

这一步很简单,就是将context和msg(byteBuf)作为参数,传递给Handler实例,完成业务处理。

在Handler中,可以获取到以上两个参数实例,作为业务处理的输入。在业务Handler中的IN_EVT方法中,可以写自己的业务处理逻辑。

1.10. 默认的handler.IN_EVT 入站处理操作

流水线小迭代第三步,完后Context实例中Handler的IN_EVT业务操作。

如果Handler中的IN_EVT方法中没有写业务逻辑,则Netty提供了默认的实现。默认源码在ChannelInboundHandlerAdapter 适配器类中。

当IN_EVT为ChannelRead的时候,第三步的默认实现源码如下:

public class ChannelInboundHandlerAdapter extends ChannelHandlerAdapter implements ChannelInboundHandler

{

//默认的通道读操作
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
    ctx.fireChannelRead(msg);
}

//...

}

读完源码发现,这份默认源码,都没有做什么实际的处理。

唯一的干的活,就是调用ctx.fireChannelRead(msg),将msg通过context再一次发射出去。

进入第四步。

1.11. context.fireIN_EVT再发射消息

流水线小迭代第四步,寻找下家,触发下一家的入站处理。

整个是流水线的流动的关键一步,实现了向下一个HandlerContext的流动。

源码如下:

abstract class AbstractChannelHandlerContext extends DefaultAttributeMap implements ChannelHandlerContext

{

private final boolean inbound;
private final boolean outbound;

//...

@Override
public ChannelHandlerContext fireChannelRead(final Object msg) {
    invokeChannelRead(findContextInbound(), msg);
    return this;
}

//..

}

第四步还是在ChannelInboundHandlerAdapter 适配器中定义。首先通过第五步,找到下一个Context,然后回到小迭代的第一步,完成了小迭代的一个闭环。

这一步,对于业务Handler而言,很重要。

在用户Handler中,如果当前 Handler 需要将此事件继续传播下去,则调用contxt.fireIN_EVT方法。如果不这样做, 那么此事件的流水线传播会提前终止。

1.12. findContextInbound()找下家

第五步是查找下家。

代码如下:

public class ChannelInboundHandlerAdapter extends ChannelHandlerAdapter implements ChannelInboundHandler

{

//...

private AbstractChannelHandlerContext findContextInbound() {
    AbstractChannelHandlerContext ctx = this;
    do {
        ctx = ctx.next;
    } while (!ctx.inbound);
    return ctx;
}

}

这个是一个标准的链表查询操作。this表示当前的context,this.next表示下一个context。通过while循环,一直往流水线的下边找,知道查找到下一个入站Context为止。

假定流水下如下图所示:

wpsF775.tmp

在上图中,如果当前context是head,则下一个是Decoder;如果当前context是Decoder,则下一个是Business;如果当前context是Business,则下一个是Tail。

第五步,是在第四步调用的。

找到之后,第四步通过 invokeChannelRead(findContextInbound(), msg)这个静态方法的调用,由回到小迭代的第一步,开始下一轮小的运行。

1.13. 最后一轮Context处理

我们在前面讲到,在Netty中,Tail是最后一个IN boundContext。

final class TailContext extends AbstractChannelHandlerContext implements ChannelInboundHandler {

@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
    onUnhandledInboundMessage(msg);
}

protected void onUnhandledInboundMessage(Object msg) {
  //…

        //释放msg的引用计数
        ReferenceCountUtil.release(msg);
  //..
}

}

在最后的一轮入站处理中。Tail没有做任何的业务逻辑,仅仅是对msg 释放一次引用计数。

这个msg ,是从channel 入站源头的过来的byteBuf。有可能是引用计数类型(ReferenceCounted)类型的缓存,则需要释放其引用。如果不是ReferenceCounted,则什么也不做。

关于缓存的引用计数,后续再开文章做专题介绍。

1.14. 小结

对入站(Inbound )事件的处理流程,做一下小节:

Inbound 事件是通知事件,当某件事情已经就绪后,从Java IO 通知上层Netty  Channel。

Inbound 事件源头是 Channel内部的UNSafe;

Inbound 事件启动者是 Channel,通过Pipeline. fire****IN_EVT启动。

Inbound 事件在 Pipeline 中传输方向是从 head 到 tail。

Inbound 事件最后一个的处理者是 TailContext, 并且其处理方法是空实现。如果没有其他的处理者,则对Inbound ,TailContext是唯一的处理者。

Inbound 事件的向后传递方法是contxt.fireIN_EVT方法。在用户Handler中,如果当前 Handler 需要将此事件继续传播下去,则调用contxt.fireIN_EVT方法。如果不这样做, 那么此事件的流水线传播会提前终止。

无编程不创客,无案例不学习。疯狂创客圈,一大波高手正在交流、学习中!

疯狂创客圈 Java 死磕系列: 【CSDN 总入口】 【博客园 总入口

**JAVA NIO  死磕系列:**NIO简介、NIO  Buffer、 NIO channel、 NIO Selector

reactor 模式 死磕系列: Reactor模式

Netty 源码 死磕系列: 环境搭建 、  EventLoop、 ChannelHandler 、 Pipeline模式、Pipeline inbound、 Pipeline outbound

**Java 类加载器 死磕系列:**双亲委托、 文件系统类加载器、 网络类加载器、 加密类加载器、 AOP 类加载器

 **面试题死磕系列:**Java面试必知必会200题 |   阿里、腾讯、百度、华为、京东、搜狗和滴滴最新面试题汇集

免费资源: 数百G免费视频资源,请参见共享《疯狂创客圈》QQ群文件

点赞
收藏

评论区

加载中...

相关推荐

MySQL:[Err] 1292 - Incorrect datetime value: ‘0000-00-00 00:00:00‘ for column ‘CREATE_TIME‘ at row 1

文章目录问题用navicat导入数据时,报错:原因这是因为当前的MySQL不支持datetime为0的情况。解决修改sql\mode:sql\mode:SQLMode定义了MySQL应支持的SQL语法、数据校验等,这样可以更容易地在不同的环境中使用MySQL。全局s

Oracle 分组与拼接字符串同时使用

SELECTT.,ROWNUMIDFROM(SELECTT.EMPLID,T.NAME,T.BU,T.REALDEPART,T.FORMATDATE,SUM(T.S0)S0,MAX(UPDATETIME)CREATETIME,LISTAGG(TOCHAR(

MySQL部分从库上面因为大量的临时表tmp_table造成慢查询

背景描述Time:20190124T00:08:14.70572408:00User@Host:@Id:Schema:sentrymetaLast_errno:0Killed:0Query_time:0.315758Lock_

皕杰报表之UUID

​在我们用皕杰报表工具设计填报报表时,如何在新增行里自动增加id呢?能新增整数排序id吗?目前可以在新增行里自动增加id,但只能用uuid函数增加UUID编码,不能新增整数排序id。uuid函数说明:获取一个UUID,可以在填报表中用来创建数据ID语法:uuid()或uuid(sep)参数说明:sep布尔值,生成的uuid中是否包含分隔符'',缺省为

手写Java HashMap源码

HashMap的使用教程HashMap的使用教程HashMap的使用教程HashMap的使用教程HashMap的使用教程22

2020年前端实用代码段,为你的工作保驾护航

有空的时候,自己总结了几个代码段,在开发中也经常使用,谢谢。1、使用解构获取json数据let jsonData  id: 1,status: "OK",data: 'a', 'b';let  id, status, data: number   jsonData;console.log(id, status, number )

Pipeline inbound(netty源码7) - HelloWorld