Netty源码分析第7章(编码器和写数据)---->第1节: writeAndFlush的事件传播

 

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的同窗应该很容易理解

 

上一节: 分隔符解码器

下一节: MessageToByteEncoder