编码器实现了ChannelOutboundHandler
,并将出站数据从 一种格式转换为另一种格式,和我们方才学习的解码器的功能正好相反。Netty 提供了一组类, 用于帮助你编写具有以下功能的编码器:
解码器通常需要在Channel
关闭之后产生最后一个消息(因此也就有了 decodeLast()
方法)
这显然不适于编码器的场景——在连接被关闭之后仍然产生一个消息是毫无意义的
其接受一Short
型实例作为消息,编码为Short
的原子类型值,并写入ByteBuf
,随后转发给ChannelPipeline
中的下一个 ChannelOutboundHandler
每个传出的 Short 值都将会占用 ByteBuf 中的 2 字节
Netty 提供了一些专门化的 MessageToByteEncoder
,可基于此实现自己的编码器
WebSocket08FrameEncoder
类提供了一个很好的实例
你已经看到了如何将入站数据从一种消息格式解码为另一种
为了完善这幅图,将展示 对于出站数据将如何从一种消息编码为另一种。MessageToMessageEncoder
类的 encode()
方法提供了这种能力
为了演示,使用IntegerToStringEncoder
扩展了 MessageToMessageEncoder
关于有趣的 MessageToMessageEncoder 的专业用法,请查看 io.netty.handler. codec.protobuf.ProtobufEncoder
类,它处理了由 Google 的 Protocol Buffers 规范所定义 的数据格式。
java对象编码过程 write:写队列 flush:刷新写队列 writeAndFlush: 写队列并刷新
数据从head节点流入,先拆包,然后解码成业务对象,最后经过业务Handler
处理,调用write
,将结果对象写出去
而写的过程先通过tail
节点,然后通过encoder
节点将对象编码成ByteBuf
,最后将该ByteBuf
对象传递到head
节点,调用底层的Unsafe写到JDK底层管道
为什么我们在pipeline中添加了encoder节点,java对象就转换成netty可以处理的ByteBuf,写到管道里?
我们先看下调用write的code
业务处理器接受到请求之后,做一些业务处理,返回一个user
handler 如果不覆盖 flush 方法,就会一直向前传递直到 head 节点
落到 Encoder
节点,下面是 Encoder
的处理流程
按照简单自定义协议,将Java对象 User 写到传入的参数 out中,这个out到底是什么?
需知User
对象,从BizHandler
传入到 MessageToByteEncoder
时,首先传到 write
encode
,这里就调回到 Encoder
这个Handler
中
ByteBuf
了,那么这个对象就已经无用,释放掉 (当传入的msg
类型是ByteBuf
时,就不需要自己手动释放了)
//112 如果buf中写入了数据,就把buf传到下一个节点,直到 header 节点
//115 否则,释放buf,将空数据传到下一个节点 // 120 如果当前节点不能处理传入的对象,直接扔给下一个节点处理 // 127 当buf在pipeline中处理完之后,释放
Handler
是否能处理写入的消息 Encoder
可以处理的 Response
对象ByteBuf
encoder
,即进入到 Encoder 的 encode方法,该方法是用户代码,用户将数据写入ByteBuf总结就是,Encoder
节点分配一个ByteBuf
,调用encode
方法,将Java对象根据自定义协议写入到ByteBuf,然后再把ByteBuf传入到下一个节点,在我们的例子中,最终会传入到head节点
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
unsafe.write(msg, promise);
}
这里的msg就是前面在Encoder节点中,载有java对象数据的自定义ByteBuf对象
以下过程分三步讲解
assertEventLoop
确保该方法的调用是在reactor
线程中filterOutboundMessage()
,将待写入的对象过滤,把非ByteBuf
对象和FileRegion
过滤,把所有的非直接内存转换成直接内存DirectBuffer
想要理解上面这段代码,须掌握写缓存中的几个消息指针
ChannelOutboundBuffer 里面的数据结构是一个单链表结构,每个节点是一个 Entry,Entry 里面包含了待写出ByteBuf 以及消息回调 promise下面分别是
ChannelOutboundBuffer
缓冲区的最后一个节点
addMessage
后
fushedEntry
指向空,unFushedEntry
和 tailEntry
都指向新加入节点
addMessage
后
addMessage
后
可得,调用n次addMessage
后
flushedEntry
指针一直指向null
,表此时尚未有节点需写到Socket缓冲区unFushedEntry
后有n个节点,表当前还有n个节点尚未写到Socket缓冲区channel.flush()
,还是ctx.flush()
,最终都会落地到pipeline
中的head
节点
AbstractUnsafe
unflushedEntry
指针,然后将flushedEntry
指向unflushedEntry
所指向的节点,调用完毕后
flush0()
doWrite
protected void doWrite(ChannelOutboundBuffer in) throws Exception {
int writeSpinCount = -1;
boolean setOpWrite = false;
for (;;) {
// 拿到第一个需要flush的节点的数据
Object msg = in.current();
if (msg instanceof ByteBuf) {
boolean done = false;
long flushedAmount = 0;
// 拿到自旋锁迭代次数
if (writeSpinCount == -1) {
writeSpinCount = config().getWriteSpinCount();
}
// 自旋,将当前节点写出
for (int i = writeSpinCount - 1; i >= 0; i --) {
int localFlushedAmount = doWriteBytes(buf);
if (localFlushedAmount == 0) {
setOpWrite = true;
break;
}
flushedAmount += localFlushedAmount;
if (!buf.isReadable()) {
done = true;
break;
}
}
in.progress(flushedAmount);
// 写完之后,将当前节点删除
if (done) {
in.remove();
} else {
break;
}
}
}
}
current()
先拿到第一个需要flush
的节点的数据
ByteBuf
写到JDK NIO的Channel
强转为ByteBuf,若发现没有数据可读,直接删除该节点
javaChannel()
,表明 JDK NIO Channel 已介入此次事件
首先拿到当前被flush
掉的节点(flushedEntry
所指)
然后拿到该节点的回调对象 ChannelPromise
, 调用 removeEntry()
移除该节点
这里是逻辑移除,只是将flushedEntry指针移到下个节点,调用后
随后,释放该节点数据的内存,调用safeSuccess
回调,用户代码可以在回调里面做一些记录,下面是一段Example
最后,调用 recycle
,将当前节点回收
writeAndFlush在某个Handler
中被调用后,最终会落到 TailContext
节点
通过一个boolean变量flush,表明调用invokeWriteAndFlush
or invokeWrite
,invokeWrite
便是我们上文中的write过程。
可以看到,最终调用的底层方法和单独调用write
和flush
一样的
由此看来,invokeWriteAndFlush
基本等价于write
之后再来一次flush
。
write
并没有将数据写到Socket缓冲区中,而是写到了一个单向链表的数据结构中,flush
才是真正的写出writeAndFlush
等价于先将数据写到netty的缓冲区,再将netty缓冲区中的数据写到Socket缓冲区中,写的过程与并发编程类似,用自旋锁保证写成功当 BizHandler 通过 writeAndFlush 方法将自定义对象往前传播时,其实可以拆分成两个过程
ByteBuf
,将Java对象转换为ByteBuf
,然后再把ByteBuf
继续向前传递,若没有再重写了,最终会传播到 head 节点,其中缓冲区列表拿到缓存写到 JDK 底层 ByteBuffer