netty系列之:channelPipeline详解

周旨
• 阅读 1430

简介

我们在介绍channel的时候提到过,几乎channel中所有的实现都是通过channelPipeline进行的,作为一个pipline,它到底是如何工作的呢?

一起来看看吧。

ChannelPipeline

ChannelPipeline是一个interface,它继承了三个接口,分别是ChannelInboundInvoker,ChannelOutboundInvoker和Iterable:

public interface ChannelPipeline
        extends ChannelInboundInvoker, ChannelOutboundInvoker, Iterable<Entry<String, ChannelHandler>> 

继承自ChannelInboundInvoker,表示ChannelPipeline可以触发channel inboud的一些事件,比如:

ChannelInboundInvoker fireChannelRegistered();
ChannelInboundInvoker fireChannelUnregistered();
ChannelInboundInvoker fireChannelActive();
ChannelInboundInvoker fireChannelInactive();
ChannelInboundInvoker fireExceptionCaught(Throwable cause);
ChannelInboundInvoker fireUserEventTriggered(Object event);
ChannelInboundInvoker fireChannelRead(Object msg);
ChannelInboundInvoker fireChannelReadComplete();
ChannelInboundInvoker fireChannelWritabilityChanged();

继承自ChannelOutboundInvoker,表示ChannelPipeline可以进行一些channel的主动操作,如:bind,connect,disconnect,close,deregister,read,write,flush等操作。

继承自Iterable,表示ChannelPipeline是可遍历的,为什么ChannelPipeline是可遍历的呢?

因为ChannelPipeline中可以添加一个或者多个ChannelHandler,ChannelPipeline可以看做是一个ChannelHandler的集合。

比如ChannelPipeline提供了一系列的添加ChannelHandler的方法:

ChannelPipeline addFirst(String name, ChannelHandler handler);
ChannelPipeline addFirst(EventExecutorGroup group, String name, ChannelHandler handler);
ChannelPipeline addFirst(EventExecutorGroup group, ChannelHandler... handlers);
ChannelPipeline addFirst(ChannelHandler... handlers);

ChannelPipeline addLast(String name, ChannelHandler handler);
ChannelPipeline addLast(EventExecutorGroup group, String name, ChannelHandler handler);
ChannelPipeline addLast(ChannelHandler... handlers);
ChannelPipeline addLast(EventExecutorGroup group, ChannelHandler... handlers);

ChannelPipeline addBefore(String baseName, String name, ChannelHandler handler);
ChannelPipeline addBefore(EventExecutorGroup group, String baseName, String name, ChannelHandler handler);
ChannelPipeline addAfter(String baseName, String name, ChannelHandler handler);
ChannelPipeline addAfter(EventExecutorGroup group, String baseName, String name, ChannelHandler handler);

可以从前面添加,也可以从后面添加,或者从特定的位置添加handler。

另外还可以从pipeline中删除特定的channelHandler,或者移出和替换特定位置的handler:

ChannelPipeline remove(ChannelHandler handler);
ChannelHandler remove(String name);
ChannelHandler removeFirst();
ChannelHandler removeLast();
ChannelPipeline replace(ChannelHandler oldHandler, String newName, ChannelHandler newHandler);
ChannelHandler replace(String oldName, String newName, ChannelHandler newHandler);

当然,更少不了对应的查询操作:

ChannelHandler first();
ChannelHandler last();
ChannelHandler get(String name);
List<String> names();

还可以根据传入的ChannelHandler获得handler对应的ChannelHandlerContext。

ChannelHandlerContext context(ChannelHandler handler);

ChannelPipeline中还有一些触发channel相关的事件,如:

    ChannelPipeline fireChannelRegistered();
    ChannelPipeline fireChannelUnregistered();
    ChannelPipeline fireChannelActive();
    ChannelPipeline fireChannelInactive();
    ChannelPipeline fireExceptionCaught(Throwable cause);
    ChannelPipeline fireUserEventTriggered(Object event);
    ChannelPipeline fireChannelRead(Object msg);
    ChannelPipeline fireChannelReadComplete();
    ChannelPipeline fireChannelWritabilityChanged();

事件传递

那么有些朋友可能会问了,既然ChannelPipeline中包含了很多个handler,那么handler中的事件是怎么传递的呢?

其实这些事件是通过调用ChannelHandlerContext中的相应方法来触发的。

对于Inbound事件来说,可以调用下面的方法,进行事件的传递:

ChannelHandlerContext.fireChannelRegistered()
ChannelHandlerContext.fireChannelActive()
ChannelHandlerContext.fireChannelRead(Object)
ChannelHandlerContext.fireChannelReadComplete()
ChannelHandlerContext.fireExceptionCaught(Throwable)
ChannelHandlerContext.fireUserEventTriggered(Object)
ChannelHandlerContext.fireChannelWritabilityChanged()
ChannelHandlerContext.fireChannelInactive()
ChannelHandlerContext.fireChannelUnregistered()

对于Outbound事件来说,可以调用下面的方法,进行事件的传递:

ChannelHandlerContext.bind(SocketAddress, ChannelPromise)
ChannelHandlerContext.connect(SocketAddress, SocketAddress, ChannelPromise)
ChannelHandlerContext.write(Object, ChannelPromise)
ChannelHandlerContext.flush()
ChannelHandlerContext.read()
ChannelHandlerContext.disconnect(ChannelPromise)
ChannelHandlerContext.close(ChannelPromise)
ChannelHandlerContext.deregister(ChannelPromise)

具体而言,就是在handler中调用ChannelHandlerContext中对应的方法:

   public class MyInboundHandler extends ChannelInboundHandlerAdapter {
        @Override
       public void channelActive(ChannelHandlerContext ctx) {
           System.out.println("Connected!");
           ctx.fireChannelActive();
       }
   }
  
   public class MyOutboundHandler extends ChannelOutboundHandlerAdapter {
        @Override
       public void close(ChannelHandlerContext ctx, ChannelPromise promise) {
           System.out.println("Closing ..");
           ctx.close(promise);
       }
   }
   

DefaultChannelPipeline

ChannelPipeline有一个官方的实现叫做DefaultChannelPipeline,因为对于pipeline来说,主要的功能就是进行handler的管理和事件传递,相对于而言功能比较简单,但是他也有一些特别的实现地方,比如它有两个AbstractChannelHandlerContext类型的head和tail。

我们知道ChannelPipeline实际上是很多handler的集合,那么这些集合是怎么进行存储的呢?这种存储的数据结构就是AbstractChannelHandlerContext。每个AbstractChannelHandlerContext中都有一个next节点和一个prev节点,用来组成一个双向链表。

同样的在DefaultChannelPipeline中使用head和tail来将封装好的handler存储起来。

注意,这里的head和tail虽然都是AbstractChannelHandlerContext,但是两者有稍许不同。先看下head和tail的定义:

    protected DefaultChannelPipeline(Channel channel) {
        this.channel = ObjectUtil.checkNotNull(channel, "channel");
        succeededFuture = new SucceededChannelFuture(channel, null);
        voidPromise =  new VoidChannelPromise(channel, true);

        tail = new TailContext(this);
        head = new HeadContext(this);

        head.next = tail;
        tail.prev = head;
    }

在DefaultChannelPipeline的构造函数中,对tail和head进行初始化,其中tail是TailContext,而head是HeadContext。

其中TailContext实现了ChannelInboundHandler接口:

final class TailContext extends AbstractChannelHandlerContext implements ChannelInboundHandler

而HeadContext实现了ChannelOutboundHandler和ChannelInboundHandler接口:

final class HeadContext extends AbstractChannelHandlerContext
            implements ChannelOutboundHandler, ChannelInboundHandler 

下面我们以addFirst方法为例,来看一下handler是怎么被加入pipline的:

    public final ChannelPipeline addFirst(EventExecutorGroup group, String name, ChannelHandler handler) {
        final AbstractChannelHandlerContext newCtx;
        synchronized (this) {
            checkMultiplicity(handler);
            name = filterName(name, handler);

            newCtx = newContext(group, name, handler);

            addFirst0(newCtx);

            // If the registered is false it means that the channel was not registered on an eventLoop yet.
            // In this case we add the context to the pipeline and add a task that will call
            // ChannelHandler.handlerAdded(...) once the channel is registered.
            if (!registered) {
                newCtx.setAddPending();
                callHandlerCallbackLater(newCtx, true);
                return this;
            }

            EventExecutor executor = newCtx.executor();
            if (!executor.inEventLoop()) {
                callHandlerAddedInEventLoop(newCtx, executor);
                return this;
            }
        }
        callHandlerAdded0(newCtx);
        return this;
    }

它的工作逻辑是首先根据传入的handler构建一个新的context,然后调用addFirst0方法,将context加入AbstractChannelHandlerContext组成的双向链表中:

    private void addFirst0(AbstractChannelHandlerContext newCtx) {
        AbstractChannelHandlerContext nextCtx = head.next;
        newCtx.prev = head;
        newCtx.next = nextCtx;
        head.next = newCtx;
        nextCtx.prev = newCtx;
    }

然后调用callHandlerAdded0方法来触发context的handlerAdded方法。

总结

channelPipeline负责管理channel的各种handler,在DefaultChannelPipeline中使用了AbstractChannelHandlerContext的head和tail来对多个handler进行存储,同时借用这个链式结构对handler进行各种管理,非常方便。

本文已收录于 http://www.flydean.com/04-3-netty-channelpipeline/

最通俗的解读,最深刻的干货,最简洁的教程,众多你不知道的小技巧等你来发现!

欢迎关注我的公众号:「程序那些事」,懂技术,更懂你!

点赞
收藏
评论区
推荐文章
blmius blmius
4年前
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(
Wesley13 Wesley13
4年前
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中是否包含分隔符'',缺省为
Jacquelyn38 Jacquelyn38
4年前
2020年前端实用代码段,为你的工作保驾护航
有空的时候,自己总结了几个代码段,在开发中也经常使用,谢谢。1、使用解构获取json数据let jsonData  id: 1,status: "OK",data: 'a', 'b';let  id, status, data: number   jsonData;console.log(id, status, number )
Wesley13 Wesley13
4年前
mysql设置时区
mysql设置时区mysql\_query("SETtime\_zone'8:00'")ordie('时区设置失败,请联系管理员!');中国在东8区所以加8方法二:selectcount(user\_id)asdevice,CONVERT\_TZ(FROM\_UNIXTIME(reg\_time),'08:00','0
Wesley13 Wesley13
4年前
NEO从源码分析看UTXO交易
_0x00前言_社区大佬:“交易是操作区块链的唯一方式。”_0x01交易类型_在NEO中,几乎除了共识之外的所有的对区块链的操作都是一种“交易”,甚至在“交易”面前,合约都只是一个小弟。交易类型的定义在Core中的TransactionType中:源码位置:neo/Core/TransactionType
Stella981 Stella981
4年前
Netty的服务端Channel不支持写操作
作为Netty的服务端Channel,即NioServerSocketChannel,它的职责就是负责监听ACCEPT事件,当有客户端连接到服务端时,NioServerSocketChannel负责读取这个ACCEPT请求.既然作为一个Channel,都是支持读和写操作的.那么这个NioServerSocketChannel是不支持写操
Stella981 Stella981
4年前
Linux中Too many open files 问题分析和解决
WARN2018061516:55:20,831NewI/Oserverboss1(id:0x55007b59,/0.0.0.0:20880)org.jboss.netty.channel.socket.nio.NioServerSocketPipelineSink DUBBOFailedt
Stella981 Stella981
4年前
Netty多人聊天室
在简单聊天室的代码中修改ChatServerHandler类,就可以模拟多人聊天的功能packagecom.cppdy.server;importio.netty.channel.Channel;importio.netty.channel.ChannelHandlerContext;importi
Stella981 Stella981
4年前
ELK学习笔记之ElasticSearch的索引详解
0x00ElasticSearch的索引和MySQL的索引方式对比Elasticsearch是通过Lucene的倒排索引技术实现比关系型数据库更快的过滤。特别是它对多条件的过滤支持非常好,比如年龄在18和30之间,性别为女性这样的组合查询。倒排索引很多地方都有介绍,但是其比关系型