Netty学习篇⑤--编、解码

荷塘月色0519 2019-11-20 我要评论

前言

学习Netty也有一段时间了，Netty作为一个高性能的异步框架，很多RPC框架也运用到了Netty中的知识，在rpc框架中丰富的数据协议及编解码可以让使用者更加青睐；
Netty支持丰富的编解码框架，其本身内部提供的编解码也可以应对各种业务场景；
今天主要就是学习下Netty中提供的编、解码类，之前只是简单的使用了下Netty提供的解码类，今天更加深入的研究下Netty中编、解码的源码及部分使用。

编、解码的概念

编码（Encoder）

编码就是将我们发送的数据编码成字节数组方便在网络中进行传输，类似Java中的序列化，将对象序列化成字节传输

解码（Decoder）

解码和编码相反，将传输过来的字节数组转化为各种对象来进行展示等，类似Java中的反序列化
如：
// 将字节数组转化为字符串
new String(byte bytes[], Charset charset)

编、解码超类

ByteToMessageDecoder: 解码超类，将字节转换成消息

解码解码一般用于将获取到的消息解码成系统可识别且自己需要的数据结构；因此ByteToMessageDecoder需要继承ChannelInboundHandlerAdapter入站适配器来获取到入站的数据，在handler使用之前通过channelRead获取入站数据进行一波解码；

ByteToMessageDecoder类图

源码分析

通过channelRead获取入站数据，将数据缓存至cumulation数据缓冲区，最后在传给decode进行解码，在read完成之后清空缓存的数据

1. 获取入站数据

/**
*  通过重写channelRead方法来获取入站数据
*/
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
    // 检测是否是byteBuf对象格式数据
    if (msg instanceof ByteBuf) {
        // 实例化字节解码成功输出集合 即List<Object> out
        CodecOutputList out = CodecOutputList.newInstance();
        try {
            // 获取到的请求的数据
            ByteBuf data = (ByteBuf) msg;
            // 如果缓冲数据区为空则代表是首次触发read方法
            first = cumulation == null;
            if (first) {
                // 如果是第一次read则当前msg数据为缓冲数据
                cumulation = data;
            } else {
                // 如果不是则触发累加，将缓冲区的旧数据和新获取到的数据通过          expandCumulation 方法累加在一起存入缓冲区cumulation
                // cumulator 累加类，将缓冲池中数据和新数据进行组合在一起
                // private Cumulator cumulator = MERGE_CUMULATOR;
                cumulation = cumulator.cumulate(ctx.alloc(), cumulation, data);
            }
            // 将缓冲区数据cumulation进行解码
            callDecode(ctx, cumulation, out);
        } catch (DecoderException e) {
            throw e;
        } catch (Throwable t) {
            throw new DecoderException(t);
        } finally {
            // 在解码完毕后释放引用和清空全局字节缓冲区
            if (cumulation != null && !cumulation.isReadable()) {
                numReads = 0;
                cumulation.release();
                cumulation = null;
                // discardAfterReads为netty中设置的读取多少次后开始丢弃字节 默认值16
                // 可通过setDiscardAfterReads(int n)来设置值不设置默认16次
            } else if (++ numReads >= discardAfterReads) {
                // We did enough reads already try to discard some bytes so we not risk to see a OOME.
                // 在我们读取了足够的数据可以尝试丢弃一些字节已保证不出现内存溢出的异常
                // 
                // See https://github.com/netty/netty/issues/4275
                // 读取次数重置为0
                numReads = 0;
                // 重置读写指针或丢弃部分已读取的字节
                discardSomeReadBytes();
            }
          // out为解码成功的传递给下一个handler
            int size = out.size();
            decodeWasNull = !out.insertSinceRecycled();
            // 结束当前read传递到下个ChannelHandler
            fireChannelRead(ctx, out, size);
            // 回收响应集合 将insertSinceRecycled设置为false；
            // insertSinceRecycled用于channelReadComplete判断使用
            out.recycle();
        }
    } else {
        // 不是的话直接fire传递给下一个handler
        ctx.fireChannelRead(msg);
    }
}

2. 初始化字节缓冲区计算器: Cumulator主要用于全局字节缓冲区和新读取的字节缓冲区组合在一起扩容

public static final Cumulator MERGE_CUMULATOR = new Cumulator() {

    /**
    * alloc ChannelHandlerContext分配的字节缓冲区
    * cumulation 当前ByteToMessageDecoder类全局的字节缓冲区
    * in 入站的字节缓冲区
    **/
    @Override
    public ByteBuf cumulate(ByteBufAllocator alloc, ByteBuf cumulation, ByteBuf in) {
        final ByteBuf buffer;
        // 如果全局ByteBuf写入的字节+当前入站的字节数据大于全局缓冲区最大的容量或者全局缓冲区的引用数大于1个或全局缓冲区只读
        if (cumulation.writerIndex() > cumulation.maxCapacity() - in.readableBytes()
            || cumulation.refCnt() > 1 || cumulation.isReadOnly()) {
            // Expand cumulation (by replace it) when either there is not more room in the buffer
            // or if the refCnt is greater then 1 which may happen when the user use slice().retain() or
            // duplicate().retain() or if its read-only.
            //
            // See:
            // - https://github.com/netty/netty/issues/2327
            // - https://github.com/netty/netty/issues/1764
            // 进行扩展全局字节缓冲区（容量大小 = 新数据追加到旧数据末尾组成新的全局字节缓冲区）
            buffer = expandCumulation(alloc, cumulation, in.readableBytes());
        } else {
            buffer = cumulation;
        }
        // 将新数据写入缓冲区
        buffer.writeBytes(in);
        // 释放当前的字节缓冲区的引用
        in.release();

        return buffer;
    }
};


/**
* alloc 字节缓冲区操作类
* cumulation 全局累加字节缓冲区
* readable 读取到的字节数长度
*/
// 字节缓冲区扩容方法
static ByteBuf expandCumulation(ByteBufAllocator alloc, ByteBuf cumulation, int readable) {
    // 旧数据
    ByteBuf oldCumulation = cumulation;
    // 通过ByteBufAllocator将缓冲区扩大到oldCumulation + readable大小
    cumulation = alloc.buffer(oldCumulation.readableBytes() + readable);
    // 将旧数据重新写入到新的字节缓冲区
    cumulation.writeBytes(oldCumulation);
    // 旧字节缓冲区引用-1
    oldCumulation.release();
    return cumulation;
}

3. ByteBuf释放当前字节缓冲区的引用: 通过调用ReferenceCounted接口中的release方法来释放

@Override
public boolean release() {
    return release0(1);
}

@Override
public boolean release(int decrement) {
    return release0(checkPositive(decrement, "decrement"));
}

/**
* decrement 减量
*/
private boolean release0(int decrement) {
    for (;;) {
        int refCnt = this.refCnt;
        // 当前引用小于减量
        if (refCnt < decrement) {
            throw new IllegalReferenceCountException(refCnt, -decrement);
        }
      // 这里就利用里线程并发中的知识CAS，线程安全的设置refCnt的值
        if (refCntUpdater.compareAndSet(this, refCnt, refCnt - decrement)) {
            // 如果减量和引用量相等
            if (refCnt == decrement) {
                // 全部释放
                deallocate();
                return true;
            }
            return false;
        }
    }
}

4. 将全局字节缓冲区进行解码

/**
* ctx ChannelHandler的上下文，用于传输数据与下一个handler来交互
* in 入站数据
* out 解析之后的出站集合 （此出站不是返回给客户端的而是传递给下个handler的）
*/
protected void callDecode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {
    try {
        // 如果入站数据还有没解析的
        while (in.isReadable()) {
            // 解析成功的出站集合长度
            int outSize = out.size();
          // 如果大于0则说明解析成功的数据还没被消费完，直接fire掉给通道中的后续handler继续                消费
            if (outSize > 0) {
                fireChannelRead(ctx, out, outSize);
                out.clear();

                // Check if this handler was removed before continuing with decoding.
                // 在这个handler删除之前检查是否还在继续解码
                // If it was removed, it is not safe to continue to operate on the buffer.
                // 如果移除了，它继续操作缓冲区是不安全的
                //
                // See:
                // - https://github.com/netty/netty/issues/4635
                if (ctx.isRemoved()) {
                    break;
                }
                outSize = 0;
            }
          // 入站数据字节长度
            int oldInputLength = in.readableBytes();
            // 开始解码数据
            decodeRemovalReentryProtection(ctx, in, out);

            // Check if this handler was removed before continuing the loop.
            // 
            // If it was removed, it is not safe to continue to operate on the buffer.
            //
            // See https://github.com/netty/netty/issues/1664
            if (ctx.isRemoved()) {
                break;
            }

            // 解析完毕跳出循环
            if (outSize == out.size()) {
                if (oldInputLength == in.readableBytes()) {
                    break;
                } else {
                    continue;
                }
            }

            if (oldInputLength == in.readableBytes()) {
                throw new DecoderException(
                    StringUtil.simpleClassName(getClass()) +
                    ".decode() did not read anything but decoded a message.");
            }

            if (isSingleDecode()) {
                break;
            }
        }
    } catch (DecoderException e) {
        throw e;
    } catch (Throwable cause) {
        throw new DecoderException(cause);
    }
}

final void decodeRemovalReentryProtection(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
      // 设置解码状态为正在解码  STATE_INIT = 0; STATE_CALLING_CHILD_DECODE = 1;             STATE_HANDLER_REMOVED_PENDING = 2; 分别为初始化； 解码； 解码完毕移除
        decodeState = STATE_CALLING_CHILD_DECODE;
        try {
            // 具体的解码逻辑（netty提供的解码器或自定义解码器中重写的decode方法）
            decode(ctx, in, out);
        } finally {
            // 此时decodeState为正在解码中 值为1，返回false
            boolean removePending = decodeState == STATE_HANDLER_REMOVED_PENDING;
            // 在设置为初始化等待解码
            decodeState = STATE_INIT;
            // 解码完成移除当前ChannelHandler标记为不处理
            // 可以看看handlerRemoved源码。如果缓冲区还有数据直接传递给下一个handler
            if (removePending) {
                handlerRemoved(ctx);
            }
        }
    }

5. 执行channelReadComplete

@Override
public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
    // 读取次数重置
    numReads = 0;
    // 重置读写index
    discardSomeReadBytes();
    // 在channelRead meth中定义赋值 decodeWasNull = !out.insertSinceRecycled();
    // out指的是解码集合List<Object> out; 咱们可以点进
    if (decodeWasNull) {
        decodeWasNull = false;
        if (!ctx.channel().config().isAutoRead()) {
            ctx.read();
        }
    }
    // fire掉readComplete传递到下一个handler的readComplete
    ctx.fireChannelReadComplete();
}

/**
*  然后我们可以搜索下insertSinceRecucled在什么地方被赋值了
* Returns {@code true} if any elements where added or set. This will be reset once {@link #recycle()} was called.
*/
boolean insertSinceRecycled() {
    return insertSinceRecycled;
}


// 搜索下insert的调用我们可以看到是CodecOutputList类即为channelRead中的out集合，众所周知在  decode完之后，解码数据就会被调用add方法，此时insertSinceRecycled被设置为true
private void insert(int index, Object element) {
    array[index] = element;
    insertSinceRecycled = true;
}


/**
* 清空回收数组内部的所有元素和存储空间
* Recycle the array which will clear it and null out all entries in the internal storage.
*/
// 搜索recycle的调用我么可以知道在channelRead的finally逻辑中 调用了out.recycle();此时      insertSinceRecycled被设置为false
void recycle() {
    for (int i = 0 ; i < size; i ++) {
        array[i] = null;
    }
    clear();
    insertSinceRecycled = false;
    handle.recycle(this);
}

至此ByteToMessageDecoder解码类应该差不多比较清晰了！！！

MessageToByteEncoder: 编码超类，将消息转成字节进行编码发出

何谓编码，就是将发送数据转化为客户端和服务端约束好的数据结构和格式进行传输，我们可以在编码过程中将消息体body的长度和一些头部信息有序的设置到ByteBuf字节缓冲区中；方便解码方灵活的运用来判断（是否完整的包等）和处理业务；解码是继承入站数据，反之编码应该继承出站的数据；接下来我们看看编码类是怎么进行编码的；

MessageToByteEncoder类图如下

源码分析

既然是继承出站类，我们直接看看write方法是怎么样的

/**
* 通过write方法获取到出站的数据即要发送出去的数据
* ctx channelHandler上下文
* msg 发送的数据 Object可以通过继承类指定的泛型来指定
* promise channelPromise异步监听，类似ChannelFuture，只不过promise可以设置监听的结果，future只能通过获取监听的成功失败结果；可以去了解下promise和future的区别
*/
@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
    ByteBuf buf = null;
    try {
        // 检测发送数据的类型 通过TypeParameterMatcher类型匹配器
        if (acceptOutboundMessage(msg)) {
            @SuppressWarnings("unchecked")
            I cast = (I) msg;
            // 分配字节缓冲区 preferDirect默认为true
            buf = allocateBuffer(ctx, cast, preferDirect);
            try {
                // 进行编码
                encode(ctx, cast, buf);
            } finally {
                // 完成编码后释放对象的引用
                ReferenceCountUtil.release(cast);
            }
           // 如果缓冲区有数据则通过ctx发送出去，promise可以监听数据传输并设置是否完成
            if (buf.isReadable()) {
                ctx.write(buf, promise);
            } else {
                // 如果没有数据则释放字节缓冲区的引用并发送一个empty的空包
                buf.release();
                ctx.write(Unpooled.EMPTY_BUFFER, promise);
            }
            buf = null;
        } else {
            // 非TypeParameterMatcher类型匹配器匹配的类型直接发送出去
            ctx.write(msg, promise);
        }
    } catch (EncoderException e) {
        throw e;
    } catch (Throwable e) {
        throw new EncoderException(e);
    } finally {
        if (buf != null) {
            buf.release();
        }
    }
}

// 初始化设置preferDirect为true
protected MessageToByteEncoder() {
    this(true);
}
protected MessageToByteEncoder(boolean preferDirect) {
    matcher = TypeParameterMatcher.find(this, MessageToByteEncoder.class, "I");
    this.preferDirect = preferDirect;
}

编码：重写encode方法，根据实际业务来进行数据编码

// 此处就是我们需要重写的编码方法了，我们和根据约束好的或者自己定义好想要的数据格式发送给对方

// 下面是我自己写的demo的编码方法；头部设置好body的长度，服务端可以根据长度来判断是否是完整的包，仅仅自学写的简单的demo非正常线上运营项目的逻辑
public class MyClientEncode extends MessageToByteEncoder<String> {

    @Override
    protected void encode(ChannelHandlerContext ctx, String msg, ByteBuf out) throws Exception {
        if (null != msg) {
            byte[] request = msg.getBytes(Charset.forName("UTF-8"));
            out.writeInt(request.length);
            out.writeBytes(request);
        }
    }
}

编码类相对要简单很多，因为只需要将发送的数据序列化，按照一定的格式进行发送数据！！！

项目实战

项目主要简单的实现下自定义编解码器的运用及LengthFieldBasedFrameDecoder的使用

项目结构如下

│  hetangyuese-netty-06.iml
│  pom.xml
│
├─src
│  ├─main
│  │  ├─java
│  │  │  └─com
│  │  │      └─hetangyuese
│  │  │          └─netty
│  │  │              ├─client
│  │  │              │      MyClient06.java
│  │  │              │      MyClientChannelInitializer.java
│  │  │              │      MyClientDecoder.java
│  │  │              │      MyClientEncode.java
│  │  │              │      MyClientHandler.java
│  │  │              │      MyMessage.java
│  │  │              │
│  │  │              └─server
│  │  │                      MyChannelInitializer.java
│  │  │                      MyServer06.java
│  │  │                      MyServerDecoder.java
│  │  │                      MyServerDecoderLength.java
│  │  │                      MyServerEncoder.java
│  │  │                      MyServerHandler.java
│  │  │
│  │  └─resources
│  └─test
│      └─java

服务端

Serverhandler: 只是简单的将解码的内容输出

public class MyServerHandler extends ChannelInboundHandlerAdapter {

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("客户端连接成功 time: " + new Date().toLocaleString());
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("客户端断开连接 time: " + new Date().toLocaleString());
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        String body = (String) msg;
        System.out.println("content:" + body);
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        // 出现异常关闭通道
        cause.printStackTrace();
        ctx.close();
    }
}

解码器

public class MyServerDecoder extends ByteToMessageDecoder {

    // 此处我头部只塞了长度字段占4个字节，别问为啥我知道，这是要客户端和服务端约束好的
    private static int min_head_length = 4;

    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
        // 解码的字节长度
        int size = in.readableBytes();
        if(size < min_head_length) {
            System.out.println("解析的数据长度小于头部长度字段的长度");
            return ;
        }
        // 读取的时候指针已经移位到长度字段的尾端
        int length = in.readInt();
        if (size < length) {
            System.out.println("解析的数据长度与长度不符合");
            return ;
        }

        // 上面已经读取到了长度字段，后面的长度就是body
        ByteBuf decoderArr = in.readBytes(length);
        byte[] request = new byte[decoderArr.readableBytes()];
        // 将数据写入空数组
        decoderArr.readBytes(request);
        String body = new String(request, Charset.forName("UTF-8"));
        out.add(body);
    }
}

将解码器加入到channelHandler中：记得加到业务handler的前面否则无效

public class MyChannelInitializer extends ChannelInitializer<SocketChannel> {

    @Override
    protected void initChannel(SocketChannel ch) throws Exception {
        ch.pipeline()
//                .addLast(new MyServerDecoderLength(10240, 0, 4, 0, 0))
//                .addLast(new LengthFieldBasedFrameDecoder(10240, 0, 4, 0, 0))
                .addLast(new MyServerDecoder())
                .addLast(new MyServerHandler())
        ;
    }
}

客户端

ClientHandler

public class MyClientHandler extends ChannelInboundHandlerAdapter {

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("与服务端连接成功");
        for (int i = 0; i<10; i++) {
            ctx.writeAndFlush("hhhhh" + i);
        }
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("与服务端断开连接");
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        System.out.println("收到服务端消息:" +msg+ " time: " + new Date().toLocaleString());
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.close();
    }
}

编码器

public class MyClientEncode extends MessageToByteEncoder<String> {

    @Override
    protected void encode(ChannelHandlerContext ctx, String msg, ByteBuf out) throws Exception {
        if (null != msg) {
            byte[] request = msg.getBytes(Charset.forName("UTF-8"));
            out.writeInt(request.length);
            out.writeBytes(request);
        }
    }
}

将编码器加到ClientHandler的前面

public class MyClientChannelInitializer extends ChannelInitializer<SocketChannel> {

    @Override
    protected void initChannel(SocketChannel ch) throws Exception {
        ch.pipeline()
                .addLast(new MyClientDecoder())
                .addLast(new MyClientEncode())
                .addLast(new MyClientHandler())
        ;

    }
}

服务端运行结果

MyServer06 is start ...................
客户端连接成功 time: 2019-11-19 16:35:47
content:hhhhh0
content:hhhhh1
content:hhhhh2
content:hhhhh3
content:hhhhh4
content:hhhhh5
content:hhhhh6
content:hhhhh7
content:hhhhh8
content:hhhhh9

如果不用自定义的解码器怎么获取到body内容呢

将自定义编码器换成LengthFieldBasedFrameDecoder(10240, 0, 4, 0, 0)

public class MyChannelInitializer extends ChannelInitializer<SocketChannel> {

    @Override
    protected void initChannel(SocketChannel ch) throws Exception {
        ch.pipeline()
//                .addLast(new MyServerDecoderLength(10240, 0, 4, 0, 0))
                .addLast(new LengthFieldBasedFrameDecoder(10240, 0, 4, 0, 0))
//                .addLast(new MyServerDecoder())
                .addLast(new MyServerHandler())
        ;
    }
}

// 怕忘记的各个参数的含义在这在说明一次，自己不断的修改每个值观察结果就可以更加深刻的理解
/**
* maxFrameLength：消息体的最大长度，好像默认最大值为1024*1024
* lengthFieldOffset 长度字段所在字节数组的下标 （我这是第一个write的所以下标是0）
* lengthFieldLength 长度字段的字节长度（int类型占4个字节）
* lengthAdjustment 长度字段补偿的数值 （lengthAdjustment =  数据包长度 - lengthFieldOffset - lengthFieldLength - 长度域的值），解析需要减去对应的数值
* initialBytesToStrip 是否去掉长度字段（0不去除，对应长度域字节长度）
*/
public LengthFieldBasedFrameDecoder(
            int maxFrameLength,
            int lengthFieldOffset, int lengthFieldLength,
            int lengthAdjustment, int initialBytesToStrip)

结果: 前都带上了长度

MyServer06 is start ...................
客户端连接成功 time: 2019-11-19 17:53:42
收到客户端发来的消息:   hhhhh0, time: 2019-11-19 17:53:42
收到客户端发来的消息:   hhhhh1, time: 2019-11-19 17:53:42
收到客户端发来的消息:   hhhhh2, time: 2019-11-19 17:53:42
收到客户端发来的消息:   hhhhh3, time: 2019-11-19 17:53:42
收到客户端发来的消息:   hhhhh4, time: 2019-11-19 17:53:42
收到客户端发来的消息:   hhhhh5, time: 2019-11-19 17:53:42
收到客户端发来的消息:   hhhhh6, time: 2019-11-19 17:53:42
收到客户端发来的消息:   hhhhh7, time: 2019-11-19 17:53:42
收到客户端发来的消息:   hhhhh8, time: 2019-11-19 17:53:42
收到客户端发来的消息:   hhhhh9, time: 2019-11-19 17:53:42

如果我们在客户端的长度域中做手脚 LengthFieldBasedFrameDecoder(10240, 0, 4, 0, 0)

旧： out.writeInt(request.length);
新： out.writeInt(request.length + 1);

// 看结果就不正常，0后面多了一个0；但是不知道为啥只解码了一次？？？ 求解答
MyServer06 is start ...................
客户端连接成功 time: 2019-11-19 17:56:55
收到客户端发来的消息:   hhhhh0 , time: 2019-11-19 17:56:55

// 正确修改为 LengthFieldBasedFrameDecoder(10240, 0, 4, -1, 0)
// 结果：
MyServer06 is start ...................
客户端连接成功 time: 2019-11-19 18:02:18
收到客户端发来的消息:   hhhhh0, time: 2019-11-19 18:02:18
收到客户端发来的消息:   hhhhh1, time: 2019-11-19 18:02:18
收到客户端发来的消息:   hhhhh2, time: 2019-11-19 18:02:18
收到客户端发来的消息:   hhhhh3, time: 2019-11-19 18:02:18
收到客户端发来的消息:   hhhhh4, time: 2019-11-19 18:02:18
收到客户端发来的消息:   hhhhh5, time: 2019-11-19 18:02:18
收到客户端发来的消息:   hhhhh6, time: 2019-11-19 18:02:18
收到客户端发来的消息:   hhhhh7, time: 2019-11-19 18:02:18
收到客户端发来的消息:   hhhhh8, time: 2019-11-19 18:02:18
收到客户端发来的消息:   hhhhh9, time: 2019-11-19 18:02:18

舍弃长度域：LengthFieldBasedFrameDecoder(10240, 0, 4, 0, 4)

// 结果
MyServer06 is start ...................
客户端连接成功 time: 2019-11-19 18:03:44
收到客户端发来的消息:hhhhh0, time: 2019-11-19 18:03:44
收到客户端发来的消息:hhhhh1, time: 2019-11-19 18:03:44
收到客户端发来的消息:hhhhh2, time: 2019-11-19 18:03:44
收到客户端发来的消息:hhhhh3, time: 2019-11-19 18:03:44
收到客户端发来的消息:hhhhh4, time: 2019-11-19 18:03:44
收到客户端发来的消息:hhhhh5, time: 2019-11-19 18:03:44
收到客户端发来的消息:hhhhh6, time: 2019-11-19 18:03:44
收到客户端发来的消息:hhhhh7, time: 2019-11-19 18:03:44
收到客户端发来的消息:hhhhh8, time: 2019-11-19 18:03:44
收到客户端发来的消息:hhhhh9, time: 2019-11-19 18:03:44

备注

分析源码示例中的 lengthAdjustment = 消息字节长度 - lengthFieldOffset-lengthFieldLength-长度域中的值

源码中的示例

 * <pre>
 * lengthFieldOffset   =  0
 * lengthFieldLength   =  2
 * <b>lengthAdjustment</b>    = <b>-2</b> (= the length of the Length field)
 * initialBytesToStrip =  0
 *
 * BEFORE DECODE (14 bytes)         AFTER DECODE (14 bytes)
 * +--------+----------------+      +--------+----------------+
 * | Length | Actual Content |----->| Length | Actual Content |
 * | 0x000E | "HELLO, WORLD" |      | 0x000E | "HELLO, WORLD" |
 * +--------+----------------+      +--------+----------------+
 * </pre>

长度域中0x000E为16进制，转换成10进制是14，说明消息体长度为14；根据公式：14-0-2-14 = -2

* <pre>
 * lengthFieldOffset   = 0
 * lengthFieldLength   = 3
 * <b>lengthAdjustment</b>    = <b>2</b> (= the length of Header 1)
 * initialBytesToStrip = 0
 *
 * BEFORE DECODE (17 bytes)                      AFTER DECODE (17 bytes)
 * +----------+----------+----------------+      +----------+----------+----------------+
 * |  Length  | Header 1 | Actual Content |----->|  Length  | Header 1 | Actual Content |
 * | 0x00000C |  0xCAFE  | "HELLO, WORLD" |      | 0x00000C |  0xCAFE  | "HELLO, WORLD" |
 * +----------+----------+----------------+      +----------+----------+----------------+
 * </pre>

从上的例子可以知道；lengthAdjustment(2) = 17- 12（00000C）-lengthFieldOffset(0) - lengthFieldLength(3);

.......等等

Netty学习篇⑤--编、解码

前言

编、解码的概念

编、解码超类

ByteToMessageDecoder类图

源码分析

2. 初始化字节缓冲区计算器: Cumulator主要用于全局字节缓冲区和新读取的字节缓冲区组合在一起扩容

3. ByteBuf释放当前字节缓冲区的引用: 通过调用ReferenceCounted接口中的release方法来释放

5. 执行channelReadComplete

MessageToByteEncoder类图如下

源码分析

结果: 前都带上了长度

备注

分析源码示例中的 lengthAdjustment = 消息字节长度 - lengthFieldOffset-lengthFieldLength-长度域中的值

相关文章

猜您喜欢

今日热门