? ?在我们了解了netty的基本组件之后,我们来了解netty一下使用netty带来的问题以及如何解决.
?1.Netty编解码
1.1? Netty涉及到编解码的组件有Channel、ChannelHandler、ChannelPipe,我们再来复习一下
一个 Channel 包含了一个 ChannelPipeline，而 ChannelPipeline 中又维护了一个由 ChannelHandlerContext 组成的双向链表，并且每个 ChannelHandlerContext 中又关联着一个 ChannelHandler,我们业务逻辑写在ChannelHandler里面.
read事件(入站事件)和write事件(出站事件)在一个双向链表中，入站事件会从链表 head 往后传递到最后一个入站的 handler，出站事件会从链表 tail 往前传递到最前一个出站的 handler，两种类型的 handler 互不干扰.
1.2? 为什么我们需要编解码器?
? ? ? ?首先接单介绍一下,序列化和反序列化大概念,在网络传输通信中,会发生两种数据转换的操作,一种是把消息对象转换成字节码,这种是序列化,还有一种是要把字节码对象再转换成消息对象,称为反序列化.和netty的对应关系,序列化对应的是编码过程,反序列化对应的解码过程.业界里面也有其他编码框架： google的 protobuf（PB)、Facebook的Trift、Jboss的Marshalling、Kyro等.
? ? ? 个人理解就是网络中传输都得是二进制0,1,所以我们写到网络中需要把我的字符串或者对象,转变为字节码,这叫编码.然后从网络中把字节码读取出来再解码成我们对应的字符串或者对象.
? ? ? 如果要实现高效的编解码可以用protobuf，但是protobuf需要维护大量的proto文件比较麻烦，现在一般可以使用protostuff。protostuff是一个基于protobuf实现的序列化方法，它较于protobuf最明显的好处是，在几乎不损耗性能的情况下做到了不用我们写.proto文件来实现序列化。使用它也非常简单，这里不做过多介绍.
1.3??为啥jdk有编解码，还要netty自己开发编解码？
? ? ? 1）无法跨语言
? ? ? 2) 序列化后的码流太大，也就是数据包太大
? ? ? 3) 序列化和反序列化性能比较差
1.4? Netty里面的编解码：
? ??解码器：负责处理“入站 InboundHandler”数据
? ? 编码器：负责“出站 OutboundHandler” 数据
? ? Netty里面提供默认的编解码器，也支持自定义编解码器
? ? ? ? ?Encoder:编码器
? ? ? ? ?Decoder:解码器
1.5??Netty的解码器Decoder和使用场景
-
Decoder对应的就是ChannelInboundHandler，主要就是字节数组转换为消息对象
-
主要是两个方法 decode decodeLast
-
抽象解码器
- ByteToMessageDecoder用于将字节转为消息，需要检查缓冲区是否有足够的字节
- ReplayingDecoder继承ByteToMessageDecoder，不需要检查缓冲区是否有足够的字节，但是ReplayingDecoder速度略满于ByteToMessageDecoder，不是所有的ByteBuf都支持
- 选择：项目复杂性高则使用ReplayingDecoder，否则使用 ByteToMessageDecoder
- MessageToMessageDecoder用于从一种消息解码为另外一种消息（例如POJO到POJO）
-
解码器具体的实现，用的比较多的是(更多是为了解决TCP底层的粘包和拆包问题)
- DelimiterBasedFrameDecoder： 指定消息分隔符的解码器
- LineBasedFrameDecoder: 以换行符为结束标志的解码器
- FixedLengthFrameDecoder：固定长度解码器
- LengthFieldBasedFrameDecoder：message = header+body, 基于长度解码的通用解码器
- StringDecoder：文本解码器，将接收到的字节码转化为字符串，一般会与上面的进行配合，然后在后面添加业务handle
1.6??Netty编码器Encoder
- Encoder对应的就是ChannelOutboundHandler，消息对象转换为字节数组
- Netty本身未提供和解码一样的编码器，是因为场景不同，两者非对等的
- MessageToByteEncoder消息转为字节数组,调用write方法，会先判断当前编码器是否支持需要发送的消息类型，如果不支持，则透传；
- MessageToMessageEncoder用于从一种消息编码为另外一种消息（例如POJO到POJO）
- StringEncoder：文本编码器，将我们的字符串编码为字节码.
1.7? 代码演示
下面我们加的
1.StringDecoder为解码器(其实就是入站Handler,继承了ChannelInboundHandler),当我的服务端有消息流入的时候会经过我们入站Handler,这个Handler就是为了把字节码转换为我们的String,并将已经解码的字节转发给ChannelPipeline中的下一个ChannelInboundHandler.
2.StringEncoder为编码器(其实就是出站Handler,继承了ChannelOutboundHandler),当我们服务端写出消息给客户端的时候会经过我们的出站Handler,将我们的字符串编码为字节码.
3.ChatServerHandler,我们自定义的入站Handler,StringDecoder把字节码处理完就给到我们的这个Handler去处理业务.
public static void main(String[] args) throws Exception {
EventLoopGroup bossGroup = new NioEventLoopGroup(1);
EventLoopGroup workerGroup = new NioEventLoopGroup(8);
try {
ServerBootstrap bootstrap = new ServerBootstrap();
bootstrap.group(bossGroup, workerGroup)
.channel(NioServerSocketChannel.class)
.option(ChannelOption.SO_BACKLOG, 1024)
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
//加入特殊分隔符分包解码器
//pipeline.addLast(new DelimiterBasedFrameDecoder(1024, Unpooled.copiedBuffer("_"
// .getBytes())));
//向pipeline加入解码器
pipeline.addLast("decoder", new StringDecoder());
//向pipeline加入编码器
pipeline.addLast("encoder", new StringEncoder());
//加入自己的业务处理handler
pipeline.addLast(new ChatServerHandler());
}
});
System.out.println("聊天室server启动。。");
ChannelFuture channelFuture = bootstrap.bind(9000).sync();
//关闭通道
channelFuture.channel().closeFuture().sync();
} finally {
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
}
}2.TCP粘包,拆包
? ? ?我们常常说TCP粘包,拆包,其实这个概念是伪科学.
2.1演示
? 首先客户端发送100000次"你好,世界,我是Netty_",服务端仅仅是输出message
客户端代码
ChannelFuture channelFuture = bootstrap.connect("127.0.0.1", 9000).sync();
//得到 channel
Channel channel = channelFuture.channel();
for (int i = 0; i < 100000; i++) {
channel.writeAndFlush("你好,世界,我是Netty_");
}
服务端Handler
@Override
protected void channelRead0(ChannelHandlerContext ctx, String msg) {
System.out.println(msg);
}
下面是我们服务端获取到的结果:
?
? ? ?我们可以发现,服务端获取到的结果红标4是两条'你好,世界,我是Netty_'合并在一起,是因为发生了粘包,而上面第2行世界的'界'乱码了是因为UTF-8的中文是3个字节,因为发生了拆包,3个字节被拆分了两部分,所以我们看到红标2的结尾和红标3的开头是乱码.不仅如此,红标2也发生了粘包问题.
??
2.2粘包拆包的原因,以及为什么他是伪科学
? ? ? TCP是一个流协议，就是没有界限的一长串二进制数据。TCP作为传输层协议并不不了解上层业务数据的具体含义，它会根据TCP缓冲区的实际情况进行数据包的划分，所以在业务上认为是一个完整的包，可能会被TCP拆分成多个包进行发送，也有可能把多个小的包封装成一个大的数据包发送，这就是所谓的TCP粘包和拆包问题。面向流的通信是无消息保护边界的。
1）TCP拆包: 一个完整的包可能会被TCP拆分为多个包进行发送
2）TCP粘包: 把多个小的包封装成一个大的数据包发送, client发送的若干数据包 Server接收时粘成一包 ? ?
发送方和接收方都可能出现这个原因 ? ? ? ?
发送方的原因：TCP是流协议,TCP默认会使用Nagle算法?(Nagle算法就是为了尽可能发送大块数据，避免网络中充斥着许多小数据块,当你连续发送小数据块的时候,他会收集到一定的大小才发出去)
接收方的原因: TCP接收到数据放置缓存中，应用程序从缓存中读取?
3）所以这并不是TCP的锅,TCP只需要完整和有顺序的把要发的数据发给服务端就好了,你服务端怎么接收也不关我的事情,TCP粘包拆包说得像TCP的BUG一样.
4）UDP: 是没有粘包和拆包的问题，有边界协议
如下图所示，client发了两个数据包D1和D2，但是server端可能会收到如下几种情况的数据。
2.3 如何解决粘包,拆包问题(也叫半包读写)
? ? ? 解决粘包和拆包问题,得回到我们服务端,因为只有我们上层的服务端才准确知道发送数据的含义,以及如何分割包
解决方案 :
1）消息定长度，传输的数据大小固定长度，例如每段的长度固定为100字节，如果不够空位补空格
2）在数据包尾部添加特殊分隔符，比如下划线，中划线等，这种方法简单易行，但选择分隔符的时候一定要注意每条数据的内部一定不能出现分隔符。
3）发送长度：发送每条数据的时候，将数据的长度一并发送，比如可以选择每条数据的前4位是数据的长度，应用层处理时可以根据长度来判断每条数据的开始和结束。 Netty提供了多个解码器，可以进行分包的操作，如下：
LineBasedFrameDecoder （回车换行分包）
DelimiterBasedFrameDecoder（特殊分隔符分包）
FixedLengthFrameDecoder（固定长度报文来分包）
这些分包解码器只需要放在我们的字符串解码器之前就好了,这些分包解码器也是入站Handler,继承了我们的InboundHandler
?
DelimiterBasedFrameDecoder解决TCP半包读写问题
构造函数
public DelimiterBasedFrameDecoder(
int maxFrameLength, boolean stripDelimiter, boolean failFast,
ByteBuf delimiter)
- maxLength：表示一行最大的长度，如果超过这个长度依然没有检测自定义分隔符，将会抛出TooLongFrameException
- failFast：如果为true，则超出maxLength后立即抛出TooLongFrameException，不进行继续解码.如果为false，则等到完整的消息被解码后，再抛出TooLongFrameException异常
- stripDelimiter：解码后的消息是否去除掉分隔符
- delimiters：分隔符,ByteBuf类型
使用方式: 下面代码增加了DelimiterBasedFrameDecoder,特殊分隔符,用下划线'_'分割每一条消息
服务端
public static void main(String[] args) throws Exception {
EventLoopGroup bossGroup = new NioEventLoopGroup(1);
EventLoopGroup workerGroup = new NioEventLoopGroup(8);
try {
ServerBootstrap bootstrap = new ServerBootstrap();
bootstrap.group(bossGroup, workerGroup)
.channel(NioServerSocketChannel.class)
.option(ChannelOption.SO_BACKLOG, 1024)
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
//加入特殊分隔符分包解码器
pipeline.addLast(new DelimiterBasedFrameDecoder(1024, Unpooled.copiedBuffer("_"
.getBytes())));
//向pipeline加入解码器
pipeline.addLast("decoder", new StringDecoder());
//向pipeline加入编码器
pipeline.addLast("encoder", new StringEncoder());
//加入自己的业务处理handler
pipeline.addLast(new ChatServerHandler());
}
});
System.out.println("聊天室server启动。。");
ChannelFuture channelFuture = bootstrap.bind(9000).sync();
//关闭通道
channelFuture.channel().closeFuture().sync();
} finally {
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
}2.4?自定义解码器
? ? ?可以看到我们自定义了一个协议包,包含两个属性,第一个是len代表消息体长度,第二个是contetn代码内容,我们的解码器每一次读4个字节,如果是4个字节就继续往下的逻辑,不够4个字节,就return,为什么是4个字节?因为我们int就是4个字节,代表了消息体长度.
/**
* 自定义协议包
*/
@Data
public class MyMessageProtocol {
//定义一次发送包体长度
private int len;
//一次发送包体内容
private byte[] content;
}
public class MyMessageDecoder extends ByteToMessageDecoder {
int length = 0;
@Override
protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
System.out.println();
System.out.println("MyMessageDecoder decode 被调用");
//需要将得到二进制字节码-> MyMessageProtocol 数据包(对象)
System.out.println(in);
if(in.readableBytes() >= 4) {
if (length == 0){
length = in.readInt(); //这里readInt 代表读取int,4个字节
}
if (in.readableBytes() < length) {
System.out.println("当前可读数据不够，继续等待。。");
return;
}
byte[] content = new byte[length];
if (in.readableBytes() >= length){
in.readBytes(content);
//封装成MyMessageProtocol对象，传递到下一个handler业务处理
MyMessageProtocol messageProtocol = new MyMessageProtocol();
messageProtocol.setLen(length);
messageProtocol.setContent(content);
out.add(messageProtocol);
}
length = 0;
}
}
}2.5? 官方自定义长度半包读写器LengthFieldBasedFrameDecoder
这个解码器也是根据消息的?Header+Body来拆分,即我的消息前多少位代表长度,剩下的是我真正的消息
使用方法也和之前使用DelimiterBasedFrameDecoder一样,就不多说,下面介绍一下他的构造函数的参数.
官方文档：https://netty.io/4.0/api/io/netty/handler/codec/LengthFieldBasedFrameDecoder.html
构造函数
public LengthFieldBasedFrameDecoder(
int maxFrameLength,
int lengthFieldOffset, int lengthFieldLength,
int lengthAdjustment, int initialBytesToStrip)
- maxFrameLength 数据包的最大长度? ??
- lengthFieldOffset 长度字段的偏移位，长度字段开始的地方，意思是跳过指定长度个字节之后的才是消息体字段
- lengthFieldLength 长度字段占的字节数, 帧数据长度的字段本身的长度
- lengthAdjustment ?? ?一般 Header + Body，添加到长度字段的补偿值,如果为负数，开发人员认为这个 Header的长度字段是整个消息包的长度，则Netty应该减去对应的数字
- initialBytesToStrip 从解码帧中第一次去除的字节数, 获取完一个完整的数据包之后，忽略前面的指定位数的长度字节，应用解码器拿到的就是不带长度域的数据包
- failFast 是否快速失败