Netty 是如何解決 TCP 粘包拆包的？-有解無憂

作者：rickiyang

出處：www.cnblogs.com/rickiyang/p/11074235.html

我們都知道TCP是基于位元組流的傳輸協議，

那么資料在通信層傳播其實就像河水一樣并沒有明顯的分界線，而資料具體表示什么意思什么地方有句號什么地方有分號這個對于TCP底層來說并不清楚，應用層向TCP層發送用于網間傳輸的、用8位位元組表示的資料流，然后TCP把資料流磁區成適當長度的報文段，之后TCP把結果包傳給IP層，由它來通過網路將包傳送給接收端物體的TCP層，

所以對于這個資料拆分成大包小包的問題就是我們今天要講的粘包和拆包的問題，

1、TCP粘包拆包問題說明

粘包和拆包這兩個概念估計大家還不清楚，通過下面這張圖我們來分析一下：

假設客戶端分別發送兩個資料包D1,D2個服務端，但是發送程序中資料是何種形式進行傳播這個并不清楚，分別有下列4種情況：

服務端一次接受到了D1和D2兩個資料包，兩個包粘在一起，稱為粘包；
服務端分兩次讀取到資料包D1和D2，沒有發生粘包和拆包；
服務端分兩次讀到了資料包，第一次讀到了D1和D2的部分內容，第二次讀到了D2的剩下部分，這個稱為拆包；
服務器分三次讀到了資料部分，第一次讀到了D1包，第二次讀到了D2包的部分內容，第三次讀到了D2包的剩下內容，

2、TCP粘包產生原因

我們知道在TCP協議中，應用資料分割成TCP認為最適合發送的資料塊，這部分是通過“MSS”（最大資料包長度）選項來控制的，通常這種機制也被稱為一種協商機制，MSS規定了TCP傳往另一端的最大資料塊的長度，這個值TCP協議在實作的時候往往用MTU值代替（需要減去IP資料包包頭的大小20Bytes和TCP資料段的包頭20Bytes）所以往往MSS為1460，通訊雙方會根據雙方提供的MSS值得最小值確定為這次連接的最大MSS值，

tcp為提高性能，發送端會將需要發送的資料發送到緩沖區，等待緩沖區滿了之后，再將緩沖中的資料發送到接收方，同理，接收方也有緩沖區這樣的機制，來接收資料，

發生粘包拆包的原因主要有以下這些：

應用程式寫入資料的位元組大小大于套接字發送緩沖區的大小將發生拆包；
進行MSS大小的TCP分段，MSS是TCP報文段中的資料欄位的最大長度，當TCP報文長度-TCP頭部長度>mss的時候將發生拆包；
應用程式寫入資料小于套接字緩沖區大小，網卡將應用多次寫入的資料發送到網路上,將發生粘包；
資料包大于MTU的時候將會進行切片，MTU即(Maxitum Transmission Unit) 最大傳輸單元,由于以太網傳輸電氣方面的限制，每個以太網幀都有最小的大小64bytes最大不能超過1518bytes,刨去以太網幀的幀頭14Bytes和幀尾CRC校驗部分4Bytes,那么剩下承載上層協議的地方也就是Data域最大就只能有1500Bytes這個值我們就把它稱之為MTU，這個就是網路層協議非常關心的地方，因為網路層協議比如IP協議會根據這個值來決定是否把上層傳下來的資料進行分片，

3、如何解決TCP粘包拆包

我們知道tcp是無界的資料流，且協議本身無法避免粘包，拆包的發生，那我們只能在應用層資料協議上，加以控制，通常在制定傳輸資料時，可以使用如下方法：

設定定長訊息，服務端每次讀取既定長度的內容作為一條完整訊息；
使用帶訊息頭的協議、訊息頭存盤訊息開始標識及訊息長度資訊，服務端獲取訊息頭的時候決議出訊息長度，然后向后讀取該長度的內容；
設定訊息邊界，服務端從網路流中按訊息邊界分離出訊息內容，比如在訊息末尾加上換行符用以區分訊息結束，

當然應用層還有更多復雜的方式可以解決這個問題，這個就屬于網路層的問題了，我們還是用java提供的方式來解決這個問題，我們先看一個例子看看粘包是如何發生的，

服務端：

public class HelloWordServer {
    private int port;

    public HelloWordServer(int port) {
        this.port = port;
    }

    public void start(){
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        EventLoopGroup workGroup = new NioEventLoopGroup();

        ServerBootstrap server = new ServerBootstrap().group(bossGroup,workGroup)
                                    .channel(NioServerSocketChannel.class)
                                    .childHandler(new ServerChannelInitializer());

        try {
            ChannelFuture future = server.bind(port).sync();
            future.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }finally {
            bossGroup.shutdownGracefully();
            workGroup.shutdownGracefully();
        }
    }

    public static void main(String[] args) {
        HelloWordServer server = new HelloWordServer(7788);
        server.start();
    }
}

服務端Initializer：

public class ServerChannelInitializer extends ChannelInitializer<SocketChannel> {
    @Override
    protected void initChannel(SocketChannel socketChannel) throws Exception {
        ChannelPipeline pipeline = socketChannel.pipeline();

        // 字串解碼 和 編碼
        pipeline.addLast("decoder", new StringDecoder());
        pipeline.addLast("encoder", new StringEncoder());

        // 自己的邏輯Handler
        pipeline.addLast("handler", new HelloWordServerHandler());
    }
}

服務端handler：

public class HelloWordServerHandler extends ChannelInboundHandlerAdapter {
    private int counter;

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        String body = (String)msg;
        System.out.println("server receive order : " + body + ";the counter is: " + ++counter);
    }

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

客戶端：

public class HelloWorldClient {
    private  int port;
    private  String address;

    public HelloWorldClient(int port,String address) {
        this.port = port;
        this.address = address;
    }

    public void start(){
        EventLoopGroup group = new NioEventLoopGroup();

        Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(group)
                .channel(NioSocketChannel.class)
                .handler(new ClientChannelInitializer());

        try {
            ChannelFuture future = bootstrap.connect(address,port).sync();         
            future.channel().closeFuture().sync();
        } catch (Exception e) {
            e.printStackTrace();
        }finally {
            group.shutdownGracefully();
        }

    }

    public static void main(String[] args) {
        HelloWorldClient client = new HelloWorldClient(7788,"127.0.0.1");
        client.start();
    }
}

客戶端Initializer：

public class ClientChannelInitializer extends  ChannelInitializer<SocketChannel> {

    protected void initChannel(SocketChannel socketChannel) throws Exception {
        ChannelPipeline pipeline = socketChannel.pipeline();

        pipeline.addLast("decoder", new StringDecoder());
        pipeline.addLast("encoder", new StringEncoder());

        // 客戶端的邏輯
        pipeline.addLast("handler", new HelloWorldClientHandler());
    }
}

客戶端handler：

public class HelloWorldClientHandler extends ChannelInboundHandlerAdapter {
    private byte[] req;
    private int counter;

    public BaseClientHandler() {
        req = ("Unless required by applicable law or agreed to in writing, software\n" +
                "  distributed under the License is distributed on an \"AS IS\" BASIS,\n" +
                "  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n" +
                "  See the License for the specific language governing permissions and\n" +
                "  limitations under the License.This connector uses the BIO implementation that requires the JSSE\n" +
                "  style configuration. When using the APR/native implementation, the\n" +
                "  penSSL style configuration is required as described in the APR/native\n" +
                "  documentation.An Engine represents the entry point (within Catalina) that processes\n" +
                "  every request.  The Engine implementation for Tomcat stand alone\n" +
                "  analyzes the HTTP headers included with the request, and passes them\n" +
                "  on to the appropriate Host (virtual host)# Unless required by applicable law or agreed to in writing, software\n" +
                "# distributed under the License is distributed on an \"AS IS\" BASIS,\n" +
                "# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n" +
                "# See the License for the specific language governing permissions and\n" +
                "# limitations under the License.# For example, set the org.apache.catalina.util.LifecycleBase logger to log\n" +
                "# each component that extends LifecycleBase changing state:\n" +
                "#org.apache.catalina.util.LifecycleBase.level = FINE"
                ).getBytes();
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        ByteBuf message;

        //將上面的所有字串作為一個訊息體發送出去
        message = Unpooled.buffer(req.length);
        message.writeBytes(req);
        ctx.writeAndFlush(message);
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        String buf = (String)msg;
        System.out.println("Now is : " + buf + " ; the counter is : "+ (++counter));
    }

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

運行客戶端和服務端我們能看到：

我們看到這個長長的字串被截成了2段發送，這就是發生了拆包的現象，同樣粘包我們也很容易去模擬，我們把BaseClientHandler中的channelActive方法里面的：

message = Unpooled.buffer(req.length);
message.writeBytes(req);
ctx.writeAndFlush(message);

這幾行代碼是把我們上面的一長串字符轉成的byte陣列寫進流里發送出去，那么我們可以在這里把上面發送訊息的這幾行回圈幾遍這樣發送的內容增多了就有可能在拆包的時候把上一條訊息的一部分分配到下一條訊息里面了，修改如下：

for (int i = 0; i < 3; i++) {
    message = Unpooled.buffer(req.length);
    message.writeBytes(req);
    ctx.writeAndFlush(message);
}

改完之后我們再運行一下，輸出太長不好截圖，我們在輸出結果中能看到回圈3次之后的訊息服務端收到的就不是之前的完整的一條了，而是被拆分了4次發送，

對于上面出現的粘包和拆包的問題，Netty已有考慮，并且有實施的方案：LineBasedFrameDecoder，
我們重新改寫一下ServerChannelInitializer：

public class ServerChannelInitializer extends ChannelInitializer<SocketChannel> {
    @Override
    protected void initChannel(SocketChannel socketChannel) throws Exception {
        ChannelPipeline pipeline = socketChannel.pipeline();


        pipeline.addLast(new LineBasedFrameDecoder(2048));       
        // 字串解碼 和 編碼
        pipeline.addLast("decoder", new StringDecoder());
        pipeline.addLast("encoder", new StringEncoder());

        // 自己的邏輯Handler
        pipeline.addLast("handler", new BaseServerHandler());
    }
}

新增：pipeline.addLast(new LineBasedFrameDecoder(2048))，同時，我們還得對上面發送的訊息進行改造BaseClientHandler：

public class BaseClientHandler extends ChannelInboundHandlerAdapter {
    private byte[] req;
    private int counter;

    req = ("Unless required by applicable dfslaw or agreed to in writing, software" +
                "  distributed under the License is distributed on an \"AS IS\" BASIS," +
                "  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied." +
                "  See the License for the specific language governing permissions and" +
                "  limitations under the License.This connector uses the BIO implementation that requires the JSSE" +
                "  style configuration. When using the APR/native implementation, the" +
                "  penSSL style configuration is required as described in the APR/native" +
                "  documentation.An Engine represents the entry point (within Catalina) that processes" +
                "  every request.  The Engine implementation for Tomcat stand alone" +
                "  analyzes the HTTP headers included with the request, and passes them" +
                "  on to the appropriate Host (virtual host)# Unless required by applicable law or agreed to in writing, software" +
                "# distributed under the License is distributed on an \"AS IS\" BASIS," +
                "# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied." +
                "# See the License for the specific language governing permissions and" +
                "# limitations under the License.# For example, set the org.apache.catalina.util.LifecycleBase logger to log" +
                "# each component that extends LifecycleBase changing state:" +
                "#org.apache.catalina.util.LifecycleBase.level = FINE\n"
                ).getBytes();  


    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        ByteBuf message;

        message = Unpooled.buffer(req.length);
        message.writeBytes(req);
        ctx.writeAndFlush(message);
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        String buf = (String)msg;
        System.out.println("Now is : " + buf + " ; the counter is : "+ (++counter));
    }

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

去掉所有的”\n”,只保留字串末尾的這一個，原因稍后再說，channelActive方法中我們不必再用回圈多次發送訊息了，只發送一次就好(第一個例子中發送一次的時候是發生了拆包的)，然后我們再次運行，大家會看到這么長一串字符只發送了一串就發送完畢，程式輸出我就不截圖了，下面來解釋一下LineBasedFrameDecoder，

LineBasedFrameDecoder的作業原理是它依次遍歷ByteBuf 中的可讀位元組，判斷看是否有”\n” 或者” \r\n”，如果有，就以此位置為結束位置，從可讀索引到結束位置區間的位元組就組成了一行，它是以換行符為結束標志的解碼器，支持攜帶結束符或者不攜帶結束符兩種解碼方式，同時支持配置單行的最大長度，如果連續讀取到最大長度后仍然沒有發現換行符，就會拋出例外，同時忽略掉之前讀到的例外碼流，這個對于我們確定訊息最大長度的應用場景還是很有幫助，

對于上面的判斷看是否有”\n” 或者” \r\n”以此作為結束的標志我們可能回想，要是沒有”\n” 或者” \r\n”那還有什么別的方式可以判斷訊息是否結束呢，別擔心，Netty對于此已經有考慮，還有別的解碼器可以幫助我們解決問題，

近期熱文推薦：

1.1,000+ 道 Java面試題及答案整理(2021最新版)

2.終于靠開源專案弄到 IntelliJ IDEA 激活碼了，真香！

3.阿里 Mock 工具正式開源，干掉市面上所有 Mock 工具！

4.Spring Cloud 2020.0.0 正式發布，全新顛覆性版本！

5.《Java開發手冊（嵩山版）》最新發布，速速下載！

覺得不錯，別忘了隨手點贊+轉發哦！

轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/288931.html

標籤：Java

上一篇：上海某大公司：你是了解Redis對吧？

下一篇：Python學習（1）-基礎語法學習丨【生長吧！Python】