Netty 是如何解決半包和粘包問題?

作者：猿java 2024-08-16 21:47:18

Netty 是一個高性能、異步事件驅(qū)動的網(wǎng)絡(luò)應(yīng)用框架，廣泛應(yīng)用于各種網(wǎng)絡(luò)通信場景。這篇文章，我們將詳細(xì)分析 Netty 是如何解決半包和粘包問題。

一、什么是半包和粘包？

1.半包問題

半包問題是指一個完整的應(yīng)用層消息被分成多個 TCP 數(shù)據(jù)包發(fā)送，接收端在一次讀取操作中只接收到消息的一部分。

例如，發(fā)送端發(fā)送了一條 100 字節(jié)的消息，但由于網(wǎng)絡(luò)原因，這條消息被拆分成了兩個 TCP 數(shù)據(jù)包，一個 60 字節(jié)，另一個 40 字節(jié)。接收端可能在第一次讀取時只接收到前 60 字節(jié)的數(shù)據(jù)，剩下的 40 字節(jié)需要在后續(xù)的讀取操作中才能接收到。

2.粘包問題

粘包問題是指多個應(yīng)用層消息在傳輸過程中被粘在一起，接收端在一次讀取操作中接收到大于 1個消息的情況。

例如，發(fā)送端發(fā)送了兩條消息，每條 50 字節(jié)，但接收端在一次讀取操作中收到了 80 字節(jié)的數(shù)據(jù)，超過了 1條消息的內(nèi)容。

3.產(chǎn)生原因

產(chǎn)生半包和粘包問題主要是以下 3個原因：

TCP 的流式特性：TCP 是面向字節(jié)流的協(xié)議，沒有消息邊界的概念，它保證數(shù)據(jù)的順序和可靠性，但不保證每次發(fā)送的數(shù)據(jù)對應(yīng)每次接收的數(shù)據(jù)。
網(wǎng)絡(luò)狀況：網(wǎng)絡(luò)的擁塞、延遲、抖動等因素可能導(dǎo)致數(shù)據(jù)包的拆分和重組。
操作系統(tǒng)和緩沖區(qū)：操作系統(tǒng) TCP/IP 協(xié)議棧和應(yīng)用程序的緩沖區(qū)大小也會影響數(shù)據(jù)的讀取方式。

4.示例

假設(shè)發(fā)送端發(fā)送了兩條消息：

消息1：Hello
消息2：World

在半包情況下，接收端可能會這樣接收：

第一次讀取：Hel
第二次讀取：loWo
第三次讀取：rld

在粘包情況下，接收端可能會這樣接收：

第一次讀取：HelloWor
第二次讀取：ld

二、解決方案

1.基于固定長度的解碼器

基于固定長度的解碼器是指發(fā)消息時，每條消息的長度固定，讀消息時也通過固定長度來讀取消息，從而解決半包和粘包問題。

(1) 實(shí)現(xiàn)方式

Netty 提供了 FixedLengthFrameDecoder 類來實(shí)現(xiàn)這一功能，核心源碼如下：

public class FixedLengthFrameDecoder extends ByteToMessageDecoder {
private final int frameLength;

    public FixedLengthFrameDecoder(int frameLength) {
        this.frameLength = frameLength;
    }

    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
        while (in.readableBytes() >= frameLength) {
            ByteBuf buf = in.readBytes(frameLength);
            out.add(buf);
        }
    }
}

(2) 注意點(diǎn)

使用定長幀需要注意以下幾點(diǎn)：

固定長度：消息長度必須是固定的，發(fā)送端需要確保消息長度一致。如果長度超出固定長度，解包時消息就會錯位，如果消息不足固定長度，需要使用填充字符補(bǔ)齊。
填充字符：選擇合適的填充字符（如空格）來補(bǔ)齊消息長度，接收端在處理時需要去除這些填充字符。

(3) 優(yōu)點(diǎn)

簡單易實(shí)現(xiàn)：實(shí)現(xiàn)起來非常簡單，不需要額外的頭部信息或分隔符。
解析效率高：由于每個消息長度固定，接收端解析時只需按照固定長度讀取。

(4) 缺點(diǎn)

不靈活：消息長度固定，可能會造成空間浪費(fèi)（如果消息長度較短）或不足（如果消息長度較長）。
適用場景有限：適用于固定格式和長度的協(xié)議，不適用于可變長度消息的場景。

(5) 示例

下面我們通過一個示例來展示使用定長幀是如何解決半包粘包問題的。

發(fā)送端，確保每個消息的長度固定。如果實(shí)際消息長度不足，可以使用填充字符（如空格）來補(bǔ)齊。

public class FixedLengthFrameSender {

    private static final int FRAME_LENGTH = 10; // 固定消息長度

    public static void send(Channel channel, String message) {
        // 確保消息長度不超過固定長度
        if (message.length() > FRAME_LENGTH) {
            throw new IllegalArgumentException("Message too long");
        }
        // 使用空格填充消息到固定長度
        String paddedMessage = String.format("%-" + FRAME_LENGTH + "s", message);
        
        // 將消息轉(zhuǎn)換為字節(jié)數(shù)組并發(fā)送
        ByteBuf buffer = Unpooled.copiedBuffer(paddedMessage.getBytes());
        channel.writeAndFlush(buffer);
    }
}

接收端，使用 Netty 提供的 FixedLengthFrameDecoder 解碼器來處理固定長度的消息。

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.handler.codec.FixedLengthFrameDecoder;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;

public class FixedLengthFrameReceiver {
    private static final int FRAME_LENGTH = 10; // 固定消息長度

    public static void main(String[] args) throws Exception {
        NioEventLoopGroup bossGroup = new NioEventLoopGroup(1);
        NioEventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workerGroup)
             .channel(NioServerSocketChannel.class)
             .childHandler(new ChannelInitializer<SocketChannel>() {
                 @Override
                 public void initChannel(SocketChannel ch) throws Exception {
                     ChannelPipeline p = ch.pipeline();
                     // 添加定長幀解碼器
                     p.addLast(new FixedLengthFrameDecoder(FRAME_LENGTH));
                     // 添加自定義處理器
                     p.addLast(new FixedLengthFrameHandler());
                 }
             });
            // 啟動服務(wù)器
            b.bind(8888).sync().channel().closeFuture().sync();
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }

    public static class FixedLengthFrameHandler extends ChannelInboundHandlerAdapter {
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) {
            ByteBuf in = (ByteBuf) msg;
            byte[] receivedBytes = new byte[in.readableBytes()];
            in.readBytes(receivedBytes);
            String receivedMsg = new String(receivedBytes).trim(); // 去除填充字符
            System.out.println("Received: " + receivedMsg);
        }
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
            cause.printStackTrace();
            ctx.close();
        }
    }
}

2.基于換行符解碼器

3.自定義分隔符解碼器

基于換行符解碼器和自定義分隔符解碼器（比如特殊字符）來劃分消息邊界，從而解決半包和粘包問題，使用者可以根據(jù)自己的需求靈活確定分隔符。

(1) 實(shí)現(xiàn)方式

Netty 提供了 DelimiterBasedFrameDecoder 類來實(shí)現(xiàn)這一功能，核心源碼如下：

public DelimiterBasedFrameDecoder(
        int maxFrameLength, boolean stripDelimiter, boolean failFast, ByteBuf... delimiters) {
   validateMaxFrameLength(maxFrameLength);
   ObjectUtil.checkNonEmpty(delimiters, "delimiters");

   if (isLineBased(delimiters) && !isSubclass()) {
      lineBasedDecoder = new LineBasedFrameDecoder(maxFrameLength, stripDelimiter, failFast);
      this.delimiters = null;
   } else {
      this.delimiters = new ByteBuf[delimiters.length];
      for (int i = 0; i < delimiters.length; i ++) {
         ByteBuf d = delimiters[i];
         validateDelimiter(d);
         this.delimiters[i] = d.slice(d.readerIndex(), d.readableBytes());
      }
      lineBasedDecoder = null;
   }
   this.maxFrameLength = maxFrameLength;
   this.stripDelimiter = stripDelimiter;
   this.failFast = failFast;
}

(2) 注意點(diǎn)

分隔符選擇：選擇一個不會出現(xiàn)在消息內(nèi)容中的分隔符（如換行符 \n 或特定字符 |）。
消息格式：發(fā)送端在每個消息的末尾添加分隔符，確保接收端能夠正確解析消息邊界。

(3) 優(yōu)點(diǎn)

靈活性高：可以處理可變長度的消息。
實(shí)現(xiàn)相對簡單：只需在消息末尾添加特定的分隔符，接收端根據(jù)分隔符拆分消息。

(4) 缺點(diǎn)

分隔符沖突：如果消息內(nèi)容中包含分隔符，可能導(dǎo)致解析錯誤，需要對消息內(nèi)容進(jìn)行轉(zhuǎn)義處理。
解析效率低：需要掃描整個數(shù)據(jù)流尋找分隔符，效率較低。

(5) 示例

下面我們通過一個示例來展示使用分隔符是如何解決半包粘包問題的。

發(fā)送端，確保每個消息以特定的分隔符結(jié)尾。常用的分隔符包括換行符（\n）、特定字符（如 |）等。

public class DelimiterBasedFrameSender {

    private static final String DELIMITER = "\n"; // 分隔符

    public static void send(Channel channel, String message) {
        // 在消息末尾添加分隔符
        String delimitedMessage = message + DELIMITER;
        
        // 將消息轉(zhuǎn)換為字節(jié)數(shù)組并發(fā)送
        ByteBuf buffer = Unpooled.copiedBuffer(delimitedMessage.getBytes());
        channel.writeAndFlush(buffer);
    }
}

接收端，使用 Netty 提供的 DelimiterBasedFrameDecoder 解碼器來處理以分隔符結(jié)尾的消息。

import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
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.string.StringDecoder;

public class DelimiterBasedFrameReceiver {

    private static final String DELIMITER = "\n"; // 分隔符
    private static final int MAX_FRAME_LENGTH = 1024; // 最大幀長度

    public static void main(String[] args) throws Exception {
        NioEventLoopGroup bossGroup = new NioEventLoopGroup(1);
        NioEventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workerGroup)
             .channel(NioServerSocketChannel.class)
             .childHandler(new ChannelInitializer<SocketChannel>() {
                 @Override
                 public void initChannel(SocketChannel ch) throws Exception {
                     ChannelPipeline p = ch.pipeline();
                     // 添加分隔符解碼器
                     ByteBuf delimiter = Unpooled.copiedBuffer(DELIMITER.getBytes());
                     p.addLast(new DelimiterBasedFrameDecoder(MAX_FRAME_LENGTH, delimiter));
                     // 添加字符串解碼器
                     p.addLast(new StringDecoder());
                     // 添加自定義處理器
                     p.addLast(new DelimiterBasedFrameHandler());
                 }
             });

            // 啟動服務(wù)器
            b.bind(8888).sync().channel().closeFuture().sync();
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }

    public static class DelimiterBasedFrameHandler extends ChannelInboundHandlerAdapter {
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) {
            String receivedMsg = (String) msg;
            System.out.println("Received: " + receivedMsg);
        }

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

4.基于長度字段的解碼器

基于長度字段的解碼器是指在消息頭部添加長度字段，指示消息的總長度。

(1) 實(shí)現(xiàn)方式

Netty 提供了 LengthFieldBasedFrameDecoder 類來實(shí)現(xiàn)這一功能，核心源碼如下：

public class LengthFieldBasedFrameDecoder extends ByteToMessageDecoder {
private final int maxFrameLength;
private final int lengthFieldOffset;
private final int lengthFieldLength;

    public LengthFieldBasedFrameDecoder(int maxFrameLength, int lengthFieldOffset, int lengthFieldLength) {
        this.maxFrameLength = maxFrameLength;
        this.lengthFieldOffset = lengthFieldOffset;
        this.lengthFieldLength = lengthFieldLength;
    }

    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
        if (in.readableBytes() < lengthFieldOffset + lengthFieldLength) {
            return;
        }

        in.markReaderIndex();
        int length = in.getInt(in.readerIndex() + lengthFieldOffset);
        if (in.readableBytes() < lengthFieldOffset + lengthFieldLength + length) {
            in.resetReaderIndex();
            return;
        }

        in.skipBytes(lengthFieldOffset + lengthFieldLength);
        ByteBuf frame = in.readBytes(length);
        out.add(frame);
    }
}

(2) 關(guān)鍵點(diǎn)

長度字段位置：長度字段通常位于消息的頭部，用于指示消息的總長度。

解碼器參數(shù)：

maxFrameLength：消息的最大長度，防止內(nèi)存溢出。
lengthFieldOffset：長度字段在消息中的偏移量。
lengthFieldLength：長度字段的字節(jié)數(shù)（通常為 4 字節(jié)）。
lengthAdjustment：長度調(diào)整值，如果長度字段不包含消息頭的長度，需要進(jìn)行調(diào)整。
initialBytesToStrip：解碼后跳過的字節(jié)數(shù)，通常為長度字段的長度。

(3) 優(yōu)點(diǎn)

靈活性高：支持可變長度的消息。
解析效率高：通過長度字段可以直接讀取完整消息，無需掃描整個數(shù)據(jù)流。

(4) 缺點(diǎn)

實(shí)現(xiàn)復(fù)雜：需要在消息頭部添加長度字段，接收端需要解析頭部信息。
額外開銷：消息頭部的長度字段會增加一些額外的字節(jié)數(shù)。

(5) 示例

下面我們通過一個示例來展示使用長度字段是如何解決半包粘包問題的。

發(fā)送端，確保每個消息在發(fā)送前都包含長度字段。長度字段通常放在消息的頭部，用于指示消息的總長度。

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;

public class LengthFieldBasedFrameSender {

    public static void send(Channel channel, String message) {
        // 將消息轉(zhuǎn)換為字節(jié)數(shù)組
        byte[] messageBytes = message.getBytes();
        int messageLength = messageBytes.length;

        // 創(chuàng)建一個 ByteBuf 來存儲長度字段和消息內(nèi)容
        ByteBuf buffer = Unpooled.buffer(4 + messageLength);

        // 寫入長度字段（4 字節(jié)，表示消息長度）
        buffer.writeInt(messageLength);

        // 寫入消息內(nèi)容
        buffer.writeBytes(messageBytes);

        // 發(fā)送消息
        channel.writeAndFlush(buffer);
    }
}

接收端，使用 Netty 提供的 LengthFieldBasedFrameDecoder 解碼器來處理包含長度字段的消息。

import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.codec.string.StringDecoder;

public class LengthFieldBasedFrameReceiver {

    private static final int MAX_FRAME_LENGTH = 1024; // 最大幀長度

    public static void main(String[] args) throws Exception {
        NioEventLoopGroup bossGroup = new NioEventLoopGroup(1);
        NioEventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workerGroup)
             .channel(NioServerSocketChannel.class)
             .childHandler(new ChannelInitializer<SocketChannel>() {
                 @Override
                 public void initChannel(SocketChannel ch) throws Exception {
                     ChannelPipeline p = ch.pipeline();
                     // 添加長度字段解碼器
                     p.addLast(new LengthFieldBasedFrameDecoder(
                         MAX_FRAME_LENGTH, 0, 4, 0, 4));
                     // 添加字符串解碼器
                     p.addLast(new StringDecoder());
                     // 添加自定義處理器
                     p.addLast(new LengthFieldBasedFrameHandler());
                 }
             });

            // 啟動服務(wù)器
            b.bind(8888).sync().channel().closeFuture().sync();
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }

    public static class LengthFieldBasedFrameHandler extends ChannelInboundHandlerAdapter {
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) {
            String receivedMsg = (String) msg;
            System.out.println("Received: " + receivedMsg);
        }

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

5. 自定義解碼器

如果上述 Netty提供的方案無法滿足業(yè)務(wù)需求的話，Netty還提供了一個擴(kuò)展點(diǎn)，使用者可以通過自定義解碼器來處理消息，

(1) 實(shí)現(xiàn)方式

例如，自定義頭部信息來表示消息長度或結(jié)束標(biāo)志，示例代碼如下：

public class CustomProtocolDecoder extends ByteToMessageDecoder {
    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
        // 根據(jù)自定義協(xié)議解析消息
        if (in.readableBytes() < 4) {
            return;
        }

        in.markReaderIndex();
        int length = in.readInt();
        if (in.readableBytes() < length) {
            in.resetReaderIndex();
            return;
        }

        ByteBuf frame = in.readBytes(length);
        out.add(frame);
    }
}

(2) 優(yōu)點(diǎn)