Netty笔记:ReplayingDecoder
程序员文章站
2022-03-02 13:49:01
...
为什么需要ReplayingDecoder,它和FrameDecoder有什么不同呢?ReplayingDecoder是一种特殊FrameDecoder,它能实现在IO阻塞的情况下实现无阻塞的解码。听起来比较拗口而且不好理解,举个例子来说,假设你在用netty设计一个文件传输的模块,采用的Header-Content的协议,在Header中说明文件的长度,我们用一个int来容纳长度,Content则是该int长度的字节数组。该如何设计这个Decoder呢?代码有可能是下面这样的。可能你会觉得比你想象的要复杂,多了markReadIndex(),多了是否可读的判断,假如读一个int,我们得首先判断可读的字节数是否大于等于4个。为什么需要这样呢?这就回到了开头的问题,一种原因是和FrameDecoder有关,假设每一帧都是100bytes,其中4bytes用来标示Header中的int,其余96字节为content。假设NioWorker读取了1003bytes交给FrameDecoder去处理,那么前10帧都能完美地处理,当处理第11帧的时候,要读取int的时候,但是只有3bytes了。为了防范这类问题我们需要在每次解码之前mark一下当前的index,发现没有足够的bytes时resetindex为之前mark的值。并且返回null在下一个loop中处理。
buffer.markReaderIndex();
i++; if(buffer.readableBytes() < 4){ //buffer.resetReaderIndex(); return null; } int nameLen = buffer.readInt(); if(buffer.readableBytes() < nameLen){ buffer.resetReaderIndex(); return null; } System.out.println("文件名长" + nameLen); byte[] fileNameBytes = new byte[nameLen]; buffer.readBytes(fileNameBytes); String fileName = new String(fileNameBytes); int contentLen = buffer.readInt(); byte[] fileContent = new byte[contentLen]; if(buffer.readableBytes() < contentLen){ System.out.println("只读到了" + buffer.readableBytes()); System.out.println("待下次接收完成再处理"); // byte[] read = buffer.readBytes(buffer.readableBytes()).array(); // System.out.println(new String(read)); buffer.resetReaderIndex(); return null; }
这样使用FrameDecoder,虽然能解决问题但看起来不怎么优雅。上面的代码还有一些性能的问题,我们每次都是在解码每一帧最开始的地方mark,假设我们的Header有N项,而不仅仅只有一项长度。那么读取任何一项的时候,发现没有足够的bytes都会回到帧初始的index,直到能满足该帧所有的bytes,才会解析成完整的帧,这样的每一次解码的时候都会做一些重复的工作,以上面的code为例,每次一次编码都解析文件名长度,文件名,文件内容长度,直到读取到完整的文件内容,假设文件内容为1m,每次从NioWorker读取的byte的数量为4096,那么我们得需要做25次重复尝试的解析。并且这些解析都是徒劳无功的。有没有更好的办法,当然是有的,我们在每次成功读取以后mark,这样我们还得需要一个状态机的状态流转而不能每次都从头开始,比如读取文件名长度,文件名,文件长度,文件内容都是一个状态,而且这几个状态之间顺序流转。有没有优雅的方案,不需要多次mark、check、reset,ReplayingDecoder应运而生。他是怎么做到的呢?
为了解决读取时没有足够的bytes,设计了ReplayingDecoderBuffer,他重写所有read相关的方法,在read之前首先check长度。当长度不够时抛出ReplayError,在Decoder时catch到ReplayError,重置index
@Override
public int readInt() {
checkReadableBytes(4);
return buffer.readInt();
}
为了解决多次mark的问题,引入了checkpoint和State状态机。checkpoint为cumulation的readIndex。使用ReplayingDecoder以后最初的代码是这样的
switch (state) {
case READ_FILENAME_LENGTH: {
int fileNameLength = buffer.readInt();
currentTransfer = new TransferFile();
currentTransfer.setFileNameLength(fileNameLength);
checkpoint(State.READ_FILENAME);
}
case READ_FILENAME:{
byte[] fileNameBytes = new byte[currentTransfer.getFileNameLength()];
buffer.readBytes(fileNameBytes);
String fileName = new String(fileNameBytes);
try{
currentTransfer.setFileName(fileName);
FileOutputStream fos = new FileOutputStream(fileName);
currentTransfer.setFos(fos);
}catch(Exception e){
currentTransfer.setSuccess(false);
return reset();
}
}
case READ_FILECONTENT_LENGTH:{
currentTransfer.setFileContentLength(buffer.readInt());
checkpoint(State.READ_FILECONTENT);
}
case READ_FILECONTENT:{
int currentChunkSize = this.chunkSize;
int maxCanRead = buffer.readableBytes();
if(maxCanRead < currentTransfer.getFileContentLength()){
currentChunkSize = Math.min(maxCanRead, chunkSize);
}else{
currentChunkSize = currentTransfer.getFileContentLength();
}
try{
currentTransfer.getFos().write(buffer.readBytes(currentChunkSize).array());
currentTransfer.decrement(currentChunkSize);
}catch(Exception e){
currentTransfer.setSuccess(false);
return reset();
}
if(currentTransfer.isLast()){
currentTransfer.setSuccess(true);
return reset();//读取完一个完整的帧
}
}
}
return null;
就到这儿吧。基本上把ReplayingDecoder的设计初衷表达出来了。
上一篇: Flex Slide Show