netty二 - netty一些使用体会
2018-09-08 本文已影响0人
Wu杰语
netty学习的意义
对于分布式框架，最底层是什么，毫无疑问就是socket，既然要分布式，那么就要进行通信，靠什么通信，靠soc- ket。
-
要理解一个知识，对于其底层不能不理解，对socket的知识要理解，对socket报文、握手、重传、拥塞等这些基本机制要了解。一般情况下我们用JAVA，那么要知道JAVA的BIO、NIO，甚至AIO，如果是C，那么就需要掌握更加原生的操作系统提供的函数，不同操作系统提供的函数不一样，没有JAVA来的方便。
知道这么多，你会发现Socket不会像JAVA封装的这么简单，即使用JAVA的BIO、NIO封装，如果想写一个工业级的socket通信系统，是非常不容易的。所以netty的出现是对程序员的一个极大的解放。
-
netty让我们理解的不止socket通信那么多。对于一个分布式框架底层通讯系统，有socket通讯、序列化和反序列化，在这个基础上才建立服务。netty的学习让这一切细节暴露的非常清楚，非常有利于人去学习分布式架构通信这一层次的知识。
-
netty什么时候会用到，我想除了需求是私有协议、或者架构上为了提升效率采用私有协议，一般是用不到的。netty几乎支持了所有的主流协议，如http、ftp等，很多开源的项目也直接使用了netty来开发，如Dubbo、dropwizard等，但是更多的如tomcat、jetty会直接使用socket上搭上http，这样直接针对性的框架。如果对比就是想让你明白，netty是一个通用框架。
-
类似的框架还有MINA。
netty使用体会
拆包和粘包
socket最重要的问题就是粘包和拆包，上一篇已经讲过有三种方法，固定长度、设置边界符、和使用协议。前两种方法都有一定的限制，所以一般是使用协议。对于协议，或者说报文，要有基础知识，一般来说，协议分协议头和协议体，协议头中会存报文长度。
粘包和拆包netty怎么解决呢，提供了解码器，三种解决方法有三种解码器：
| 解码器 | 描述 |
|---|---|
| FixedLengthFrameDecoder | 定长解码器 |
| DelimiterBasedFrameDecoder | 边界符解码器， |
| LineBasedFrameDecoder | 可以认为是边界符解码器的特例，边界符是'\n'或'\r\n' |
| LengthFieldBasedFrameDecoder | 私有协议解码器 |
- FixedLengthFrameDecoder
固定长度解码器，例如收到四个包，| A | BC | DEFG | HI |，将被解码成| ABC | DEF | GHI |三个报文。源码上怎么处理的呢？
protected Object decode(
@SuppressWarnings("UnusedParameters") ChannelHandlerContext ctx, ByteBuf in) throws Exception {
if (in.readableBytes() < frameLength) {
return null;
} else {
return in.readSlice(frameLength).retain();
}
}
当读的长度足够时，就按照长度读取一个报文，处理，否则就返回null。返回null时，pipline（后面会讲）就会跳过职责链后面的handler。继续等待数据流的下次读入。
从代码逻辑上看很简单，这得益于netty的设计，Bytebuf的高效设计、netty职责链的设计。
- LineBasedFrameDecoder
这个很好理解，就是遇到'\n' 、'\r\n'的时候认为是一个报文的结束。
protected Object decode(ChannelHandlerContext ctx, ByteBuf buffer) throws Exception {
final int eol = findEndOfLine(buffer);
if (!discarding) {
if (eol >= 0) {
final ByteBuf frame;
final int length = eol - buffer.readerIndex();
final int delimLength = buffer.getByte(eol) == '\r'? 2 : 1;
if (length > maxLength) {
buffer.readerIndex(eol + delimLength);
fail(ctx, length);
return null;
}
if (stripDelimiter) {
frame = buffer.readSlice(length);
buffer.skipBytes(delimLength);
} else {
frame = buffer.readSlice(length + delimLength);
}
return frame.retain();
} else {
final int length = buffer.readableBytes();
if (length > maxLength) {
discardedBytes = length;
buffer.readerIndex(buffer.writerIndex());
discarding = true;
if (failFast) {
fail(ctx, "over " + discardedBytes);
}
}
return null;
}
} else {
if (eol >= 0) {
final int length = discardedBytes + eol - buffer.readerIndex();
final int delimLength = buffer.getByte(eol) == '\r'? 2 : 1;
buffer.readerIndex(eol + delimLength);
discardedBytes = 0;
discarding = false;
if (!failFast) {
fail(ctx, length);
}
} else {
discardedBytes += buffer.readableBytes();
buffer.readerIndex(buffer.writerIndex());
}
return null;
}
}
private void fail(final ChannelHandlerContext ctx, int length) {
fail(ctx, String.valueOf(length));
}
private void fail(final ChannelHandlerContext ctx, String length) {
ctx.fireExceptionCaught(
new TooLongFrameException(
"frame length (" + length + ") exceeds the allowed maximum (" + maxLength + ')'));
}
/**
* Returns the index in the buffer of the end of line found.
* Returns -1 if no end of line was found in the buffer.
*/
private static int findEndOfLine(final ByteBuf buffer) {
final int n = buffer.writerIndex();
for (int i = buffer.readerIndex(); i < n; i ++) {
final byte b = buffer.getByte(i);
if (b == '\n') {
return i;
} else if (b == '\r' && i < n - 1 && buffer.getByte(i + 1) == '\n') {
return i; // \r\n
}
}
return -1; // Not found.
}
findEndOfLine，没有'\n' '\r\n'，返回-1，此时分为两种情况，一是可读取的数据比maxlength大，此时要进行错误处理，否则，就根据findEndOfLine的结果处理，如果-1，就返回null，等待下一次读入。
- DelimiterBasedFrameDecoder
看下构造函数
public DelimiterBasedFrameDecoder(
int maxFrameLength, boolean stripDelimiter, boolean failFast,
ByteBuf delimiter)
public DelimiterBasedFrameDecoder(
int maxFrameLength, boolean stripDelimiter, boolean failFast, ByteBuf... delimiters)
传入最大值、分隔符、是否跳过分隔符和failFast
protected Object decode(ChannelHandlerContext ctx, ByteBuf buffer) throws Exception {
if (lineBasedDecoder != null) {
return lineBasedDecoder.decode(ctx, buffer);
}
// Try all delimiters and choose the delimiter which yields the shortest frame.
int minFrameLength = Integer.MAX_VALUE;
ByteBuf minDelim = null;
for (ByteBuf delim: delimiters) {
int frameLength = indexOf(buffer, delim);
if (frameLength >= 0 && frameLength < minFrameLength) {
minFrameLength = frameLength;
minDelim = delim;
}
}
if (minDelim != null) {
int minDelimLength = minDelim.capacity();
ByteBuf frame;
if (discardingTooLongFrame) {
// We've just finished discarding a very large frame.
// Go back to the initial state.
discardingTooLongFrame = false;
buffer.skipBytes(minFrameLength + minDelimLength);
int tooLongFrameLength = this.tooLongFrameLength;
this.tooLongFrameLength = 0;
if (!failFast) {
fail(tooLongFrameLength);
}
return null;
}
if (minFrameLength > maxFrameLength) {
// Discard read frame.
buffer.skipBytes(minFrameLength + minDelimLength);
fail(minFrameLength);
return null;
}
if (stripDelimiter) {
frame = buffer.readSlice(minFrameLength);
buffer.skipBytes(minDelimLength);
} else {
frame = buffer.readSlice(minFrameLength + minDelimLength);
}
return frame.retain();
} else {
if (!discardingTooLongFrame) {
if (buffer.readableBytes() > maxFrameLength) {
// Discard the content of the buffer until a delimiter is found.
tooLongFrameLength = buffer.readableBytes();
buffer.skipBytes(buffer.readableBytes());
discardingTooLongFrame = true;
if (failFast) {
fail(tooLongFrameLength);
}
}
} else {
// Still discarding the buffer since a delimiter is not found.
tooLongFrameLength += buffer.readableBytes();
buffer.skipBytes(buffer.readableBytes());
}
return null;
}
}
···
这里逻辑上其实和LineBasedFrameDecoder差不多，而且做了个优化，如果分解符是'\r'或者'\n\n‘，则直接使用LineBasedFrameDecoder
- LengthFieldBasedFrameDecoder
这个就是私有协议的，构造函数有四个参数：
| 参数 | 含义 |
|---|---|
| lenthFieldOffset | offset |
| lenthFieldLength | 长度，一般指body的长度 |
| lenthAdjustment | 调整长度，用于长度字节后有其它字节 |
| initialBytesToStrip | 报文去掉的头部长度 |
看下解码函数
protected Object decode(ChannelHandlerContext ctx, ByteBuf in) throws Exception {
if (discardingTooLongFrame) {
long bytesToDiscard = this.bytesToDiscard;
int localBytesToDiscard = (int) Math.min(bytesToDiscard, in.readableBytes());
in.skipBytes(localBytesToDiscard);
bytesToDiscard -= localBytesToDiscard;
this.bytesToDiscard = bytesToDiscard;
failIfNecessary(false);
}
if (in.readableBytes() < lengthFieldEndOffset) {
return null;
}
int actualLengthFieldOffset = in.readerIndex() + lengthFieldOffset;
long frameLength = getUnadjustedFrameLength(in, actualLengthFieldOffset, lengthFieldLength, byteOrder);
if (frameLength < 0) {
in.skipBytes(lengthFieldEndOffset);
throw new CorruptedFrameException(
"negative pre-adjustment length field: " + frameLength);
}
frameLength += lengthAdjustment + lengthFieldEndOffset;
if (frameLength < lengthFieldEndOffset) {
in.skipBytes(lengthFieldEndOffset);
throw new CorruptedFrameException(
"Adjusted frame length (" + frameLength + ") is less " +
"than lengthFieldEndOffset: " + lengthFieldEndOffset);
}
if (frameLength > maxFrameLength) {
long discard = frameLength - in.readableBytes();
tooLongFrameLength = frameLength;
if (discard < 0) {
// buffer contains more bytes then the frameLength so we can discard all now
in.skipBytes((int) frameLength);
} else {
// Enter the discard mode and discard everything received so far.
discardingTooLongFrame = true;
bytesToDiscard = discard;
in.skipBytes(in.readableBytes());
}
failIfNecessary(true);
return null;
}
// never overflows because it's less than maxFrameLength
int frameLengthInt = (int) frameLength;
if (in.readableBytes() < frameLengthInt) {
return null;
}
if (initialBytesToStrip > frameLengthInt) {
in.skipBytes(frameLengthInt);
throw new CorruptedFrameException(
"Adjusted frame length (" + frameLength + ") is less " +
"than initialBytesToStrip: " + initialBytesToStrip);
}
in.skipBytes(initialBytesToStrip);
// extract frame
int readerIndex = in.readerIndex();
int actualFrameLength = frameLengthInt - initialBytesToStrip;
ByteBuf frame = extractFrame(ctx, in, readerIndex, actualFrameLength);
in.readerIndex(readerIndex + actualFrameLength);
return frame;
}
算法逻辑相当复杂
ChannelPipeLine
* +---------------------------------------------------+---------------+
* | ChannelPipeline | |
* | \|/ |
* | +---------------------+ +-----------+----------+ |
* | | Inbound Handler N | | Outbound Handler 1 | |
* | +----------+----------+ +-----------+----------+ |
* | /|\ | |
* | | \|/ |
* | +----------+----------+ +-----------+----------+ |
* | | Inbound Handler N-1 | | Outbound Handler 2 | |
* | +----------+----------+ +-----------+----------+ |
* | /|\ . |
* | . . |
* | ChannelHandlerContext.fireIN_EVT() ChannelHandlerContext.OUT_EVT()|
* | [ method call] [method call] |
* | . . |
* | . \|/ |
* | +----------+----------+ +-----------+----------+ |
* | | Inbound Handler 2 | | Outbound Handler M-1 | |
* | +----------+----------+ +-----------+----------+ |
* | /|\ | |
* | | \|/ |
* | +----------+----------+ +-----------+----------+ |
* | | Inbound Handler 1 | | Outbound Handler M | |
* | +----------+----------+ +-----------+----------+ |
* | /|\ | |
* +---------------+-----------------------------------+---------------+
* | \|/
* +---------------+-----------------------------------+---------------+
* | | | |
* | [ Socket.read() ] [ Socket.write() ] |
* | |
* | Netty Internal I/O Threads (Transport Implementation) |
* +-------------------------------------------------------------------+
PipeLine是设计的关键，这是基于责任链去扩展，如图是源码注释给出的解释。这里责任链会分为两个方向，一个是Inbound，一个是OutBound，Inbound由socket本身出发，读取到数据出发，OutBound由Write动作触发，最终到socket。
netty职责链交互简要图片
如上图，编解码是非常重要的一环，通过解码反序列化为JAVA Object（msg），然后在servicehandler中处理，随后把msg交给encoder处理，encoder序列化为二进制流。
这个过程有两个关键点：
- decoder肯定是除了默认的handler外第一个手写handler，，
后面的servicehandler是若干个，但由于是责任链方式，所以每个消息来了，需要判断一下是否是本业务的servicehandler需要处理的，如果不是，就需要用fiireRead往下传递消息，否则消息就在此handler终结。 - encoder则是OutBound方向最后一个手写的handler，但是write搜索逻辑是搜索outbound类型的handler，搜索的方向是向职责链的前方搜索，搜索至第一个outbound类型。
所以在BootStrap中addLast的时候，编解码需要在业务之前add，否则在write的时候就找不到encoder。
架构设计
私有协议和架构设计是更高一层次需要积累的知识，计划下周再写一篇文章。
总的来说，我个人认为基础框架中，netty虽然基本用不到，但是是必学框架，因为对于分布式服务框架的理解非常有帮助。
