Netty编解码器&TCP粘包拆包

　　　　在?络应?中需要实现某种编解码器，将原始字节数据与?定义的消息对象进?互相转换。?络中都是以字节码的数据形式来传输数据的，服务器编码数据后发送到客户端，客户端需要对数据进?解码。

　　　　netty提供了强?的编解码器框架，使得我们编写?定义的编解码器很容易，也容易封装重?。同时Netty 的编（解）码器实现了 ChannelHandlerAdapter，也是?种特殊的 ChannelHandler，所以依赖于 ChannelPipeline，可以将多个编（解）码器链接在?起，以实现复杂的转换逻辑。对于Netty??，编解码器由两部分组成：编码器、解码器。

　　　　　　解码器：负责将消息从字节或其他序列形式转成指定的消息对象。负责处理?站 InboundHandler数据；

　　　　　　编码器：将消息对象转成字节或其他序列形式在?络上传输。负责处理出站 OutboundHandler”数据；

　　　不论解码器 handler 还是编码器 handler ，接收的消息类型必须与待处理的消息类型?致，否则该 handler 不会被执?。Netty提供了很多编解码器，例如：

　　　　StringEncoder字符串编码器

　　　　StringDecoder字符串解码器

　　　　ObjectEncoder 对象编码器

　　　　ObjectDecoder 对象解码器

　　　　FixedLengthFrameDecoder 固定?度的解码器

　　　　LineBasedFrameDecoder 以换?符为结束标识的解码器

　　　　DelimiterBasedFrameDecoder 指定消息分隔符的解码器

　　　　LengthFieldBasedFrameDecoder基于?度通?解码器

    protected abstract void decode(ChannelHandlerContext var1, ByteBuf var2, List<Object> var3) throws Exception;
final void decodeRemovalReentryProtection(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
　　　　 ......
this.decode(ctx, in, out);
......
}
protected void decodeLast(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
if (in.isReadable()) {
this.decodeRemovalReentryProtection(ctx, in, out);
}
}

public class ToIntegerDecoder extends ByteToMessageDecoder {
　　@Override
　　public void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
　　　　if (in.readableBytes() >= 4) {
　　　　　　out.add(in.readInt());
　　　　}
　　}
}

/**
* Integer解码器,ReplayingDecoder实现
*/
public class ToIntegerReplayingDecoder extends ReplayingDecoder<Void> {
　　@Override
　　protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
　　　　System.out.println("ToIntegerReplayingDecoder 被调?");
　　　　//在 ReplayingDecoder 不需要判断数据是否?够读取，内部会进?处理判断
　　　　out.add(in.readInt());
　　}
}

/**
* 其中泛型参数I表示我们要解码的消息类型。例前?，我们在ToIntegerDecoder中，把?进制字节
流转换成了?个int类型的整数。
*/
public abstract class MessageToMessageDecoder<I> extends ChannelInboundHandlerAdapter

/**
* 参数msg，需要进?解码的参数。例如ByteToMessageDecoder解码后的得到的包含完整报?信息
ByteBuf
* List<Object> out参数：将msg经过解析后得到的java对象，添加到放到List<Object> out中
*/
protected abstract void decode(ChannelHandlerContext ctx, I msg, List<Object> out) throws Exception;

public class IntegerToStringDecoder extends MessageToMessageDecoder<Integer> {
　　@Override
　　public void decode(ChannelHandlerContext ctx, Integer msg, List<Object> out) throws Exception {
　　　　out.add(String.valueOf(msg));
　　}
}

/**
* todo ?定义解码器
*/
public class ByteToLongDecoder extends ByteToMessageDecoder {
　　@Override
　　protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
　　　　System.out.println("ByteToLongDecoder decode 被调?");
　　　　//todo 因为long占8个字节, 需要判断?于等于8个字节时才能读取?个long
　　　　if(in.readableBytes() >= 8) {
　　　　　　out.add(in.readLong());
　　　　}
　　}
}

public abstract class MessageToByteEncoder<I> extends ChannelOutboundHandlerAdapter {
　　....
　　protected abstract void encode(ChannelHandlerContext ctx, I msg, ByteBuf out) throws Exception;
}

　　2、抽象类MessageToMessageEncoder

　　　　MessageToMessageEncoder同样是?个泛型类，泛型参数I表示将需要编码的对象的类型，编码的结果是将信息放到?个List中。?类通过覆写其抽象?法encode，来实现编码，如下所示：

public abstract class MessageToMessageEncoder<I> extends ChannelOutboundHandlerAdapter {
　　...
　　protected abstract void encode(ChannelHandlerContext ctx, I msg, List<Object> out) throws Exception;
　　...
}

 　　　　与MessageToByteEncoder不同的，MessageToMessageEncoder编码后的结果放到的out参数类型是?个List中。例如，你?次发送2个报?，因此msg参数中实际上包含了2个报?，因此应该解码出两个报?对象放到List中。

　　　　MessageToMessageEncoder提供的常??类包括：

　　　　　　LineEncoder：按?编码，给定?个CharSequence(如String)，在其之后添加换?符\n或者\r\n，并封装到ByteBuf进?输出，与LineBasedFrameDecoder相对应。

　　　　　　Base64Encoder：给定?个ByteBuf，得到对其包含的?进制数据进?Base64编码后的新的ByteBuf进?输出，与Base64Decoder相对应。

　　　　　　LengthFieldPrepender：给定?个ByteBuf，为其添加报?头Length字段，得到?个新的ByteBuf进?输出。Length字段表示报??度，与LengthFieldBasedFrameDecoder相对应。

　　　　　　StringEncoder：给定?个CharSequence(如：StringBuilder、StringBuffer、String等)，将其转换成ByteBuf进?输出，与StringDecoder对应。

　　　　这些MessageToMessageEncoder实现类最终输出的都是ByteBuf，因为最终在?络上传输的都要是?进制数据。

　　3、?定义编码器

　　　　通过继承MessageToByteEncoder?定义编码器

public class LongToByteEncoder extends MessageToByteEncoder<Long> {
　　@Override
　　protected void encode(ChannelHandlerContext ctx, Long msg, ByteBuf out) throws Exception {
　　　　System.out.println("LongToByteEncoder encode被调?");
　　　　System.out.println("msg=" + msg);
　　　　out.writeLong(msg);
　　}
}

　　编码解码器同时具有编码与解码功能，特点是同时实现了ChannelInboundHandler和ChannelOutboundHandler接?，因此在数据输?和输出时都能进?处理。

　　Netty提供提供了?个ChannelDuplexHandler适配器类，编码解码器的抽象基类ByteToMessageCodec 、MessageToMessageCodec都继承与此类。

　　ByteToMessageCodec内部维护了?个ByteToMessageDecoder和?个MessageToByteEncoder实例，可以认为是?者的功集合，泛型参数I是接受的编码类型：

public abstract class ByteToMessageCodec<I> extends ChannelDuplexHandler {
　　private final TypeParameterMatcher outboundMsgMatcher;
　　private final MessageToByteEncoder<I> encoder;
　　private final ByteToMessageDecoder decoder = new ByteToMessageDecoder(){…}
　　...
　　protected abstract void encode(ChannelHandlerContext ctx, I msg, ByteBuf out) throws Exception;
　　protected abstract void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception;
　　...
}

public abstract class MessageToMessageCodec<INBOUND_IN, OUTBOUND_IN> extends ChannelDuplexHandler {
　　private final MessageToMessageEncoder<Object> encoder= ...
　　private final MessageToMessageDecoder<Object> decoder =…
　　...
　　protected abstract void encode(ChannelHandlerContext ctx, OUTBOUND_IN msg, List<Object> out) throws Exception;
　　protected abstract void decode(ChannelHandlerContext ctx, INBOUND_IN msg, List<Object> out) throws Exception;
}

/**
* 解码器，将byte转成char
*/
public class ByteToCharDecoder extends ByteToMessageDecoder {
　　@Override
　　protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
　　　　while(in.readableBytes() >= 2){
　　　　out.add(Character.valueOf(in.readChar()));
　　}
}

/**
* 编码器，将char转成byte
*/
public class CharToByteEncoder extends MessageToByteEncoder<Character> {
　　@Override
　　protected void encode(ChannelHandlerContext ctx, Character msg, ByteBuf out) throws Exception {
　　　　out.writeChar(msg);
　　}
}

/**
* 继承CombinedChannelDuplexHandler，?于绑定解码器和编码器
*/
public class CharCodec extends CombinedChannelDuplexHandler<ByteToCharDecoder, CharToByteEncoder> {
　　public CharCodec(){
　　　　super(new ByteToCharDecoder(), new CharToByteEncoder());
　　}
}

二、TCP粘包和拆包

（一）TCP粘包和拆包基本介绍

　　TCP是个流协议，所谓流，就是没有界限的?串数据。TCP底层并不了解上层业务数据的具体含义，它会根据TCP缓冲区的实际情况进?包的划分，所以在业务上认为，?个完整的包可能会被TCP拆分成多个包进?发送，也有可能把多个?的包封装成?个?的数据包发送，这就是所谓的TCP粘包和拆包问题。

　　TCP粘包：把多个?的包封装成?个?的数据包发送，发送?发送的若?数据包到接收?时粘成?个包

　　TCP拆包：把?个完整的包拆分为多个?包进?发送，发送?发送?个数据包到接收?时被拆分为若?个?包

　　如上图所示，就是粘包拆包的各个场景演示。假设客户端分别发送了两个数据包 D1 和 D2 给服务端，由于服务端?次读取到字节数是不确定的，故可能存在以下四种情况：

　　　　1. 服务端分两次读取到了两个独?的数据包，分别是 D1 和 D2 ，没有粘包和拆包

　　　　2. 服务端?次接受到了两个数据包， D1 和 D2 粘合在?起，称之为 TCP 粘包

　　　　3. 服务端分两次读取到了数据包，第?次读取到了完整的 D1 包和 D2 包的部分内容，第?次读取到了 D2 包的剩余内容，这称之为 TCP 拆包

　　　　4. 服务端分两次读取到了数据包，第?次读取到了 D1 包的部分内容 D1_1 ，第?次读取到了 D1包的剩余部分内容 D1_2 和完整的 D2 包。

（二）TCP粘包和拆包产生原因

　　发?TCP粘包、拆包主要是由于下??些原因：

　　　　应?程序写?的数据?于缓冲区??，这将会发?拆包。

　　　　应?程序写?数据?于字缓冲区??，?卡将应?多次写?的数据发送到?络上，这将会发?粘包。

　　　　进?MSS（最?报??度）??的TCP分段，当TCP报??度-TCP头部?度>MSS的时候将发?拆包。

　　　　接收?法不及时读取套接字缓冲区数据，这将发?粘包。

　　MSS: 是Maximum Segement Size缩写，表示TCP报?中data部分的最??度，是TCP协议在OSI五层?络模型中传输层对?次可以发送的最?数据的限制。

　　MTU: 最?传输单元，是Maxitum Transmission Unit的简写，是OSI五层?络模型中链路层(datalink layer)对?次可以发送的最?数据的限制。

　　当需要传输的数据?于MSS或者MTU时，数据会被拆分成多个包进?传输。由于MSS是根据MTU计算出来的，因此当发送的数据满?MSS时，必然满?MTU。

　　发送端的字节流都会先传?缓冲区，再通过?络传?到接收端的缓冲区中，最终由接收端获取。当我们发送两个完整包到接收端的时候，正常情况会接收到两个完整的报?。但也有可能接收到的是?个报?，它是由发送的两个报?组成的，这样对于应?程序来说就很难处理了（这样称为粘包）；还有可能出现上?这样的虽然收到了两个包，但是??的内容却是互相包含，对于应?来说依然?法解析（拆包）。

（三）解决?案

　　拆包解决思路：

　　　　基本思路就是不断的从TCP缓冲区中读取数据，每次读取完都需要判断是否是?个完整的数据包。

　　　　　　若当前读取的数据不?以拼接成?个完整的业务数据包，那就保留该数据，继续从tcp缓冲区中读取，直到得到?个完整的数据包

　　　　　　若当前读到的数据加上已经读取的数据?够拼接成?个数据包，那就将已经读取的数据拼接上本次读取的数据，构成?个完整的业务数据包传递到业务逻辑，多余的数据仍然保留，以便和下次读到的数据尝试拼接

　　关键点是如何判断是?个完整的数据包，解决策略：

　　　　（1）设置消息边界(分隔符，对应Netty提供的LineBasedFrameDecoder、DelimiterBasedFrameDecoder解码器)，例如换行符或者自定义的分隔符。

　　　　（2）设置定?消息(对应Netty提供的FixedLengthFrameDecoder解码器)，例如对Integer指定4个字节。

　　　　（3）使?带消息头的协议，消息头存储消息开始标识及消息的?度信息Header+Body(对应Netty提供的LengthFieldBasedFrameDecoder解码器)，例如http形式的处理，使用报文头中的长度截取报文体。

　　　　（4）发送消息?度，?定义消息解码器

　　1、LineBasedFrameDecoder

　　　　LineBasedFrameDecoder是回?换?解码器，如果?户发送的消息以回?换?符作为消息结束的标识，则可以直接使?Netty的LineBasedFrameDecoder对消息进?解码，只需要在初始化Netty服务端或者客户端时将LineBasedFrameDecoder正确的添加到ChannelPipeline中即可，不需要??重新实现?套换?解码器。

　　　　LineBasedFrameDecoder的?作原理是它依次遍历ByteBuf中的可读字节，判断是否有“\n”或“\r\n”，如果有，就以此位置为结束位置，从可读索引到结束位置区间的字节就组成了??。它是以换?符为结束标志的解码器，?持携带结束符或不携带结束符两种解码?式，同时?持配置单?的最??度。如果连接读取到最??度后仍然没有发现换?符，就会抛出异常，同时忽略掉之前读到的异常码流。防?由于数据报没有携带换?符导致接收到 ByteBuf ?限制积压，引起系统内存溢出。

　　　　通常LineBasedFrameDecoder会和StringDecoder搭配使?。StringDecoder的功能?常简单，就是将接收到的对象转换成字符串，然后继续调?后?的Handler。LineBasedFrameDecoder+StringDecoder组合就是按?切换的?本解码器，?来?持TCP的粘包和拆包。

　　　　对于?本类协议的解析，?本换?解码器?常实?，例如对 HTTP 消息头的解析、FTP 协议消息的解析等。

　　　　LineBasedFrameDecoder使?起来?分简单，只需要在ChannelPipeline 中添加即可，如下所示

ServerBootstrap b = new ServerBootstrap();
　　b.group(bossGroup, workerGroup)
　　　　.channel(NioServerSocketChannel.class)
　　　　.option(ChannelOption.SO_BACKLOG, 1024)
　　　　.childHandler(new ChannelInitializer<SocketChannel>() {
　　@Override
　　public void initChannel(SocketChannel ch) throws Exception {
　　　　ChannelPipeline p = ch.pipeline();
　　　　p.addLast(new LineBasedFrameDecoder(1024));
　　　　p.addLast(new StringDecoder());
　　　　p.addLast(new StringEncoder());
　　　　p.addLast(new LineServerHandler());
　　}
});

　　2、DelimiterBasedFrameDecoder

　　　　DelimiterBasedFrameDecoder是分隔符解码器，?户可以指定消息结束的分隔符，它可以?动完成以分隔符作为码流结束标识的消息的解码。回?换?解码器实际上是?种特殊的DelimiterBasedFrameDecoder解码器。

　　　　?先将分隔符转换成 ByteBuf 对象，作为参数构造 DelimiterBasedFrameDecoder，将其添加到ChannelPipeline 中，然后依次添加字符串解码器（通常?于?本解码）和?户 Handler。

　　　　DelimiterBasedFrameDecoder 原理分析：解码时，判断当前已经读取的 ByteBuf 中是否包含分隔符 ByteBuf，如果包含，则截取对应的 ByteBuf 返回。 

　　　　示例：发消息时自定义使用&_作为分隔符，解码时也使用其作为分隔符。

// 客户端发送数据
public class ClientHandler extends ChannelInboundHandlerAdapter {
　　/**
　　* 当客户端连接服务器完成就会触发该?法
　　* @param ctx
　　* @throws Exception
　　*/
　　@Override
　　public void channelActive(ChannelHandlerContext ctx) throws Exception {
　　　　String message = "aaaaaaaaaaaaaaaa&_bbbbbbbbbbbbbbbbbb&_ccccccccccc&_";
　　　　ByteBuf byteBuf = Unpooled.buffer(message.getBytes().length);
　　　　byteBuf.writeBytes(message.getBytes());
　　　　ctx.writeAndFlush(byteBuf);
}
}  

// 服务端添加分隔符解码器
public class EchoServer {
　　public static void main(String[] args) {
　　　　// 设置两个线程组
　　　　serverBootstrap.group(bossGroup,workerGroup)
　　　　　　.channel(NioServerSocketChannel.class)
　　　　　　.option(ChannelOption.SO_BACKLOG,1024)
　　　　　　.childHandler(new ChannelInitializer<SocketChannel>() {
　　　　　　　　　　@Override
　　　　　　　　　　protected void initChannel(SocketChannel ch) throws Exception {
　　　　　　　　　　　　// 向pipeline加?分隔符解码器
　　　　　　　　　　　　ByteBuf delimiter = Unpooled.copiedBuffer("&_".getBytes());
　　　　　　　　　　　　ch.pipeline().addLast(new DelimiterBasedFrameDecoder(1024,true,delimiter));
　　　　　　　　　　　　ch.pipeline().addLast(new StringDecoder());
　　　　　　　　　　　　ch.pipeline().addLast(new ServerHandler());
　　　　　　　　　　}
　　　　　　});
　　}
}

　　3、FixedLengthFrameDecoder

　　　　FixedLengthFrameDecoder是固定?度解码器，它能够按照指定的?度对消息进??动解码，开发者不需要考虑TCP的粘包/拆包等问题，?常实?。对于定?消息，如果消息实际?度?于定?，则往往会进?补位操作，它在?定程度上导致了空间和资源的浪费。但是它的优点也是?常明显的，编解码?较简单。

ServerBootstrap b = new ServerBootstrap();
b.group(bossGroup, workerGroup)
　　.channel(NioServerSocketChannel.class)
　　.option(ChannelOption.SO_BACKLOG, 100)
　　.handler(new LoggingHandler(LogLevel.INFO))//配置?志输出
　　.childHandler(new ChannelInitializer<SocketChannel>() {
　　　　　　@Override
　　　　　　protected void initChannel(SocketChannel ch) throws Exception {
　　　　　　　　ch.pipeline().addLast(new FixedLengthFrameDecoder(10));
　　　　　　　　ch.pipeline().addLast(new StringDecoder());
　　　　　　　　ch.pipeline().addLast(new StringEncoder());
　　　　　　　　ch.pipeline().addLast(new ServerHandler());
　　　　　　}
　　 });

public LengthFieldBasedFrameDecoder(int maxFrameLength, int lengthFieldOffset, int lengthFieldLength, int lengthAdjustment, int initialBytesToStrip) {
　　this(maxFrameLength, lengthFieldOffset, lengthFieldLength, lengthAdjustment, initialBytesToStrip, true);
}

* <pre>
* lengthFieldOffset = 0
* lengthFieldLength = 2
* lengthAdjustment = 0
* <b>initialBytesToStrip</b> = <b>2</b> (= the length of the Length field)
* BEFORE DECODE (14 bytes) 　　　　　AFTER DECODE (12 bytes)
* +--------+----------------+ 　　　+----------------+
* | Length | Actual Content |----->| Actual Content |
* | 0x000C | "HELLO, WORLD" | 　　　| "HELLO, WORLD" |
* +--------+----------------+ +----------------+
* </pre>
lengthFieldOffset = 0，?度字段偏移位置为0表示从包的第?个字节开始读取；
lengthFieldLength = 2，?度字段?为2，从包的开始位置往后2个字节的?度为?度字段；
lengthAdjustment = 0 ，解析的时候?需跳过任何?度；
initialBytesToStrip = 2，去掉当前数据包的开头2字节，去掉 header。
0x000C 转为 int = 12。　　

//协议包
public class MessageProtocol {
　　private int len; //关键
　　private byte[] content;
　　public int getLen() {
　　　　return len;
　}
　　public void setLen(int len) {
　　　　this.len = len;
　　}
　　public byte[] getContent() {
　　　　return content;
　}
　　public void setContent(byte[] content) {
　　　　this.content = content;
　}
}

public class MyProtocolServer {
　　private int port;
　　public MyProtocolServer(int port) {
　　　　this.port = port;
　}
　　public void start(){
　　　　EventLoopGroup bossGroup = new NioEventLoopGroup();
　　　　EventLoopGroup workGroup = new NioEventLoopGroup();
　　　　ServerBootstrap server = new ServerBootstrap()
　　　　　　.group(bossGroup,workGroup)
　　　　　.channel(NioServerSocketChannel.class)
　　　　　.childHandler(new ServerChannelInitializer());
　　　　try {
　　　　　　ChannelFuture future = server.bind(port).sync();
　　　　　　future.channel().closeFuture().sync();
　　　　} catch (InterruptedException e) {
　　　　　　e.printStackTrace();
　　　}finally {
　　　　　　bossGroup.shutdownGracefully();
　　　　　　workGroup.shutdownGracefully();
　　　}
　}
　　public static void main(String[] args) {
　　　　MyProtocolServer server = new MyProtocolServer(7788);
　　　　server.start();
　}
} 

public class MyProtocolClient {
　　private int port;
　　private String address;
　　public MyProtocolClient(int port, String address) {
　　　　this.port = port;
　　　　this.address = address;
　}
　　public void start(){
　　　　EventLoopGroup group = new NioEventLoopGroup();
　　　　Bootstrap bootstrap = new Bootstrap();
　　　　bootstrap.group(group)
　　　　　　.channel(NioSocketChannel.class)
　　　　　　.option(ChannelOption.TCP_NODELAY, true)
　　　　　　.handler(new ClientChannelInitializer());
　　　　try {
　　　　　　ChannelFuture future = bootstrap.connect(address,port).sync();
　　　　　　future.channel().writeAndFlush("Hello world, i‘m online");
　　　　　　future.channel().closeFuture().sync();
　　　} catch (Exception e) {
　　　　　　e.printStackTrace();
　　　}finally {
　　　　　　group.shutdownGracefully();
　　　}
　}
　　public static void main(String[] args) {
　　　　MyProtocolClient client = new MyProtocolClient(7788,"127.0.0.1");
　　　　client.start();
　}
}