Netty解决粘包/半包问题（十）

刺骨的言语ヽ痛彻心扉 2022-10-29 12:29 230阅读 0赞

今天分享Netty解决粘包/半包问题，通常传输均存在这种问题，比如下面默认的代码：

**一、存在粘包、半包问题的示例：**

**1、服务端代码：**

public class EchoServer  {
    
        private final int port;
    
        public EchoServer(int port) {
            this.port = port;
        }
    
        public static void main(String[] args) throws InterruptedException {
            int port = 9999;
            EchoServer echoServer = new EchoServer(port);
            System.out.println("服务器即将启动");
            echoServer.start();
            System.out.println("服务器关闭");
        }
    
        public void start() throws InterruptedException {
            final EchoServerHandler serverHandler = new EchoServerHandler();
            /**线程组*/
            EventLoopGroup group  = new NioEventLoopGroup();
            try {
                /**服务端启动必备*/
                ServerBootstrap b = new ServerBootstrap();
                b.group(group)
                        .channel(NioServerSocketChannel.class)/**指定使用NIO的通信模式*/
                .localAddress(new InetSocketAddress(port))/**指定监听端口*/
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
                        ch.pipeline().addLast(serverHandler);
                    }
                });
                ChannelFuture f = b.bind().sync();/**异步绑定到服务器，sync()会阻塞到完成*/
                f.channel().closeFuture().sync();/**阻塞当前线程，直到服务器的ServerChannel被关闭*/
            } finally {
                group.shutdownGracefully().sync();
            }
        }
    }

服务端操作的业务代码：

@ChannelHandler.Sharable
    public class EchoServerHandler extends ChannelInboundHandlerAdapter {
    
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            ByteBuf in = (ByteBuf)msg;
            System.out.println("Server accept: "+in.toString(CharsetUtil.UTF_8));
    //
            ctx.writeAndFlush(in);
            //ctx.close();
        }
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            cause.printStackTrace();
            ctx.close();
        }
    }

2、客户端代码：

public class EchoClient {
    
        private final int port;
        private final String host;
    
        public EchoClient(int port, String host) {
            this.port = port;
            this.host = host;
        }
    
        public void start() throws InterruptedException {
            EventLoopGroup group = new NioEventLoopGroup();/*线程组*/
            try {
                final Bootstrap b = new Bootstrap();;/*客户端启动必须*/
                b.group(group)/*将线程组传入*/
                        .channel(NioSocketChannel.class)/*指定使用NIO进行网络传输*/
                        .remoteAddress(new InetSocketAddress(host,port))/*配置要连接服务器的ip地址和端口*/
                        .handler(new ChannelInitializer<SocketChannel>() {
                            protected void initChannel(SocketChannel ch) throws Exception {
                                ch.pipeline().addLast(new EchoClientHandler());
                            }
                        });
                ChannelFuture f = b.connect().sync();
                f.channel().closeFuture().sync();
            } finally {
                group.shutdownGracefully().sync();
            }
        }
    
        public static void main(String[] args) throws InterruptedException {
            new EchoClient(9999,"127.0.0.1").start();
        }
    }

客户端操作业务的代码：

public class EchoClientHandler extends SimpleChannelInboundHandler<ByteBuf> {
    
        private AtomicInteger counter = new AtomicInteger(0);
    
        /*** 客户端读取到网络数据后的处理*/
        protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception {
            System.out.println("client Accept["+msg.toString(CharsetUtil.UTF_8)
                    +"] and the counter is:"+counter.incrementAndGet());
        }
    
        /*** 客户端被通知channel活跃后，做事*/
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            ByteBuf msg = null;
            String request = "LILI,Lisi,zhangsan,Wangwu,zhaoliu"
                    + System.getProperty("line.separator");
            for(int i=0;i<100;i++){
                msg = Unpooled.buffer(request.length());
                msg.writeBytes(request.getBytes());
                ctx.writeAndFlush(msg);
            }
        }
    
        /*** 发生异常后的处理*/
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            cause.printStackTrace();
            ctx.close();
        }
    }

3、分别启动服务端和客户端，执行：

服务端执行：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70][]

客户端执行结果：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 1][]

结果显示仅推送了两次数据，证明发生了粘包问题。

**二、粘包/半包详解**

**1、什么是**  **TCP**  **粘包半包？如下图所示:**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 2][]

假设客户端分别发送了两个数据包  D1  和  D2  给服务端，由于服务端一次读取到的字节 数是不确定的，故可能存在以下 4  种情况。

（ 1 ）服务端分两次读取到了两个独立的数据包，分别是  D1  和  D2 ，没有粘包和拆包；

（ 2 ）服务端一次接收到了两个数据包， D1  和  D2  粘合在一起，被称为  TCP  粘包；

（ 3 ）服务端分两次读取到了两个数据包，第一次读取到了完整的  D1  包和  D2  包的部分 内容，第二次读取到了 D2  包的剩余内容，这被称为  TCP  拆包；

（ 4 ）服务端分两次读取到了两个数据包，第一次读取到了  D1  包的部分内容  D1\_1 ，第 二次读取到了 D1  包的剩余内容  D1\_2  和  D2  包的整包。 如果此时服务端 TCP  接收滑窗非常小，而数据包  D1  和  D2  比较大，很有可能会发生第 五种可能，即服务端分多次才能将 D1  和  D2  包接收完全，期间发生多次拆包。

**三、TCP**  **粘包** **/** **半包发生的原因**

由于  TCP  协议本身的机制（面向连接的可靠地协议 \- 三次握手机制）客户端与服务器会 维持一个连接（Channel ），数据在连接不断开的情况下，可以持续不断地将多个数据包发 往服务器，但是如果发送的网络数据包太小，那么他本身会启用 Nagle  算法（可配置是否启 用）对较小的数据包进行合并（基于此，TCP  的网络延迟要  UDP  的高些）然后再发送（超 时或者包大小足够）。那么这样的话，服务器在接收到消息（数据流）的时候就无法区分哪 些数据包是客户端自己分开发送的，这样产生了粘包；服务器在接收到数据库后，放到缓冲 区中，如果消息没有被及时从缓存区取走，下次在取数据的时候可能就会出现一次取出多个 数据包的情况，造成粘包现象 UDP：本身作为无连接的不可靠的传输协议（适合频繁发送较小的数据包），他不会对 数据包进行合并发送（也就没有 Nagle  算法之说了），他直接是一端发送什么数据，直接就 发出去了，既然他不会对数据合并，每一个数据包都是完整的（数据+UDP  头 \+IP  头等等发一 次数据封装一次）也就没有粘包一说了。 分包产生的原因就简单的多：可能是 IP  分片传输导致的，也可能是传输过程中丢失部分包导致出现的半包，还有可能就是一个包可能被分成了两次传输，在取数据的时候，先取 到了一部分（还可能与接收的缓冲区大小有关系），总之就是一个数据包被分成了多次接收。

更具体的原因有三个，分别如下。

1.  应用程序写入数据的字节大小大于套接字发送缓冲区的大小

2.  进行  MSS  大小的  TCP  分段。 MSS  是最大报文段长度的缩写。 MSS  是  TCP  报文段中的数据字段的最大长度。数据字段加上 TCP  首部才等于整个的  TCP  报文段。所以  MSS  并不是 TCP 报文段的最大长度，而是： MSS=TCP  报文段长度 \-TCP  首部长度

3.  以太网的  payload  大于  MTU  进行  IP  分片。 MTU  指：一种通信协议的某一层上面所能 通过的最大数据包大小。如果 IP  层有一个数据包要传，而且数据的长度比链路层的  MTU  大， 那么IP  层就会进行分片，把数据包分成托干片，让每一片都不超过  MTU 。注意， IP  分片可以发生在原始发送端主机上，也可以发生在中间路由器上。

**四、解决粘包半包问题**

由于底层的  TCP  无法理解上层的业务数据，所以在底层是无法保证数据包不被拆分和重组的，这个问题只能通过上层的应用协议栈设计来解决，根据业界的主流协议的解决方案，

可以归纳如下。

1、  在包尾增加分割符，比如回车换行符进行分割，例如  FTP 协议；（回车换行符进行分割）测试代码如下：

服务端核心代码：

public class LineBaseEchoServer {
    
        public static final int PORT = 9998;
    
        public static void main(String[] args) throws InterruptedException {
            LineBaseEchoServer lineBaseEchoServer = new LineBaseEchoServer();
            System.out.println("服务器即将启动");
            lineBaseEchoServer.start();
        }
    
        public void start() throws InterruptedException {
            final LineBaseServerHandler serverHandler = new LineBaseServerHandler();
            EventLoopGroup group = new NioEventLoopGroup();/*线程组*/
            try {
                ServerBootstrap b = new ServerBootstrap();/*服务端启动必须*/
                b.group(group)/*将线程组传入*/
                    .channel(NioServerSocketChannel.class)/*指定使用NIO进行网络传输*/
                    .localAddress(new InetSocketAddress(PORT))/*指定服务器监听端口*/
                    /*服务端每接收到一个连接请求，就会新启一个socket通信，也就是channel，
                    所以下面这段代码的作用就是为这个子channel增加handle*/
                    .childHandler(new ChannelInitializerImp());
                ChannelFuture f = b.bind().sync();/*异步绑定到服务器，sync()会阻塞直到完成*/
                System.out.println("服务器启动完成，等待客户端的连接和数据.....");
                f.channel().closeFuture().sync();/*阻塞直到服务器的channel关闭*/
            } finally {
                group.shutdownGracefully().sync();/*优雅关闭线程组*/
            }
        }
    
        private static class ChannelInitializerImp extends ChannelInitializer<Channel> {
    
            @Override
            protected void initChannel(Channel ch) throws Exception {
                ch.pipeline().addLast(new LineBasedFrameDecoder(1024));
                ch.pipeline().addLast(new LineBaseServerHandler());
            }
        }
    
    }

服务端业务代码：

@ChannelHandler.Sharable
    public class LineBaseServerHandler extends ChannelInboundHandlerAdapter {
    
        private AtomicInteger counter = new AtomicInteger(0);
    
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            System.out.println("客户端：["+ctx.channel().remoteAddress()+"]已连接.........");
        }
    
        /*** 服务端读取到网络数据后的处理*/
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            ByteBuf in = (ByteBuf)msg;
            String request = in.toString(CharsetUtil.UTF_8);
            System.out.println("Server Accept["+request
                    +"] and the counter is:"+counter.incrementAndGet());
            String resp = "Hello,"+request+". Welcome to Netty World!"
                    + System.getProperty("line.separator");
            ctx.writeAndFlush(Unpooled.copiedBuffer(resp.getBytes()));
        }
    
        /*** 服务端读取完成网络数据后的处理*/
        @Override
        public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
            //ctx.writeAndFlush(Unpooled.EMPTY_BUFFER).addListener(ChannelFutureListener.CLOSE);
        }
    
        /*** 发生异常后的处理*/
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            cause.printStackTrace();
            ctx.close();
        }
    
        @Override
        public void channelInactive(ChannelHandlerContext ctx) throws Exception {
            System.out.println(ctx.channel().remoteAddress()+"即将关闭...");
        }
    }

客户端核心代码：

public class LineBaseEchoClient {
    
        private final String host;
    
        public LineBaseEchoClient(String host) {
            this.host = host;
        }
    
        public void start() throws InterruptedException {
            EventLoopGroup group = new NioEventLoopGroup();/*线程组*/
            try {
                final Bootstrap b = new Bootstrap();;/*客户端启动必须*/
                b.group(group)/*将线程组传入*/
                        .channel(NioSocketChannel.class)/*指定使用NIO进行网络传输*/
                        .remoteAddress(new InetSocketAddress(host,LineBaseEchoServer.PORT))/*配置要连接服务器的ip地址和端口*/
                        .handler(new ChannelInitializerImp());
                ChannelFuture f = b.connect().sync();
                System.out.println("已连接到服务器.....");
                f.channel().closeFuture().sync();
            } finally {
                group.shutdownGracefully().sync();
            }
        }
    
        private static class ChannelInitializerImp extends ChannelInitializer<Channel> {
    
            @Override
            protected void initChannel(Channel ch) throws Exception {
                ch.pipeline().addLast(new LineBasedFrameDecoder(1024));
                ch.pipeline().addLast(new LineBaseClientHandler());
            }
        }
    
        public static void main(String[] args) throws InterruptedException {
            new LineBaseEchoClient("127.0.0.1").start();
        }
    }

客户端业务代码：

public class LineBaseClientHandler extends SimpleChannelInboundHandler<ByteBuf> {
    
        private AtomicInteger counter = new AtomicInteger(0);
    
        /*** 客户端读取到网络数据后的处理*/
        protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception {
            System.out.println("client Accept["+msg.toString(CharsetUtil.UTF_8)
                    +"] and the counter is:"+counter.incrementAndGet());
            ctx.close();
        }
    
        /*** 客户端被通知channel活跃后，做事*/
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            ByteBuf msg = null;
            String request = "LILI,Lisi,zhangsan,Wangwu,zhaoliu"
                    + System.getProperty("line.separator");
            for(int i=0;i<10;i++){
                msg = Unpooled.buffer(request.length());
                msg.writeBytes(request.getBytes());
                ctx.writeAndFlush(msg);
            }
        }
    
        /*** 发生异常后的处理*/
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            cause.printStackTrace();
            ctx.close();
        }
    }

服务端执行结果：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 3][]

客户端执行结果：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 4][]

2、（自定义分割符）测试核心代码

服务端核心代码：

public class DelimiterEchoServer {
    
        public static final String DELIMITER_SYMBOL = "@~";
        public static final int PORT = 9997;
    
        public static void main(String[] args) throws InterruptedException {
            DelimiterEchoServer delimiterEchoServer = new DelimiterEchoServer();
            System.out.println("服务器即将启动");
            delimiterEchoServer.start();
        }
    
        public void start() throws InterruptedException {
            final DelimiterServerHandler serverHandler = new DelimiterServerHandler();
            EventLoopGroup group = new NioEventLoopGroup();/*线程组*/
            try {
                ServerBootstrap b = new ServerBootstrap();/*服务端启动必须*/
                b.group(group)/*将线程组传入*/
                    .channel(NioServerSocketChannel.class)/*指定使用NIO进行网络传输*/
                    .localAddress(new InetSocketAddress(PORT))/*指定服务器监听端口*/
                    /*服务端每接收到一个连接请求，就会新启一个socket通信，也就是channel，
                    所以下面这段代码的作用就是为这个子channel增加handle*/
                    .childHandler(new ChannelInitializerImp());
                ChannelFuture f = b.bind().sync();/*异步绑定到服务器，sync()会阻塞直到完成*/
                System.out.println("服务器启动完成，等待客户端的连接和数据.....");
                f.channel().closeFuture().sync();/*阻塞直到服务器的channel关闭*/
            } finally {
                group.shutdownGracefully().sync();/*优雅关闭线程组*/
            }
        }
    
        private static class ChannelInitializerImp extends ChannelInitializer<Channel> {
    
            @Override
            protected void initChannel(Channel ch) throws Exception {
                ByteBuf delimiter = Unpooled.copiedBuffer(DELIMITER_SYMBOL.getBytes());
                ch.pipeline().addLast(new DelimiterBasedFrameDecoder(1024,
                        delimiter));
                ch.pipeline().addLast(new DelimiterServerHandler());
            }
        }
    }

服务端业务代码：

@ChannelHandler.Sharable
    public class DelimiterServerHandler extends ChannelInboundHandlerAdapter {
    
        private AtomicInteger counter = new AtomicInteger(0);
    
        /*** 服务端读取到网络数据后的处理*/
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            ByteBuf in = (ByteBuf) msg;
            String request = in.toString(CharsetUtil.UTF_8);
            System.out.println("Server Accept["+request
                    +"] and the counter is:"+counter.incrementAndGet());
            String resp = "Hello,"+request+". Welcome to Netty World!"
                    + DelimiterEchoServer.DELIMITER_SYMBOL;
            ctx.writeAndFlush(Unpooled.copiedBuffer(resp.getBytes()));
        }
    
        /*** 服务端读取完成网络数据后的处理*/
        @Override
        public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
            ctx.writeAndFlush(Unpooled.EMPTY_BUFFER).addListener(ChannelFutureListener.CLOSE);
        }
    
        /*** 发生异常后的处理*/
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            cause.printStackTrace();
            ctx.close();
        }
    }

客户端核心代码：

public class DelimiterEchoClient {
    
        private final String host;
    
        public DelimiterEchoClient(String host) {
            this.host = host;
        }
    
        public void start() throws InterruptedException {
            EventLoopGroup group = new NioEventLoopGroup();/*线程组*/
            try {
                final Bootstrap b = new Bootstrap();;/*客户端启动必须*/
                b.group(group)/*将线程组传入*/
                        .channel(NioSocketChannel.class)/*指定使用NIO进行网络传输*/
                        .remoteAddress(new InetSocketAddress(host,DelimiterEchoServer.PORT))/*配置要连接服务器的ip地址和端口*/
                        .handler(new ChannelInitializerImp());
                ChannelFuture f = b.connect().sync();
                System.out.println("已连接到服务器.....");
                f.channel().closeFuture().sync();
            } finally {
                group.shutdownGracefully().sync();
            }
        }
    
        private static class ChannelInitializerImp extends ChannelInitializer<Channel> {
    
            @Override
            protected void initChannel(Channel ch) throws Exception {
                ByteBuf delimiter = Unpooled.copiedBuffer(
                        DelimiterEchoServer.DELIMITER_SYMBOL.getBytes());
                ch.pipeline().addLast(new DelimiterBasedFrameDecoder(1024,
                        delimiter));
                ch.pipeline().addLast(new DelimiterClientHandler());
            }
        }
    
        public static void main(String[] args) throws InterruptedException {
            new DelimiterEchoClient("127.0.0.1").start();
        }
    }

客户端业务代码：

public class DelimiterClientHandler extends SimpleChannelInboundHandler<ByteBuf> {
    
        private AtomicInteger counter = new AtomicInteger(0);
    
        /*** 客户端读取到网络数据后的处理*/
        protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception {
            System.out.println("client Accept["+msg.toString(CharsetUtil.UTF_8)
                    +"] and the counter is:"+counter.incrementAndGet());
        }
    
        /*** 客户端被通知channel活跃后，做事*/
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            ByteBuf msg = null;
            String request = "LILI,Lisi,zhangsan,Wangwu,zhaoliu"
                    + DelimiterEchoServer.DELIMITER_SYMBOL;
            for(int i=0;i<10;i++){
                msg = Unpooled.buffer(request.length());
                msg.writeBytes(request.getBytes());
                ctx.writeAndFlush(msg);
            }
        }
    
        /*** 发生异常后的处理*/
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            cause.printStackTrace();
            ctx.close();
        }
    }

服务端执行结果：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 5][]

客户端执行结果：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 6][]

3、消息定长，例如每个报文的大小为固定长度 200 字节，如果不够，空位补空格；

服务端核心代码：

public class FixedLengthEchoServer {
    
        public static final String RESPONSE = "Welcome to Netty!";
        public static final int PORT = 9996;
    
        public static void main(String[] args) throws InterruptedException {
            FixedLengthEchoServer fixedLengthEchoServer = new FixedLengthEchoServer();
            System.out.println("服务器即将启动");
            fixedLengthEchoServer.start();
        }
    
        public void start() throws InterruptedException {
            final FixedLengthServerHandler serverHandler = new FixedLengthServerHandler();
            EventLoopGroup group = new NioEventLoopGroup();/*线程组*/
            try {
                ServerBootstrap b = new ServerBootstrap();/*服务端启动必须*/
                b.group(group)/*将线程组传入*/
                    .channel(NioServerSocketChannel.class)/*指定使用NIO进行网络传输*/
                    .localAddress(new InetSocketAddress(PORT))/*指定服务器监听端口*/
                    /*服务端每接收到一个连接请求，就会新启一个socket通信，也就是channel，
                    所以下面这段代码的作用就是为这个子channel增加handle*/
                    .childHandler(new ChannelInitializerImp());
                ChannelFuture f = b.bind().sync();/*异步绑定到服务器，sync()会阻塞直到完成*/
                System.out.println("服务器启动完成，等待客户端的连接和数据.....");
                f.channel().closeFuture().sync();/*阻塞直到服务器的channel关闭*/
            } finally {
                group.shutdownGracefully().sync();/*优雅关闭线程组*/
            }
        }
    
        private static class ChannelInitializerImp extends ChannelInitializer<Channel> {
    
            @Override
            protected void initChannel(Channel ch) throws Exception {
                ch.pipeline().addLast(
                        new FixedLengthFrameDecoder(
                                FixedLengthEchoClient.REQUEST.length()));
                ch.pipeline().addLast(new FixedLengthServerHandler());
            }
        }
    }

服务端业务代码：

@ChannelHandler.Sharable
    public class FixedLengthServerHandler extends ChannelInboundHandlerAdapter {
    
        private AtomicInteger counter = new AtomicInteger(0);
    
        /*** 服务端读取到网络数据后的处理*/
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            ByteBuf in = (ByteBuf) msg;
            String request = in.toString(CharsetUtil.UTF_8);
            System.out.println("Server Accept["+request
                    +"] and the counter is:"+counter.incrementAndGet());
            ctx.writeAndFlush(Unpooled.copiedBuffer(
                    FixedLengthEchoServer.RESPONSE.getBytes()));
        }
    
        /*** 发生异常后的处理*/
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            cause.printStackTrace();
            ctx.close();
        }
    }

客户端核心代码：

public class FixedLengthEchoClient {
    
        public final static String REQUEST = "LILI,Lisi,zhangsan,Wangwu,zhaoliu";
    
        private final String host;
    
        public FixedLengthEchoClient(String host) {
            this.host = host;
        }
    
        public void start() throws InterruptedException {
            EventLoopGroup group = new NioEventLoopGroup();/*线程组*/
            try {
                final Bootstrap b = new Bootstrap();;/*客户端启动必须*/
                b.group(group)/*将线程组传入*/
                        .channel(NioSocketChannel.class)/*指定使用NIO进行网络传输*/
                        .remoteAddress(new InetSocketAddress(host,FixedLengthEchoServer.PORT))/*配置要连接服务器的ip地址和端口*/
                        .handler(new ChannelInitializerImp());
                ChannelFuture f = b.connect().sync();
                System.out.println("已连接到服务器.....");
                f.channel().closeFuture().sync();
            } finally {
                group.shutdownGracefully().sync();
            }
        }
    
        private static class ChannelInitializerImp extends ChannelInitializer<Channel> {
    
            @Override
            protected void initChannel(Channel ch) throws Exception {
                ch.pipeline().addLast(
                        new FixedLengthFrameDecoder(
                                FixedLengthEchoServer.RESPONSE.length()));
                ch.pipeline().addLast(new FixedLengthClientHandler());
            }
        }
    
        public static void main(String[] args) throws InterruptedException {
            new FixedLengthEchoClient("127.0.0.1").start();
        }
    }

客户端业务代码：

public class FixedLengthClientHandler extends SimpleChannelInboundHandler<ByteBuf> {
    
        private AtomicInteger counter = new AtomicInteger(0);
    
        /*** 客户端读取到网络数据后的处理*/
        protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception {
            System.out.println("client Accept["+msg.toString(CharsetUtil.UTF_8)
                    +"] and the counter is:"+counter.incrementAndGet());
        }
    
        /*** 客户端被通知channel活跃后，做事*/
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            ByteBuf msg = null;
            for(int i=0;i<10;i++){
                msg = Unpooled.buffer(FixedLengthEchoClient.REQUEST.length());
                msg.writeBytes(FixedLengthEchoClient.REQUEST.getBytes());
                ctx.writeAndFlush(msg);
            }
        }
    
        /*** 发生异常后的处理*/
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            cause.printStackTrace();
            ctx.close();
        }
    }

执行结果：

服务端执行结果：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 7][]

客户端执行结果：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 8][]

以上三种方案，均能解决粘包和半包问题。

4、 将消息分为消息头和消息体，消息头中包含表示消息总长度（或者消息体长度） 的字段，通常设计思路为消息头的第一个字段使用 int32  来表示消息的总长度，使用

LengthFieldBasedFrameDecoder ，也可以解决，后期有时间会详细说明和使用，其实以上三种方案足够使用解决问题。

Netty解决粘包/半包问题到此分析完成，下篇我们分析Netty 编解码器框架，敬请期待！

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70]: /images/20221024/ab8c6a4a7b814d7eacdc3d07e453906f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 1]: /images/20221024/9abdabf6e5914a5285d6cb7a400b6adb.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 2]: /images/20221024/50a79bf288af43fbbf24c42832d8d844.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 3]: /images/20221024/6866731af5974445bc5b104df3c92478.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 4]: /images/20221024/2635adf2061148af8594dace400abc5e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 5]: /images/20221024/51d1688b0c8546119c27abcf8ee6bb07.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 6]: /images/20221024/cf6422a50f9e4fdd87e734fd97988f0e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 7]: /images/20221024/a151b74d47f84d409ab97b488827fe4d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 8]: /images/20221024/853b7787013d4a1790b058f9ad4be3e3.png