欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识!
您现在的位置是: 首页

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的设计初衷表达出来了。