Netty分布式编码器及写数据事件处理使用场景

时间：2022-03-29 15:01 来源：未知作者：玩人必被人玩收藏

这篇文章主要为大家介绍了Netty分布式编码器及写数据事件处理使用场景剖析，有需要的朋友可以借鉴参考下，希望能够有所帮助，

概述

上一小章我们介绍了解码器,这一章我们介绍编码器

其实编码器和解码器比较类似,编码器也是一个handler,并且属于outbounfHandle,就是将准备发出去的数据进行拦截,拦截之后进行相应的处理之后再次进发送处理,如果理解了解码器,那么编码器的相关内容理解起来也比较容易

编码器

第一节:writeAndFlush的事件传播

我们之前在学习pipeline的时候,讲解了write事件的传播过程,但在实际使用的时候,我们通常不会调用channel的write方法,因为该方法只会写入到发送数据的缓存中,并不会直接写入channel中,如果想写入到channel中,还需要调用flush方法

实际使用过程中,我们用的更多的是writeAndFlush方法,这方法既能将数据写到发送缓存中,也能刷新到channel中

我们看一个最简单的使用的场景

public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
    ctx.channel().writeAndFlush("test data");
}

学过netty的同学们对此肯定不陌生,通过这种方式,可以将数据发送到channel中,对方可以收到响应

我们跟到writeAndFlush方法中

首先会走到AbstractChannel的writeAndFlush:

public ChannelFuture writeAndFlush(Object msg) {
    return pipeline.writeAndFlush(msg);
}

继续跟到DefualtChannelPipeline中的writeAndFlush方法中:

public final ChannelFuture writeAndFlush(Object msg) {
    return tail.writeAndFlush(msg);
}

这里我们看到, writeAndFlush是从tail节点进行传播,有关事件传播,我们再pipeline中进行过剖析,相信这个不会陌生

继续跟,会跟到AbstractChannelHandlerContext中的writeAndFlush方法:

public ChannelFuture writeAndFlush(Object msg) {
    return writeAndFlush(msg, newPromise());
}

继续跟:

public ChannelFuture writeAndFlush(Object msg, ChannelPromise promise) {
    if (msg == null) {
        throw new NullPointerException("msg");
    }
    if (!validatePromise(promise, true)) {
        ReferenceCountUtil.release(msg);
        // cancelled
        return promise;
    } 
    write(msg, true, promise);
    return promise;
}

继续跟write方法:

private void write(Object msg, boolean flush, ChannelPromise promise) {
    //findContextOutbound()寻找前一个outbound节点
    //最后到head节点结束
    AbstractChannelHandlerContext next = findContextOutbound();
    final Object m = pipeline.touch(msg, next);
    EventExecutor executor = next.executor();
    if (executor.inEventLoop()) {
        if (flush) {
            next.invokeWriteAndFlush(m, promise);
        } else {
            //没有调flush
            next.invokeWrite(m, promise);
        }
    } else {
        AbstractWriteTask task;
        if (flush) {
            task = WriteAndFlushTask.newInstance(next, m, promise);
        }  else {
            task = WriteTask.newInstance(next, m, promise);
        }
        safeExecute(executor, task, promise, m);
    }
}

这里的逻辑我们也不陌生,找到下一个节点,因为writeAndFlush是从tail节点开始的,并且是outBound的事件,所以这里会找到tail节点的上一个outBoundHandler,有可能是编码器,也有可能是我们业务处理的handler

if(executor.inEventLoop())判断是否是eventLoop线程,如果不是,则封装成task通过nioEventLoop异步执行,我们这里先按照是eventLoop线程分析

首先,这里通过flush判断是否调用了flush,这里显然是true,因为我们调用的方法是writeAndFlush

我们跟到invokeWriteAndFlush中

private void invokeWriteAndFlush(Object msg, ChannelPromise promise) {
    if (invokeHandler()) {
        //写入
        invokeWrite0(msg, promise);
        //刷新
        invokeFlush0();
    } else {
        writeAndFlush(msg, promise);
    }
}

这里就真相大白了,其实在writeAndFlush中,首先调用write, write完成之后再调用flush方法进行的刷新

首先跟到invokeWrite0方法中:

private void invokeWrite0(Object msg, ChannelPromise promise) {
    try {
        //调用当前handler的wirte()方法
        ((ChannelOutboundHandler) handler()).write(this, msg, promise);
    } catch (Throwable t) {
        notifyOutboundHandlerException(t, promise);
    }
}

该方法我们在pipeline中已经进行过分析,就是调用当前handler的write方法,如果当前handler中write方法是继续往下传播,在会继续传播写事件,直到传播到head节点,最后会走到HeadContext的write方法中

跟到HeadContext的write方法中:

public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
    unsafe.write(msg, promise);
}

这里通过当前channel的unsafe对象对将当前消息写到缓存中,写入的过程,我们之后的小节进行分析

回到到invokeWriteAndFlush方法中:

private void invokeWriteAndFlush(Object msg, ChannelPromise promise) {
    if (invokeHandler()) {
        //写入
        invokeWrite0(msg, promise);
        //刷新
        invokeFlush0();
    } else {
        writeAndFlush(msg, promise);
    }
}

我们再看invokeFlush0方法

private void invokeFlush0() {
    try {
        ((ChannelOutboundHandler) handler()).flush(this);
    } catch (Throwable t) {
        notifyHandlerException(t);
    }
}

同样,这里会调用当前handler的flush方法,如果当前handler的flush方法是继续传播flush事件,则flush事件会继续往下传播,直到最后会调用head节点的flush方法,如果我们熟悉pipeline的话,对这里的逻辑不会陌生

跟到HeadContext的flush方法中:

public void flush(ChannelHandlerContext ctx) throws Exception {
    unsafe.flush();
}

这里同样,会通过当前channel的unsafe对象通过调用flush方法将缓存的数据刷新到channel中,有关刷新的逻辑,我们会在以后的小节进行剖析

以上就是writeAndFlush的相关逻辑,整体上比较简单,熟悉pipeline的同学应该很容易理解

更多关于Netty分布式编码器及写数据事件的资料请关注源码搜藏网其它相关文章！

标签：源码网站源码

上一篇：Java几个实例带你进阶升华下篇

下一篇：没有了

Apache Solr 初级教程（介绍、安装部署、Ja

Java枚举与普通类的区别和使用全解析

Java(5/6和8)中interface和Abstract Class

存储过程的单步调试方法

对象实例是何时被创建

免费java视频下载，struts视频，hibernate

Eclipse中配置Ajax的DWR的简单方法

动力节点_王勇_ DRP项目视频教程完整版292

Oracle密谋B计划 JDK 7或成跳票王

XML配置文件的读取处理

Java 泛型总结（二）：泛型与数组

Java动态代理

热门源码

开发教程

Netty分布式编码器及写数据事件处理使用场景

概述

编码器

第一节:writeAndFlush的事件传播

我们看一个最简单的使用的场景

我们跟到writeAndFlush方法中

我们跟到invokeWriteAndFlush中

我们再看invokeFlush0方法

java教程阅读排行

最新文章

服务咨询

整站源码

网站模板

免费源码

关于我们

扫码关注有惊喜