Netty分布式pipeline管道Handler的添加代码跟踪解析

目录
  • 添加handler
    • 我们跟到其addLast()方法中
    • 再继续跟到addLast()方法中去
    • 我们跟到checkMultiplicity(handler)中
    • 跟到filterName方法中
    • 跟到isInbound(handler)方法中
    • 我们回到最初的addLast()方法中
    • 我们跟进addLast0(newCtx)中

前文传送门:Netty分布式pipeline管道创建

添加handler

我们以用户代码为例进行剖析:

.childHandler(new ChannelInitializer<SocketChannel>() {
    protected void initChannel(SocketChannel ch) throws Exception {
        ch.pipeline().addLast(new DelimiterBasedFrameDecoder(Integer.MAX_VALUE, Delimiters.lineDelimiter()[0]));
        ch.pipeline().addLast(new StringEncoder());
        ch.pipeline().addLast(new SimpleHandler());
    }
});

用过netty的小伙伴们肯定对这段代码不会陌生, 通过addLast, 可以添加编解码器和我们自定义的handler, 某一个事件完成之后可以自动调用我们handler预先定义的方法, 具体添加和调用是怎么个执行逻辑, 在我们之后的内容会全部学习到, 以后再使用这类的功能会得心应手

在这里, 我们主要剖析 ch.pipeline().addLast(new SimpleHandler()) 这部分代码的addLast()方法

首先通过channel拿到当前的pipline, 这个上一小节进行剖析过相信不会陌生

拿到pipeline之后再为其添加handler, 因为channel初始化默认创建的是DefualtChannelPipeline

我们跟到其addLast()方法中

public final ChannelPipeline addLast(ChannelHandler... handlers) {
    return addLast(null, handlers);
}

首先看到这里的参数其实是一个可变对象, 也就是可以传递多个handler, 这里我们只传递了一个

我们继续跟addLast:

public final ChannelPipeline addLast(EventExecutorGroup executor, ChannelHandler... handlers) {
    if (handlers == null) {
        throw new NullPointerException("handlers");
    }
    //传多个参数的时候通过for循环添加
    for (ChannelHandler h: handlers) {
        if (h == null) {
            break;
        }
        addLast(executor, null, h);
    }
    return this;
}

这里如果传入多个handler则会循环添加, 我们通常只添加一个

再继续跟到addLast()方法中去

public final ChannelPipeline addLast(EventExecutorGroup group, String name, ChannelHandler handler) {
    final AbstractChannelHandlerContext newCtx;
    synchronized (this) {
        //判断handler是否被重复添加(1)
        checkMultiplicity(handler);
        //创建一个HandlerContext并添加到列表(2)
        newCtx = newContext(group, filterName(name, handler), handler);

        //添加HandlerContext(3)
        addLast0(newCtx);

        //是否已注册
        if (!registered) {
            newCtx.setAddPending();
            callHandlerCallbackLater(newCtx, true);
            return this;
        }

        EventExecutor executor = newCtx.executor();
        if (!executor.inEventLoop()) {
            newCtx.setAddPending();
            //回调用户事件
            executor.execute(new Runnable() {
                @Override
                public void run() {
                    callHandlerAdded0(newCtx);
                }
            });
            return this;
        }
    }
    //回调添加事件(4)
    callHandlerAdded0(newCtx);
    return this;
}

这部分代码比较长, 我们拆解为4个步骤:

1.重复添加验证

2.创建一个HandlerContext并添加到列表

3. 添加context

4. 回调添加事件

首先我们看第一步, 重复添加验证

我们跟到checkMultiplicity(handler)中

private static void checkMultiplicity(ChannelHandler handler) {
    if (handler instanceof ChannelHandlerAdapter) {
        ChannelHandlerAdapter h = (ChannelHandlerAdapter) handler;
        if (!h.isSharable() && h.added) {
            throw new ChannelPipelineException(
                    h.getClass().getName() +
                    " is not a @Sharable handler, so can't be added or removed multiple times.");
        }
        //满足条件设置为true, 代表已添加
        h.added = true;
    }
}

首先判断是不是ChannelHandlerAdapter类型, 因为我们自定义的handler通常会直接或者间接的继承该接口, 所以这里为true

拿到handler之后转换成ChannelHandlerAdapter类型, 然后进行条件判断

if (!h.isSharable() && h.added) 代表如果不是共享的handler, 并且是未添加状态, 则抛出异常:

我们可以跟到isSharable()方法中去:

public boolean isSharable() {
    Class<?> clazz = getClass();
    Map<Class<?>, Boolean> cache = InternalThreadLocalMap.get().handlerSharableCache();
    Boolean sharable = cache.get(clazz);
    if (sharable == null) {
        //如果这个类注解了Sharable.class, 说明这个类会被多个channel共享
        sharable = clazz.isAnnotationPresent(Sharable.class);
        cache.put(clazz, sharable);
    }
    return sharable;
}

首先拿到当前handler的class对象

然后再从netty自定义的一个ThreadLocalMap对象中获取一个盛放handler的class对象的map, 并获取其value

如果value值为空, 则会判断是否被Sharable注解, 并将自身handler的class对象和判断结果存入map对象中, 最后返回判断结果

这说明了被Sharable注解的handler是一个共享handler

从这个逻辑我们可以判断, 共享对象是可以重复添加的

我们回到checkMultiplicity(handler)方法中:

如果是共享对象或者没有被添加, 则将ChannelHandlerAdapter的added设置为true, 代表已添加

剖析完了重复添加验证, 回到addLast方法中, 我们看第二步, 创建一个HandlerContext并添加到列表:

newCtx = newContext(group, filterName(name, handler), handler);

首先看filterName(name, handler)方法, 这个方法是判断添加handler的name是否重复

跟到filterName方法中

private String filterName(String name, ChannelHandler handler) {
    if (name == null) {
        //没有名字创建默认名字
        return generateName(handler);
    }
    //检查名字是否重复
    checkDuplicateName(name);
    return name;
}

因为我们添加handler时候, 不一定会会给handler命名, 所以这一步name有可能是null, 如果是null, 则创建一个默认的名字, 这里创建名字的方法我们就不往里跟了, 有兴趣的同学可以自己跟进去看

然后再检查名字是否重复

我们跟到checkDuplicateName(name)这个方法中:

private void checkDuplicateName(String name) {
    //不为空
    if (context0(name) != null) {
        throw new IllegalArgumentException("Duplicate handler name: " + name);
    }
}

这里有个context0(name)方法, 我们跟进去:

private AbstractChannelHandlerContext context0(String name) {
    //遍历pipeline
    AbstractChannelHandlerContext context = head.next;
    while (context != tail) {
        //发现name相同, 说明存在handler
        if (context.name().equals(name)) {
            //返回
            return context;
        }
        context = context.next;
    }
    return null;
}

这里做的操作非常简单, 就是将pipeline中, 从head节点往下遍历HandlerContext, 一直遍历到tail, 如果发现名字相同则会认为重复并返回HandlerContext对象

我们回到addLast()方法中并继续看添加创建相关的逻辑:

newCtx = newContext(group, filterName(name, handler), handler)

filterName(name, handler)这步如果并没有重复则会返回handler的name

我们继续跟到newContext(group, filterName(name, handler), handler)方法中:

private AbstractChannelHandlerContext newContext(EventExecutorGroup group, String name, ChannelHandler handler) {
    return new DefaultChannelHandlerContext(this, childExecutor(group), name, handler);
}

这里我们看到创建了一个DefaultChannelHandlerContext对象, 构造方法的参数中, 第一个this代表当前的pipeline对象, group为null, 所以childExecutor(group)也会返回null, name为handler的名字, handler为新添加的handler对象

我们继续跟到DefaultChannelHandlerContext的构造方法中:

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

我们看到首先调用了父类的构造方法, 之后将handler赋值为自身handler的成员变量, HandlerConext和handler关系在此也展现了出来, 是一种组合关系

我们首先看父类的构造方法, 有这么两个参数:isInbound(handler), isOutbound(handler), 这两个参数意思是判断需要添加的handler是inboundHandler还是outBoundHandler

跟到isInbound(handler)方法中

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

这里通过是否实现ChannelInboundHandler接口来判断是否为inboundhandler

同样我们看isOutbound(handler)方法:

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

通过判断是否实现ChannelOutboundHandler接口判断是否为outboundhandler

在跟到其父类AbstractChannelHandlerContext的构造方法中:

AbstractChannelHandlerContext(DefaultChannelPipeline pipeline, EventExecutor executor, String name,
                              boolean inbound, boolean outbound) {
    this.name = ObjectUtil.checkNotNull(name, "name");
    this.pipeline = pipeline;
    this.executor = executor;
    this.inbound = inbound;
    this.outbound = outbound;
    ordered = executor == null || executor instanceof OrderedEventExecutor;
}

一切都不陌生了, 因为我们tail节点和head节点创建的时候同样走到了这里

这里初始化了name, pipeline, 以及标识添加的handler是inboundhanlder还是outboundhandler

我们回到最初的addLast()方法中

public final ChannelPipeline addLast(EventExecutorGroup group, String name, ChannelHandler handler) {
    final AbstractChannelHandlerContext newCtx;
    synchronized (this) {
        //判断handler是否被重复添加(1)
        checkMultiplicity(handler);
        //创建一个HandlerContext并添加到列表(2)
        newCtx = newContext(group, filterName(name, handler), handler);
        //添加HandlerContext(3)
        addLast0(newCtx);
        //是否已注册
        if (!registered) {
            newCtx.setAddPending();
            callHandlerCallbackLater(newCtx, true);
            return this;
        }
        EventExecutor executor = newCtx.executor();
        if (!executor.inEventLoop()) {
            newCtx.setAddPending();
            //回调用户事件
            executor.execute(new Runnable() {
                @Override
                public void run() {
                    callHandlerAdded0(newCtx);
                }
            });
            return this;
        }
    }
    //回调添加事件(4)
    callHandlerAdded0(newCtx);
    return this;
}

我们跟完了创建HandlerContext的相关逻辑, 我们继续跟第三步, 添加HandlerContext

我们跟进addLast0(newCtx)中

private void addLast0(AbstractChannelHandlerContext newCtx) {
    //拿到tail节点的前置节点
    AbstractChannelHandlerContext prev = tail.prev;
    //当前节点的前置节点赋值为tail节点的前置节点
    newCtx.prev = prev;
    //当前节点的下一个节点赋值为tail节点
    newCtx.next = tail;
    //tail前置节点的下一个节点赋值为当前节点
    prev.next = newCtx;
    //tail节点的前一个节点赋值为当前节点
    tail.prev = newCtx;
}

这一部分也非常简单, 做了一个指针的指向操作, 将新添加的handlerConext放在tail节点之前, 之前tail节点的上一个节点之后, 熟悉双向链表的同学对此逻辑应该不会陌生, 如果是第一次添加handler, 那么添加后的结构入下图所示:

添加完handler之后, 这里会判断当前channel是否已经注册, 这部分逻辑我们之后再进行剖析, 我们继续往下走

之后会判断当前线程线程是否为eventLoop线程, 如果不是eventLoop线程, 就将添加回调事件封装成task交给eventLoop线程执行, 否则, 直接执行添加回调事件callHandlerAdded0(newCtx)

跟进callHandlerAdded0(newCtx)

private void callHandlerAdded0(final AbstractChannelHandlerContext ctx) {
    try {
        ctx.handler().handlerAdded(ctx);
        ctx.setAddComplete();
    } catch (Throwable t) {
        //忽略代码
    }
}

我们重点关注这句

ctx.handler().handlerAdded(ctx);

其中ctx是我们新创建的HandlerContext, 通过handler()方法拿到绑定的handler, 也就是新添加的handler, 然后执行handlerAdded(ctx)方法, 如果我们没有重写这个方法, 则会执行父类的该方法

在ChannelHandlerAdapter类中定义了该方法的实现:

@Override
public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
}

我们看到没做任何操作, 也就是如果我们没有重写该方法时, 如果添加handler之后将不会做任何操作, 这里如果我们需要做一些业务逻辑, 可以通过重写该方法进行实现

以上就是添加handler的有关的业务逻辑,更多关于Netty分布式pipeline管道添加Handler的资料请关注我们其它相关文章!

(0)

相关推荐

  • Netty分布式源码分析监听读事件

    前文传送门:NioSocketChannel注册到selector 我们回到AbstractUnsafe的register0()方法: private void register0(ChannelPromise promise) { try { //省略代码 //做实际的注册 doRegister(); neverRegistered = false; registered = true; //触发事件 pipeline.invokeHandlerAddedIfNeeded(); safeSet

  • Netty分布式客户端接入流程初始化源码分析

    目录 前文概述: 第一节:初始化NioSockectChannelConfig 创建channel 跟到其父类DefaultChannelConfig的构造方法中 再回到AdaptiveRecvByteBufAllocator的构造方法中 继续跟到ChannelMetadata的构造方法中 回到DefaultChannelConfig的构造方法 前文概述: 之前的章节学习了server启动以及eventLoop相关的逻辑, eventLoop轮询到客户端接入事件之后是如何处理的?这一章我们循序渐

  • Netty客户端接入流程NioSocketChannel创建解析

    目录 NioSocketChannel的创建 回到上一小节的read()方法 我们首先看readBuf jdk底层相关的内容 跟到父类构造方法中 我们跟进其构造方法 前文传送门:Netty客户端处理接入事件handle创建 NioSocketChannel的创建 回到上一小节的read()方法 public void read() { //必须是NioEventLoop方法调用的, 不能通过外部线程调用 assert eventLoop().inEventLoop(); //服务端channel

  • Netty分布式客户端处理接入事件handle源码解析

    目录 处理接入事件创建handle 我们看其RecvByteBufAllocator接口 跟进newHandle()方法中 继续回到read()方法 我们跟进reset中 前文传送门 :客户端接入流程初始化源码分析 上一小节我们剖析完成了与channel绑定的ChannelConfig初始化相关的流程, 这一小节继续剖析客户端连接事件的处理 处理接入事件创建handle 回到上一章NioEventLoop的processSelectedKey ()方法 private void processS

  • Netty分布式NioEventLoop任务队列执行源码分析

    目录 执行任务队列 跟进runAllTasks方法: 我们跟进fetchFromScheduledTaskQueue()方法 回到runAllTasks(long timeoutNanos)方法中 章节小结 前文传送门:NioEventLoop处理IO事件 执行任务队列 继续回到NioEventLoop的run()方法: protected void run() { for (;;) { try { switch (selectStrategy.calculateStrategy(selectN

  • Netty分布式NioSocketChannel注册到selector方法解析

    目录 我们回到最初的NioMessageUnsafe的read()方法: public void read() { //必须是NioEventLoop方法调用的, 不能通过外部线程调用 assert eventLoop().inEventLoop(); //服务端channel的config final ChannelConfig config = config(); //服务端channel的pipeline final ChannelPipeline pipeline = pipeline(

  • Netty分布式pipeline管道Handler的添加代码跟踪解析

    目录 添加handler 我们跟到其addLast()方法中 再继续跟到addLast()方法中去 我们跟到checkMultiplicity(handler)中 跟到filterName方法中 跟到isInbound(handler)方法中 我们回到最初的addLast()方法中 我们跟进addLast0(newCtx)中 前文传送门:Netty分布式pipeline管道创建 添加handler 我们以用户代码为例进行剖析: .childHandler(new ChannelInitializ

  • Netty分布式pipeline管道Handler的删除逻辑操作

    目录 删除handler操作 我们跟到getContextPrDie这个方法中 首先要断言删除的节点不能是tail和head 回到remove(ctx)方法 上一小节我们学习了添加handler的逻辑操作, 这一小节我们学习删除handler的相关逻辑 删除handler操作 如果用户在业务逻辑中进行ctx.pipeline().remove(this)这样的写法, 或者ch.pipeline().remove(new SimpleHandler())这样的写法, 则就是对handler进行删除

  • Netty分布式pipeline管道传播outBound事件源码解析

    目录 outbound事件传输流程 这里我们同样给出两种写法 跟到其write方法中: 跟到findContextOutbound中 回到write方法: 继续跟invokeWrite0 我们跟到HeadContext的write方法中 了解了inbound事件的传播过程, 对于学习outbound事件传输的流程, 也不会太困难 outbound事件传输流程 在我们业务代码中, 有可能使用wirte方法往写数据: public void channelActive(ChannelHandlerC

  • Netty分布式pipeline管道传播事件的逻辑总结分析

    目录 问题分析 首先完成了handler的添加, 但是并没有马上执行回调 回到callHandlerCallbackLater方法中 章节总结 我们在第一章和第三章中, 遗留了很多有关事件传输的相关逻辑, 这里带大家一一回顾 问题分析 首先看两个问题: 1.在客户端接入的时候, NioMessageUnsafe的read方法中pipeline.fireChannelRead(readBuf.get(i))为什么会调用到ServerBootstrap的内部类ServerBootstrapAccep

  • Netty分布式pipeline管道创建方法跟踪解析

    目录 概述 pipeline的创建 上一章节回顾:Netty分布式源码分析监听读事件 概述 pipeline, 顾名思义, 就是管道的意思, 在netty中, 事件在pipeline中传输, 用户可以中断事件, 添加自己的事件处理逻辑, 可以直接将事件中断不再往下传输, 同样可以改变管道的流向, 传递其他事件.这里有点类似于Spring的AOP, 但是比AOP实现起来简单的多 事件通常分为两种, 一是inBound事件, 另一种是outBound事件, inBound事件, 顾名思义, 就是从另

  • Netty分布式pipeline管道异常传播事件源码解析

    目录 传播异常事件 简单的异常处理的场景 我们跟到invokeChannelRead这个方法 我还是通过两种写法来进行剖析 跟进invokeExceptionCaught方法 跟到invokeExceptionCaught方法中 讲完了inbound事件和outbound事件的传输流程, 这一小节剖析异常事件的传输流程 传播异常事件 简单的异常处理的场景 @Override public void channelRead(ChannelHandlerContext ctx, Object msg

  • Netty分布式pipeline传播inbound事件源码分析

    前一小结回顾:pipeline管道Handler删除 传播inbound事件 有关于inbound事件, 在概述中做过简单的介绍, 就是以自己为基准, 流向自己的事件, 比如最常见的channelRead事件, 就是对方发来数据流的所触发的事件, 己方要对这些数据进行处理, 这一小节, 以激活channelRead为例讲解有关inbound事件的处理流程 在业务代码中, 我们自己的handler往往会通过重写channelRead方法来处理对方发来的数据, 那么对方发来的数据是如何走到chann

  • netty中pipeline的handler添加删除分析

    目录 添加 DefaultChannelPipeline.addLast(ChannelHandler... handlers) checkMultiplicity(handler)重复添加验证 isSharable() newCtx = newContext(group, filterName(name, handler), handler) filterName(name, handler) checkDuplicateName(name) context0(name) newContext

  • Netty分布式ByteBuf使用page级别的内存分配解析

    目录 netty内存分配数据结构 我们看PoolArena中有关chunkList的成员变量 我们看PoolSubpage的属性 我们回到PoolArena的allocate方法 我们跟进allocateNormal 首先会从head节点往下遍历 这里直接通过构造函数创建了一个chunk 首先将参数传入的值进行赋值 我们再回到PoolArena的allocateNormal方法中 跟到allocate(normCapacity)中 我们跟到allocateNode方法中 我们跟进updatePa

随机推荐