本篇 主要讲的是自定义协议是如何实现的，以及自定义协议中会出现的问题和Netty是如何支持的。
分为4个部分
|-- 粘包 拆包 和解决方案
|-- 代码实现
|-- ByteToMessageDecoder的源码分析
|-- 过程流程图
粘包
在传输数据太多的时候，TCP是会将数据块分包发送的，也就是在网络延迟的问题，本来一个完整数据块分成两个包，在开始读取数据的时候，只收到了一个数据包，这个时候，怎么办？如果先处理一个数据包，这样数据不完整。等待？那如何知道数据完整了呢？
拆包
TCP是以字节流流的方式来传输的，数据是存储在缓冲区。虽然发送数据是以每个包发送的，但如果网络出现延迟，在第一个包的数据还存储在缓冲区的时候，第二个包就发送过来了。此时第二个包的数据也存储到缓冲区，这时候，就不知道第一个包在哪儿结束，第二个包的数据是从哪里开始读取了。这就是TCP粘包。
解决方案
方案一：
解决方案其实就是，如何去自定义定义这个协议包，去解决粘包的问题和数据包不全的问题。
自定义协议包括如下：
- 一个开始标志：比如定义一个Int类型，4个字节的标志。那么在读到这个开始标志的时候就判断为是一个数据包的开始。
- 数据的长度：也是Int4个字节,表明这个数据块的大小是多小个字节，这样根据这个长度就可以知道数据包是否已经接受完毕，如果还没有，那么就等待。
- 数据：真正的传输数据
代码实现
协议包对象类
/**
*
* 自己定义的协议
* 数据包格式
* +——----——+——-----——+——----——+
* |协议开始标志| 长度 | 数据 |
* +——----——+——-----——+——----——+
* 1.协议开始标志head_data，为int类型的数据，16进制表示为0X76
* 2.传输数据的长度contentLength，int类型
* 3.要传输的数据
*
*/
public class CustomDate {
/**
* 消息开头的信息标志
* 是一个常量 X077
*/
private final int head_Date = Costom.HEAD_DATA.getVaule();
/**
* 消息的长度
*/
private int contentLength;
/**
* 消息的内容
*/
private byte[] conctent;
public CustomDate() {
super();
}
public CustomDate(int contentLength, byte[] conctent) {
this.contentLength = contentLength;
this.conctent = conctent;
}
public int getContentLength() {
return contentLength;
}
public void setContentLength(int contentLength) {
this.contentLength = contentLength;
}
public byte[] getConctent() {
return conctent;
}
public void setConctent(byte[] conctent) {
this.conctent = conctent;
}
public int getHead_Date() {
return head_Date;
}
}
解码类
核心思想：
- 1 在开始读取数据的时候先判断字节大小是否基本数据长度 （标志+数据长度）
- 2 如果缓冲区数据太大，这种情况不正常，应该移动2048个字节，直接处理后面的字节。因为，可能是网络延迟导致，或者是恶意发送大量数据。
- 3 开始读取缓冲区了，对缓冲区的操作。首先标记一下阅读标记点，然后开始寻找开始标记，如果不是开始标记，那么就跳过一个标记节点。
- 4 如果找到了开始标记，那么就继续获取长度。如果长度大小大于缓冲区的可读长度，那么就证明还有数据还没到。就回滚到阅读标记点。继续等待数据。
- 5 如果数据已经到达了，那么就开始读取数据区。
继承 ByteToMessageDecoder 类。该类主要作用是将从网络缓冲区读取的字节转换成有意义的消息对象的
/**
*
* 自己定义的协议
* 数据包格式
* +——----——+——-----——+——----——+
* |协议开始标志| 长度 | 数据 |
* +——----——+——-----——+——----——+
* 1.协议开始标志head_data，为int类型的数据，16进制表示为0X76
* 2.传输数据的长度contentLength，int类型
* 3.要传输的数据,长度不应该超过2048，防止socket流的攻击
*
*/
public class CustomDecoder extends ByteToMessageDecoder {
/**
* 协议开始的标准head_data，int类型，占据4个字节.
* 表示数据的长度contentLength，int类型，占据4个字节.
*/
private final int BASE_LENGTH = 4 + 4;
@Override
protected void decode(ChannelHandlerContext ctx, ByteBuf buffer, List<Object> out) throws Exception {
//1. 首先确认可读长度大于基本长度
if (buffer.readableBytes() > BASE_LENGTH) {
//2.
// 防止socket字节流攻击
// 防止，客户端传来的数据过大
// 因为，太大的数据，是不合理的
if (buffer.readableBytes() > 2048) {
//将readerIndex移动
buffer = buffer.skipBytes(buffer.readableBytes());
}
//3. 记录阅读开始
int beginRead;
while (true) {
//获取包头开始的index;
beginRead = buffer.readerIndex();
// 标记包头开始的index
buffer.markReaderIndex();
//如果读到了数据包的协议开头，那么就结束循环
if (buffer.readInt() == Costom.HEAD_DATA.getVaule()) {
break;
}
//没读到协议开头，退回到标记
buffer.resetReaderIndex();
//跳过一个字节
buffer.readByte();
//如果可读长度小于基本长度
//
if (buffer.readableBytes() < BASE_LENGTH) {
return;
}
}
//获取消息的长度
int length = buffer.readInt();
//判断请求数据包是否到齐
if (buffer.readableBytes() < length) {
buffer.resetReaderIndex();
return;
}
byte[] date = new byte[length];
buffer.readBytes(date);
CustomDate customDate = new CustomDate(length, date);
out.add(customDate);
}
}
}
编码类
- 其实就是往缓冲区里面写数据。
- 继承 MessageToByteEncoder
public class CustomEncoder extends MessageToByteEncoder<CustomDate> {
@Override
protected void encode(ChannelHandlerContext ctx, CustomDate msg, ByteBuf out) throws Exception {
out.writeInt(msg.getHead_Date());
out.writeInt(msg.getContentLength());
out.writeBytes(msg.getConctent());
}
}
添加到管道
public class ServerHandlerInitializer extends ChannelInitializer<Channel> {
@Override
protected void initChannel(Channel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
pipeline.addLast(new CustomDecoder());
pipeline.addLast(new CustomEncoder());
pipeline.addLast(new ServerHandler());
}
}
输出协议数据
public class ServerHandler extends ChannelInboundHandlerAdapter {
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
if (msg instanceof CustomDate) {
CustomDate customDate= (CustomDate) msg;
byte[] conctent = customDate.getConctent();
System.out.println("获取到的内容"+new String(conctent));
ReferenceCountUtil.release(msg);
}
}
}
过程分析
我们研究一下解码的过程。
1 自定义的解码类是继承ByteToMessageDecoder类。先看下ByteToMessageDecoder类
public abstract class ByteToMessageDecoder extends ChannelInboundHandlerAdapter{}
可以看到ByteToMessageDecoder 是继承ChannelInboundHandlerAdapter，那也就是说，数据处理应该是通过重写channelRead（）类了。
2 那就继续看ByteToMessageDecoder 的channelRead() 方法
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
if (msg instanceof ByteBuf) {
CodecOutputList out = CodecOutputList.newInstance();
try {
// 获取到缓冲区
ByteBuf data = (ByteBuf) msg;
first = cumulation == null;
if (first) {
cumulation = data;
} else {
cumulation = cumulator.cumulate(ctx.alloc(), cumulation, data);
}
// 2 开始解码
callDecode(ctx, cumulation, out);
} finally {
// 资源释放代码
}
} else {
ctx.fireChannelRead(msg);
}
}
重点是2 callDecode（）方法。该方法是开始解码。继续往下看该方法
protected void callDecode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {
try {
// 1
while (in.isReadable()) {
int outSize = out.size();
// 2
if (outSize > 0) {
fireChannelRead(ctx, out, outSize);
out.clear();
if (ctx.isRemoved()) {
break;
}
outSize = 0;
}
//3
int oldInputLength = in.readableBytes();
//4
decodeRemovalReentryProtection(ctx, in, out);
if (ctx.isRemoved()) {
break;
}
//5
if (outSize == out.size()) {
if (oldInputLength == in.readableBytes()) {
break;
} else {
continue;
}
}
if (oldInputLength == in.readableBytes()) {
throw new DecoderException( );
}
if (isSingleDecode()) {
break;
}
}
}
}
1 用while循环不断处理缓冲区，判断条件是如果缓冲区还有可读数据，就继续执行。
2 这个是非常好的设计，Out变量存储的是解码生成的对象。如果out里面已经有对象，那么就把该对象通过fireChannelRead（）方法传到下一个handler（也就是本程序中的输出handler）。
出现这种情况是因为：粘包！！！！！ 当处理完一个数据包的数据后，缓冲区还有下一个数据包的数据，所以先把处理完的数据包交给下一个handler处理后，再进行缓冲区的读取。
3 做一个标记。记录这次解码对缓冲区数据有没有被读取（也就是有没有读取数据）。如果没有，下面就会结束while循环。
为什么会要做这个标记呢？为什么要结束循环呢？
因为：缓冲区的数据没有读取，也就是说数据还没全部到齐，需要等待数据完整再处理。所以就需要结束while循环。等待下一次的处理。
4 decodeRemovalReentryProtection（） 就是调用自己重写的decode（）方法了。
final void decodeRemovalReentryProtection(ChannelHandlerContext ctx, ByteBuf in, List<Object> out)
throws Exception {
decodeState = STATE_CALLING_CHILD_DECODE;
try {
// 自己重写的decode
decode(ctx, in, out);
} finally {
//省略
}
}
5 这里就是判断3 中的标记，是否退出循环。
过程流程图
看完代码分析还是一头雾水？ 那就再看一下流程图吧
image.png
网友评论