碼上快樂

相關內容    簡體    繁體

深入學習Netty（5）——Netty是如何解決TCP粘包/拆包問題的？

本文轉載自   查看原文   2021-07-27 09:09   482    Netty/ Netty編程

前言

　　學習Netty避免不了要去了解TCP粘包/拆包問題，熟悉各個編解碼器是如何解決TCP粘包/拆包問題的，同時需要知道TCP粘包/拆包問題是怎么產生的。

　　在此博文前，可以先學習了解前幾篇博文：

　　參考資料《Netty In Action》、《Netty權威指南》（有需要的小伙伴可以評論或者私信我）

　　博文中所有的代碼都已上傳到Github，歡迎Star、Fork

 

 

一、TCP粘包/拆包

1.什么是TCP粘包/拆包問題？

引用《Netty權威指南》原話，可以很清楚解釋什么是TCP粘包/拆包問題。

　　TCP是一個“流”協議，是沒有界限的一串數據，TCP底層並不了解上層業務數據的具體含義，它會根據TCP緩沖區的實際情況進行包的划分，所以在業務上認為，一個完整的包可能會被TCP拆成多個包進行發送，也有可能把多個小的包封裝成一個大的數據包發送，這就是所謂的TCP粘包和拆包問題。

　　一個完整的包可能會被TCP拆分成多個包進行發送，也有可能把多個小的包封裝成一個大的數據包發送，這就是TCP粘包/拆包。

假設服務端分別發送兩個數據包P1和P2給服務端，由於服務端讀取一次的字節數目是不確定的，所以可能會發生五種情況：

              

  • 服務端分兩次讀取到兩個獨立的數據包；
  • 服務端一次接收到兩個數據包，P1和P2粘合在一起，被稱為TCP粘包；
  • 服務端分兩次讀取到兩個數據包，第一次讀取到完整的P1包和P2包的部分內容，第二次讀取到P2包的剩余內容，被稱之為TCP拆包；
  • 服務端分兩次讀取到兩個數據包，第一次讀取到了P1包的部分內容P1_1，第二次讀取到了P1包的剩余內容P1_2和P2包的整包
  • 其實還有最后一種可能，就是服務端TCP接收的滑動窗非常小，而數據包P1/P2非常大，很有可能服務端需要分多次才能將P1/P2包接收完全，期間發生多次拆包。

2.TCP粘包/拆包問題發生的原因

　TCP是以流動的方式傳輸數據，傳輸的最小單位為一個報文段（segment）。主要有如下幾個指標影響或造成TCP粘包/拆包問題，分別為MSS、MTU、緩沖區，以及Nagle算法的影響。

（1）MSS(Maximum Segment Size)指的是連接層每次傳輸的數據有個最大限制MTU(Maximum Transmission Unit)，超過這個量要分成多個報文段。

（2）MTU限制了一次最多可以發送1500個字節，而TCP協議在發送DATA時，還會加上額外的TCP Header和IP Header，因此刨去這兩個部分，就是TCP協議一次可以發送的實際應用數據的最大大小，即MSS長度=MTU長度-IP Header-TCP Header。

（3）TCP為提高性能，發送端會將需要發送的數據發送到緩沖區，等待緩沖區滿了之后，再將緩沖中的數據發送到接收方。同理，接收方也有緩沖區這樣的機制，來接收數據。

由於有上述的原因，所以會造成拆包/粘包的具體原因如下：

（1）拆包發生原因

  • 要發送的數據大於TCP發送緩沖區剩余空間大小，將會發生拆包。
  • 待發送數據大於MSS（最大報文長度），TCP在傳輸前將進行拆包。

（2）粘包發生原因

  • 要發送的數據小於TCP發送緩沖區的大小，TCP將多次寫入緩沖區的數據一次發送出去，將會發生粘包。
  • 接收數據端的應用層沒有及時讀取接收緩沖區中的數據，將發生粘包。

二、TCP粘包/拆包問題解決策略

1.常用的解決策略

由於底層TCP是無法理解上層業務數據，所以在底層是無法保證數據包不被拆分和重組的，所以只能通過上層應用協議棧設計來解決

（1）消息定長，例如每個報文的大小固定長度200字節，不夠空位補空格

（2）在包尾增加回車換行符進行分割，例如FTP協議

（3）將消息分為消息頭和消息體，消息頭中包含表示消息總長度的字段

（4）更復雜的應用層協議。

2.TCP粘包異常問題案例

（1）TimeServerHandler

public class TimeServerHandler extends ChannelInboundHandlerAdapter {
    private static final Logger log = Logger.getLogger(TimeClientHandler.class.getName());


    private int counter;

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        ByteBuf buf = (ByteBuf) msg;
        byte[] req = new byte[buf.readableBytes()];
        buf.readBytes(req);
        String body = new String(req, "UTF-8").substring(0, req.length - System.getProperty("line.separator").length());
        // 每收到一條消息計數器就加1, 理論上應該接收到100條
        System.out.println("The time server receive order: " + body + "; the counter is : "+ (++counter));
        String currentTime = "QUERY TIME ORDER".equalsIgnoreCase(body) ?
                new Date(System.currentTimeMillis()).toString():"BAD ORDER";
        currentTime = currentTime + System.getProperty("line.separator");
        ByteBuf resp = Unpooled.copiedBuffer(currentTime.getBytes());
        ctx.writeAndFlush(resp);
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        log.warning("Unexpected exception from downstream: " + cause.getMessage());
        ctx.close();
    }
}

（3）TimeServer

public class TimeServer {

    public static final Logger log = LoggerFactory.getLogger(TimeServer.class);

    public static void main(String[] args) throws Exception {
        new TimeServer().bind();
    }


    public void bind() throws Exception {
        // NIO 線程組
        NioEventLoopGroup bossGroup = new NioEventLoopGroup();
        NioEventLoopGroup workerGroup = new NioEventLoopGroup();

        try {
            ServerBootstrap bootstrap = new ServerBootstrap();
            bootstrap.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .option(ChannelOption.SO_BACKLOG, 1024)
                    .handler(new LoggingHandler(LogLevel.INFO))
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel socketChannel) {
                            socketChannel.pipeline().addLast(new TimeServerHandler());
                        }
                    });
            // 綁定端口，同步等待成功
            ChannelFuture f = bootstrap.bind(NettyConstant.REMOTE_IP, NettyConstant.REMOTE_PORT).sync();
            log.info("Time server[{}] start success", NettyConstant.REMOTE_IP + ": " + NettyConstant.REMOTE_PORT);
            // 等待所有服務端監聽端口關閉
            f.channel().closeFuture().sync();
        } finally {
            // 優雅退出，釋放線程池資源
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }

    }

}

（3）TimeClientHandler

public class TimeClientHandler extends ChannelInboundHandlerAdapter {

    private static final Logger log = Logger.getLogger(TimeClientHandler.class.getName());

    private int counter;
    private byte[] req;

    public TimeClientHandler() {
        req = ("QUERY TIME ORDER" + System.getProperty("line.separator"))
                .getBytes();
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        ByteBuf message = null;
        // 循環發送100條消息，每發送一條刷新一次，服務端理論上接收到100條查詢時間指令的請求
        for (int i = 0; i < 100; i++) {
            message = Unpooled.buffer(req.length);
            message.writeBytes(req);
            ctx.writeAndFlush(message);
        }
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        ByteBuf buf = (ByteBuf) msg;
        byte[] req = new byte[buf.readableBytes()];
        buf.readBytes(req);
        String body = new String(req, "UTF-8");
        // 客戶端每接收到服務端一條應答消息之后，計數器就加1，理論上應該有100條服務端日志
        System.out.println("Now is: " + body + "; the current is "+ (++counter));
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        log.warning("Unexpected exception from downstream: " + cause.getMessage());
        ctx.close();
    }
}

（4）TimeClient

public class TimeClient {

    public static final Logger log = LoggerFactory.getLogger(TimeClient.class);



    public static void main(String[] args) throws Exception {
        new TimeClient().connect(NettyConstant.REMOTE_IP, NettyConstant.REMOTE_PORT);
    }

    public void connect(final String host, final int port) throws Exception {
        // NIO 線程組
        EventLoopGroup group = new NioEventLoopGroup();
        try {
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(group)
                    .channel(NioSocketChannel.class)
                    .option(ChannelOption.TCP_NODELAY, true)
                    .handler(new LoggingHandler(LogLevel.INFO))
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            socketChannel.pipeline().addLast(new TimeClientHandler());
                        }
                    });

            // 發起異步連接操作
            ChannelFuture f = bootstrap.connect(host, port).sync();
            // 等待所有服務端監聽端口關閉
            f.channel().closeFuture().sync();
        } finally {
            // 優雅退出，釋放線程池資源
            group.shutdownGracefully();
        }

    }

}

（5）運行測試結果

運行服務端與客戶端，觀察服務端與客戶端

服務端：

The time server receive order: QUERY TIME ORDER
QUERY TIME ORDER
... // 此處忽略96個QUERY TIME ORDER
QUERY TIME ORDER
QUERY TIME ORDER; the counter is : 1

客戶端：

Now is: BAD ORDER
; the current is 1

從結果上來看，客戶端向服務端發送的100個“QUERY TIME ORDER”命令，都粘成一個包（counter=1），服務端也只返回一個命令“BAD ORDER”，可以嘗試運行客戶端多次，每次運行的結果都是不一樣的，但是大部分都是粘包，計數器都小於了100。

 

三、Netty解決TCP粘包/拆包

1.按行文本解碼器LineBasedFramedDecoder和StringDecoder

　　LineBasedFramedDecoder：依次遍歷ByeBuf中可讀字節，判斷是否有“\n”，“\r\n”，如果有，就當前位置為結束位置，從可讀索引到結束位置區間的字節就組裝成一行，以換行符為結束標志的解碼器，同識支持最大長度。

　　StringDecoder：將接收對象轉換成字符串，然后繼續調用后面的handler。

　　LineBasedFramedDecoder和StringDecoder就是按行切換的文本解碼器，被設計用來支持TCP粘包與拆包。

（1）改造TimeServer

　　增加解碼器LineBasedFramedDecoder和StringDecoder　

public class TimeServer {

    public static final Logger log = LoggerFactory.getLogger(TimeServer.class);

    public static void main(String[] args) throws Exception {
        new TimeServer().bind();
    }


    public void bind() throws Exception {
        // NIO 線程組
        NioEventLoopGroup bossGroup = new NioEventLoopGroup();
        NioEventLoopGroup workerGroup = new NioEventLoopGroup();

        try {
            ServerBootstrap bootstrap = new ServerBootstrap();
            bootstrap.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .option(ChannelOption.SO_BACKLOG, 1024)
                    .handler(new LoggingHandler(LogLevel.INFO))
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel socketChannel) {
                            socketChannel.pipeline().addLast(new LineBasedFrameDecoder(1024));
                            socketChannel.pipeline().addLast(new StringDecoder());
                            socketChannel.pipeline().addLast(new TimeServerHandler());
                        }
                    });
            // 綁定端口，同步等待成功
            ChannelFuture f = bootstrap.bind(NettyConstant.LOCAL_IP, NettyConstant.LOCAL_PORT).sync();
            log.info("Time server[{}] start success", NettyConstant.LOCAL_IP + ": " + NettyConstant.LOCAL_PORT);
            // 等待所有服務端監聽端口關閉
            f.channel().closeFuture().sync();
        } finally {
            // 優雅退出，釋放線程池資源
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }

    }

}

（2）改造TimeServerHandler

　　不需要對消息進行解碼，直接String讀取即可

public class TimeServerHandler extends ChannelInboundHandlerAdapter {
    private static final Logger log = Logger.getLogger(TimeClientHandler.class.getName());


    private int counter;

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        // 不需要對消息進行編解碼，直接String讀取
        String body = (String) msg;
        // 每收到一條消息計數器就加1, 理論上應該接收到100條
        System.out.println("The time server receive order: " + body + "; the counter is : "+ (++counter));
        String currentTime = "QUERY TIME ORDER".equalsIgnoreCase(body) ?
                new Date(System.currentTimeMillis()).toString():"BAD ORDER";
        currentTime = currentTime + System.getProperty("line.separator");
        ByteBuf resp = Unpooled.copiedBuffer(currentTime.getBytes());
        ctx.writeAndFlush(resp);
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        log.warning("Unexpected exception from downstream: " + cause.getMessage());
        ctx.close();
    }
}

（3）改造TimeClient

　　同樣增加解碼器LineBasedFramedDecoder和StringDecoder

public class TimeClient {

    public static final Logger log = LoggerFactory.getLogger(TimeClient.class);



    public static void main(String[] args) throws Exception {
        new TimeClient().connect(NettyConstant.LOCAL_IP, NettyConstant.LOCAL_PORT);
    }

    public void connect(final String host, final int port) throws Exception {
        // NIO 線程組
        EventLoopGroup group = new NioEventLoopGroup();
        try {
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(group)
                    .channel(NioSocketChannel.class)
                    .option(ChannelOption.TCP_NODELAY, true)
                    .handler(new LoggingHandler(LogLevel.INFO))
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            socketChannel.pipeline().addLast(new LineBasedFrameDecoder(1024));
                            socketChannel.pipeline().addLast(new StringDecoder());
                            socketChannel.pipeline().addLast(new TimeClientHandler());
                        }
                    });

            // 發起異步連接操作
            ChannelFuture f = bootstrap.connect(host, port).sync();
            // 等待所有服務端監聽端口關閉
            f.channel().closeFuture().sync();
        } finally {
            // 優雅退出，釋放線程池資源
            group.shutdownGracefully();
        }

    }

}

（4）改造TimeClientHandler

　　同樣地，不需要編解碼了，直接返回了字符串的應答消息

public class TimeClientHandler extends ChannelInboundHandlerAdapter {

    private static final Logger log = Logger.getLogger(TimeClientHandler.class.getName());

    private int counter;
    private byte[] req;

    public TimeClientHandler() {
        req = ("QUERY TIME ORDER" + System.getProperty("line.separator"))
                .getBytes();
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        ByteBuf message = null;
        // 循環發送100條消息，每發送一條刷新一次，服務端理論上接收到100條查詢時間指令的請求
        for (int i = 0; i < 100; i++) {
            message = Unpooled.buffer(req.length);
            message.writeBytes(req);
            ctx.writeAndFlush(message);
        }
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        // 不需要編解碼了，直接返回了字符串的應答消息
        String body = (String) msg;
        // 客戶端每接收到服務端一條應答消息之后，計數器就加1，理論上應該有100條服務端日志
        System.out.println("Now is: " + body + "; the current is "+ (++counter));
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        log.warning("Unexpected exception from downstream: " + cause.getMessage());
        ctx.close();
    }
}

（5）運行測試結果

服務端：

The time server receive order: QUERY TIME ORDER; the counter is : 1
The time server receive order: QUERY TIME ORDER; the counter is : 2
...
The time server receive order: QUERY TIME ORDER; the counter is : 99
The time server receive order: QUERY TIME ORDER; the counter is : 100

客戶端：

Now is: Mon Jul 26 22:18:51 CST 2021; the current is 1
Now is: Mon Jul 26 22:18:51 CST 2021; the current is 2
...
Now is: Mon Jul 26 22:18:51 CST 2021; the current is 99
Now is: Mon Jul 26 22:18:51 CST 2021; the current is 100

根據結果可知，每條消息都對計數器加1，並沒有發生粘包現象。

2.按分隔符文本解碼器DelimiterBasedFrameDecoder

　　DelimiterBasedFrameDecoder是以分隔符作為碼流結束標識的消息解碼，改造代碼，以“$_”作為分隔符

（1）改造TimeServer

　　增加以“$_”為分隔符的DelimiterBasedFrameDecoder解碼器，DelimiterBasedFrameDecoder構造器其中第一個參數長度表示當達到該長度后仍然沒有查找到分隔符，就會拋出TooLongFrameException。這是防止異常碼流缺失分隔符導致內存溢出。

public class TimeServer {

    public static final Logger log = LoggerFactory.getLogger(TimeServer.class);

    public static void main(String[] args) throws Exception {
        new TimeServer().bind();
    }


    public void bind() throws Exception {
        // NIO 線程組
        NioEventLoopGroup bossGroup = new NioEventLoopGroup();
        NioEventLoopGroup workerGroup = new NioEventLoopGroup();

        try {
            ServerBootstrap bootstrap = new ServerBootstrap();
            bootstrap.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .option(ChannelOption.SO_BACKLOG, 1024)
                    .handler(new LoggingHandler(LogLevel.INFO))
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel socketChannel) {
                            // 以“$_”為分隔符
                            ByteBuf delimiter = Unpooled.copiedBuffer("$_".getBytes());
                            socketChannel.pipeline().addLast(new DelimiterBasedFrameDecoder(1024, delimiter));
                            socketChannel.pipeline().addLast(new StringDecoder());
                            socketChannel.pipeline().addLast(new TimeServerHandler());
                        }
                    });
            // 綁定端口，同步等待成功
            ChannelFuture f = bootstrap.bind(NettyConstant.LOCAL_IP, NettyConstant.LOCAL_PORT).sync();
            log.info("Time server[{}] start success", NettyConstant.LOCAL_IP + ": " + NettyConstant.LOCAL_PORT);
            // 等待所有服務端監聽端口關閉
            f.channel().closeFuture().sync();
        } finally {
            // 優雅退出，釋放線程池資源
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }

    }

}

（2）改造TimeServerHandler

　　對返回客戶端的消息增加分隔符“$_”

public class TimeServerHandler extends ChannelInboundHandlerAdapter {
    private static final Logger log = Logger.getLogger(TimeClientHandler.class.getName());


    private int counter;

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        // 不需要對消息進行編解碼，直接String讀取
        String body = (String) msg;
        // 每收到一條消息計數器就加1, 理論上應該接收到100條
        System.out.println("The time server receive order: " + body + "; the counter is : "+ (++counter));
        String currentTime = "QUERY TIME ORDER".equalsIgnoreCase(body) ?
                new Date(System.currentTimeMillis()).toString():"BAD ORDER";
        // 返回客戶端需要追加分隔符
        currentTime = currentTime + "$_";
        ByteBuf resp = Unpooled.copiedBuffer(currentTime.getBytes());
        ctx.writeAndFlush(resp);
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        log.warning("Unexpected exception from downstream: " + cause.getMessage());
        ctx.close();
    }
}

（3）改造TimeClient

　　增加以“$_”為分隔符的DelimiterBasedFrameDecoder解碼器

public class TimeClient {

    public static final Logger log = LoggerFactory.getLogger(TimeClient.class);

    public static void main(String[] args) throws Exception {
        new TimeClient().connect(NettyConstant.LOCAL_IP, NettyConstant.LOCAL_PORT);
    }

    public void connect(final String host, final int port) throws Exception {
        // NIO 線程組
        EventLoopGroup group = new NioEventLoopGroup();
        try {
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(group)
                    .channel(NioSocketChannel.class)
                    .option(ChannelOption.TCP_NODELAY, true)
                    .handler(new LoggingHandler(LogLevel.INFO))
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            // 以“$_”為分隔符
                            ByteBuf delimiter = Unpooled.copiedBuffer("$_".getBytes());
                            socketChannel.pipeline().addLast(new DelimiterBasedFrameDecoder(1024, delimiter));
                            socketChannel.pipeline().addLast(new StringDecoder());
                            socketChannel.pipeline().addLast(new TimeClientHandler());
                        }
                    });

            // 發起異步連接操作
            ChannelFuture f = bootstrap.connect(host, port).sync();
            // 等待所有服務端監聽端口關閉
            f.channel().closeFuture().sync();
        } finally {
            // 優雅退出，釋放線程池資源
            group.shutdownGracefully();
        }

    }

}

（4）改造TimeClientHandler

　　對發送命令增加“$_”分隔符

public class TimeClientHandler extends ChannelInboundHandlerAdapter {

    private static final Logger log = Logger.getLogger(TimeClientHandler.class.getName());

    private int counter;
    private byte[] req;

    public TimeClientHandler() {
        // 以$_為分隔符，發送命令
        req = ("QUERY TIME ORDER$_").getBytes();
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        ByteBuf message = null;
        // 循環發送100條消息，每發送一條刷新一次，服務端理論上接收到100條查詢時間指令的請求
        for (int i = 0; i < 100; i++) {
            message = Unpooled.buffer(req.length);
            message.writeBytes(req);
            ctx.writeAndFlush(message);
        }
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        // 不需要編解碼了，直接返回了字符串的應答消息
        String body = (String) msg;
        // 客戶端每接收到服務端一條應答消息之后，計數器就加1，理論上應該有100條服務端日志
        System.out.println("Now is: " + body + "; the current is "+ (++counter));
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        log.warning("Unexpected exception from downstream: " + cause.getMessage());
        ctx.close();
    }
}

（5）運行測試結果

服務端：

The time server receive order: QUERY TIME ORDER; the counter is : 1
The time server receive order: QUERY TIME ORDER; the counter is : 2 ... The time server receive order: QUERY TIME ORDER; the counter is : 99 The time server receive order: QUERY TIME ORDER; the counter is : 100

客戶端：

Now is: Mon Jul 26 22:18:51 CST 2021; the current is 1
Now is: Mon Jul 26 22:18:51 CST 2021; the current is 2 ... Now is: Mon Jul 26 22:18:51 CST 2021; the current is 99 Now is: Mon Jul 26 22:18:51 CST 2021; the current is 100

根據結果可知，每條消息都對計數器加1，並沒有發生粘包現象。

3.固定長度解碼器FixedLengthFrameDecoder

　　FixedLengthFrameDecoder是固定長度解碼器，能夠對固定長度的消息進行自動解碼，利用FixedLengthFrameDecoder，無論多少數據，都會按照構造函數中設置的固定長度進行解碼，如果是半包消息，FixedLengthFrameDecoder會緩存半包消息並等待下一個包到達后進行拼包，直到讀取到一個完整的包。

　　在服務端ChannelPipeline中新增FixedLengthFrameDecoder，長度為10。然后增加EchoServerHannel處理器，輸出服務端接收到的命令

（1）EchoServer

　　增加長度為10的FixedLengthFrameDecoder解碼器，同時再增加StringDecoder解碼器

public class EchoServer {

    public static final Logger log = LoggerFactory.getLogger(EchoServer.class);

    public static void main(String[] args) throws Exception {
        new EchoServer().bind();
    }


    public void bind() throws Exception {
        // NIO 線程組
        NioEventLoopGroup bossGroup = new NioEventLoopGroup();
        NioEventLoopGroup workerGroup = new NioEventLoopGroup();

        try {
            ServerBootstrap bootstrap = new ServerBootstrap();
            bootstrap.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .option(ChannelOption.SO_BACKLOG, 1024)
                    .handler(new LoggingHandler(LogLevel.INFO))
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel socketChannel) {
                            // 增加固定長度解碼器
                            socketChannel.pipeline().addLast(new FixedLengthFrameDecoder(10));
                            // 增加字符解碼器，將msg直接轉為string
                            socketChannel.pipeline().addLast(new StringDecoder());
                            socketChannel.pipeline().addLast(new EchoServerHandler());
                        }
                    });
            // 綁定端口，同步等待成功
            ChannelFuture f = bootstrap.bind(NettyConstant.LOCAL_IP, NettyConstant.LOCAL_PORT).sync();
            log.info("Time server[{}] start success", NettyConstant.LOCAL_IP + ": " + NettyConstant.LOCAL_PORT);
            // 等待所有服務端監聽端口關閉
            f.channel().closeFuture().sync();
        } finally {
            // 優雅退出，釋放線程池資源
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }

    }

}

（2）EchoServerHandler

　　輸出客戶端發送的命令，直接輸出msg即可，因為服務端已經增加了StringDecoder解碼器，直接轉為String

public class EchoServerHandler extends ChannelInboundHandlerAdapter {

    private static final Logger log = Logger.getLogger(EchoServerHandler.class.getName());

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        System.out.println("The time server receive order: " + msg);
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        log.warning("Unexpected exception from downstream: " + cause.getMessage());
        ctx.close();
    }
}

（3）Telnet命令測試結果

CMD窗口Telnet窗口連接 telnet 127.0.0.1 8888

回顯輸入消息welcome Lijian

查看服務端console

2021-07-26 23:25:21,921  INFO [nioEventLoopGroup-2-1] - [id: 0xe4d49ee6, L:/127.0.0.1:8888] READ: [id: 0x928b38a4, L:/127.0.0.1:8888 - R:/127.0.0.1:62315]
2021-07-26 23:25:21,922  INFO [nioEventLoopGroup-2-1] - [id: 0xe4d49ee6, L:/127.0.0.1:8888] READ COMPLETE
The time server receive order: welcome Li

根據結果可知，服務端只接收到客戶端發送的“welcome Lijian”的前10個字符，及說明FixedLengthFrameDecoder是有效的

 

本篇博文是Netty的基礎篇，主要介紹Netty針解決TCP粘包/拆包而產生的解碼器，Netty基礎篇還涉及到序列化的問題，后面將會繼續介紹。

 


免責聲明！

本站轉載的文章為個人學習借鑒使用，本站對版權不負任何法律責任。如果侵犯了您的隱私權益，請聯系本站郵箱yoyou2525@163.com刪除。



猜您在找 Netty(三) 什么是 TCP 拆、粘包？如何解決？ netty 解決TCP粘包與拆包問題(二) Netty TCP 粘包和拆包 及解決方案 Netty解決TCP粘包/拆包問題 - 按行分隔字符串解碼器 Netty（三）TCP粘包拆包處理 【Netty】TCP粘包和拆包 Netty（二）——TCP粘包/拆包 Netty解決粘包和拆包問題的四種方案 Netty解決粘包和拆包問題的四種方案 【轉】Netty之解決TCP粘包拆包(自定義協議)
 
粵ICP備18138465號   © 2018-2025 CODEPRJ.COM