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

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

目录概述编码器第一节: writeAndFlush的事件传播我们看一个最简单的使用的场景我们跟到writeAndFlush方法中我们跟到invokeWriteAndFlush中我们再看invokeFlush0方法

概述

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

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

编码器

第一节: writeAndFlush的事件传播

我们之前在学习pipeline的时候, 讲解了write事件的传播过程, 但在实际使用的时候, 我们通常不会调用channel的write方法, 因为该方法只会写入到发送数据的缓存中, 并不会直接写入channel中, 如果想写qHewvmwK入到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分布式编码器及写数据事件的资料请关注我们其它相关文章！

版权声明：本文内容由网络用户投稿，版权归原作者所有，本站不拥有其著作权，亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容，请联系我们jiasou666@gmail.com 处理，核实后本网站将在24小时内删除侵权内容。