熟悉Netty之?dāng)?shù)據(jù)傳輸--TCP粘/拆包和加解碼器(二)

敦煌天空的沙粒，帶著我們的記憶

我從半路看回去，這秦關(guān)漫漫好蜿踞

夢想穿過了西域，包含了多少的禪意

愛情像一本游記，我會找尋它的密語

看月牙灣下的淚光，在絲路之上被遺忘

上次用一個(gè)小例子提到了Netty是如何處理粘包/拆包的，今天來分析下其中具體的細(xì)節(jié)，首先我們看到例子中用到的一個(gè)類：

io.netty.handler.codec.LineBasedFrameDecoder

下圖是該類的繼承關(guān)系圖：

從上圖可以看出，該類屬于

io.netty.channel.ChannelHandler，

所以，我們可以通過io.netty.channel.ChannelPipeline

添加至Handler中，即：

那么

io.netty.handler.codec.LineBasedFrameDecoder

是如何解決粘包/拆包的呢？

直接進(jìn)去看源代碼：

這里是其主要的decode方法，這里不做逐行翻譯了，只是描述其大致意思

首先通過獲取讀取緩沖區(qū)的數(shù)據(jù)，通過字節(jié)循環(huán)匹配，看是否有"\n"，如果有，則以此位置結(jié)束，從可讀取索引到此位置區(qū)間的字節(jié)組成一行，如果連續(xù)讀取設(shè)定的長度，都沒有發(fā)現(xiàn)"\n"，則拋出異常。

循環(huán)獲取：

拋出異常：

然后，通過

io.netty.handler.codec.string.StringDecoder

將接收到的對象，轉(zhuǎn)化為字符串，調(diào)用后續(xù)的Handler，處理數(shù)據(jù)

io.netty.handler.codec.LineBasedFrameDecoder

和

io.netty.handler.codec.string.StringDecoder

結(jié)合，是一種典型的按換行切換的文本解碼器，它被設(shè)計(jì)用來支持TCP的粘包和拆包

此時(shí)，有看官肯定有一種疑惑，那么如果我現(xiàn)在需求不要按照換行符號切換文本呢，比如我們常常用到的聊天軟件，通常有些是可以指定，按回車鍵也可以不發(fā)送數(shù)據(jù)，繼續(xù)輸入文本，最后按個(gè)什么組合件發(fā)送數(shù)據(jù)，類似QQ或者微信，會提供按Enter鍵發(fā)送數(shù)據(jù)或者Ctrl+Enter發(fā)送數(shù)據(jù)，此時(shí)，需要怎么處理呢？

其實(shí)，無論是Enter或者Ctrl+Enter發(fā)送數(shù)據(jù)，程序都會指定一種形式發(fā)送數(shù)據(jù)，那么在Netty中，同樣支持利用分隔符解碼器來處理數(shù)據(jù)，這個(gè)分隔符則可以任意指定。

如上圖所示，Netty提供了

io.netty.handler.codec.FixedLengthFrameDecoder

和

io.netty.handler.codec.DelimiterBasedFrameDecoder

自定義分隔符和定長解碼器，來解決上面提到的問題

那么先來看看其實(shí)際應(yīng)用：

用前面例子為例：

服務(wù)端：

package com.lgli.netty.tcppackage;
import io.netty.bootstrap.ServerBootstrap;import io.netty.buffer.ByteBuf;import io.netty.buffer.Unpooled;import io.netty.channel.*;import io.netty.channel.nio.NioEventLoopGroup;import io.netty.channel.socket.SocketChannel;import io.netty.channel.socket.nio.NioServerSocketChannel;import io.netty.handler.codec.DelimiterBasedFrameDecoder;import io.netty.handler.codec.LineBasedFrameDecoder;import io.netty.handler.codec.string.StringDecoder;
import java.net.InetSocketAddress;
/** * Server * @author lgli */public class MyServer {
    public static void main(String[] args) {        EventLoopGroup bossGroup = new NioEventLoopGroup();        EventLoopGroup workGroup = new NioEventLoopGroup();        try{            ServerBootstrap bootstrap = new ServerBootstrap();            bootstrap.group(bossGroup,workGroup)                    .channel(NioServerSocketChannel.class)                    .option(ChannelOption.SO_BACKLOG,10240)                    .childHandler(new ServerInitial());            ChannelFuture future = bootstrap.bind(new InetSocketAddress("localhost", 8080)).sync();            future.channel().closeFuture().sync();        }catch (Exception e){            e.printStackTrace();        }finally {            bossGroup.shutdownGracefully();            workGroup.shutdownGracefully();        }    }
    static class ServerInitial extends ChannelInitializer<SocketChannel>{
        @Override        protected void initChannel(SocketChannel ch) throws Exception {            ByteBuf byteBuf = Unpooled.copiedBuffer("$_end".getBytes());            ch.pipeline()                    .addLast("frame",new DelimiterBasedFrameDecoder(1024,byteBuf))                    .addLast("decode",new StringDecoder())                    .addLast("handler",new ServerHandler());        }    }
    static class ServerHandler extends SimpleChannelInboundHandler<Object>{        private int counter;
        @Override        protected void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {            String receive = (String) msg;            System.out.println("服務(wù)端收到客戶端數(shù)據(jù)："+receive+"；目前計(jì)數(shù)counter = "+ ++counter);            String result = "hello,how are you?".equalsIgnoreCase(receive)?"I am fine,thank you":"wrong msg";            result += "$_end";            ByteBuf resultBuf = Unpooled.copiedBuffer(result.getBytes());            ctx.writeAndFlush(resultBuf);        }????}}

客戶端：

package com.lgli.netty.tcppackage;
import io.netty.bootstrap.Bootstrap;import io.netty.buffer.ByteBuf;import io.netty.buffer.Unpooled;import io.netty.channel.*;import io.netty.channel.nio.NioEventLoopGroup;import io.netty.channel.socket.SocketChannel;import io.netty.channel.socket.nio.NioSocketChannel;import io.netty.handler.codec.DelimiterBasedFrameDecoder;import io.netty.handler.codec.LineBasedFrameDecoder;import io.netty.handler.codec.string.StringDecoder;
import java.net.InetSocketAddress;
/** * Client * @author lgli */public class MyClient {    public static void main(String[] args) {        EventLoopGroup group = new NioEventLoopGroup();        try{            Bootstrap bootstrap = new Bootstrap();            bootstrap.group(group)                    .channel(NioSocketChannel.class)                    .option(ChannelOption.SO_KEEPALIVE,true)                    .handler(new ClientInitial());            ChannelFuture future = bootstrap.connect(new InetSocketAddress("localhost", 8080)).sync();            future.channel().closeFuture().sync();        }catch (Exception e){            e.printStackTrace();        }finally {            group.shutdownGracefully();        }    }
    static class ClientInitial extends ChannelInitializer<SocketChannel>{        @Override        protected void initChannel(SocketChannel ch) throws Exception {
            ByteBuf byteBuf = Unpooled.copiedBuffer("$_end".getBytes());            ch.pipeline()                    .addLast("frame",new DelimiterBasedFrameDecoder(1024,byteBuf))                    .addLast("decode",new StringDecoder())                    .addLast("handler",new ClientHandler());        }    }    static class ClientHandler extends SimpleChannelInboundHandler<Object>{        private int counter;        private byte[] req;
        public ClientHandler(){            req = ("hello,how are you?$_end").getBytes();????????}        @Override        public void channelActive(ChannelHandlerContext ctx) throws Exception {            ByteBuf buff = null;            for(int i = 0 ; i < 100 ;i++){                buff = Unpooled.buffer(req.length);                buff.writeBytes(req);                ctx.writeAndFlush(buff);            }        }
        @Override        protected void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {            String msgs = (String) msg;            System.out.println("服務(wù)器回復(fù)："+msgs+"；此時(shí)計(jì)數(shù)器counter:"+ ++counter);        }????}}

這里，主要用了

io.netty.handler.codec.DelimiterBasedFrameDecoder

代替前面的

io.netty.handler.codec.LineBasedFrameDecoder

上述代碼，說明，對于數(shù)據(jù)的截取，是根據(jù)"$_end"作為結(jié)束分隔符，

即遇到"$_end"，則截取數(shù)據(jù)，發(fā)送數(shù)據(jù)。

運(yùn)行上述代碼，其實(shí)際結(jié)果如下：

服務(wù)端（部分打印）：

客戶端（部分打印）：

至于

io.netty.handler.codec.DelimiterBasedFrameDecoder

則是根據(jù)指定長度的截取數(shù)據(jù)

因?yàn)楸容^簡單這里就不做詳述了

有興趣的可以自己做個(gè)實(shí)驗(yàn)

后續(xù)分享：Netty底層實(shí)現(xiàn)原理

有喜歡的煩請點(diǎn)個(gè)關(guān)注，謝謝！