FixedLengthFrameDecoder 固定长度解码器，解决 TCP 粘包

短命女 2022-04-17 01:39 339阅读 0赞

# FixedLengthFrameDecoder  #

1、消息长度固定，累积读取到长度总和为定长 LEN 的报文后，就认为读取到了一个完整的消息，再将计数器置位，重新读取下一个数据报。例如可以让每个报文的大小为固定长度 1024 字节，如果消息长度不够，则使用空位填补空缺，这样读取到了之后，只需要 trim 去掉空格即可。

2、FixedLengthFrameDecoder 固定长度解码器，它能够按照指定的长度对消息进行自动解码，开发者不需要考虑 TCP 的粘包与拆包问题，非常实用。无论一次接收到多少数据报，它都会按照构造器中设置的固定长度进行解码，如果是半包消息，FixedLengthFrameDecoder  会缓存半包消息并等待下个包到达之后进行拼包合并，直到读取一个完整的消息包。

3、Netty 提供了各种解码器，它们都继承了 io.netty.handler.codec.ByteToMessageDecoder 抽象类。

4、本文仍然以《[DelimiterBasedFrameDecoder 自定义分隔符解码器，解决 TCP 粘包][DelimiterBasedFrameDecoder _ TCP]》中的代码进行改写。

![20181111123055137.png][]

5、实例中将设置消息固定长度为 64 字节，发现消息的实际长度会有小于 64 字节的，会有等于 64 字节的，会有大于 64 字节的，这样接收时，对于小于等于 64 字节的消息将会正确接收，而对于大于 64 字节的消息显然会进行拆分了。

# 服务器 #

**FixedServer 内容如下：**

import com.lct.netty.echo.EchoServerHandler;
    import io.netty.bootstrap.ServerBootstrap;
    import io.netty.buffer.ByteBuf;
    import io.netty.buffer.Unpooled;
    import io.netty.channel.ChannelFuture;
    import io.netty.channel.ChannelInitializer;
    import io.netty.channel.ChannelOption;
    import io.netty.channel.EventLoopGroup;
    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.FixedLengthFrameDecoder;
    import io.netty.handler.codec.string.StringDecoder;
    import io.netty.handler.logging.LogLevel;
    import io.netty.handler.logging.LoggingHandler;
    /**
     * Created by Administrator on 2018/11/11 0011.
     * Fixed 服务器
     */
    public class FixedServer {
        public static void main(String[] args) {
            int port = 9898;
            new FixedServer().bind(port);
        }
    
        public void bind(int port) {
            /**
             * interface EventLoopGroup extends EventExecutorGroup extends ScheduledExecutorService extends ExecutorService
             * 配置服务端的 NIO 线程池,用于网络事件处理，实质上他们就是 Reactor 线程组
             * bossGroup 用于服务端接受客户端连接，workerGroup 用于进行 SocketChannel 网络读写*/
            EventLoopGroup bossGroup = new NioEventLoopGroup();
            EventLoopGroup workerGroup = new NioEventLoopGroup();
            try {
                /** ServerBootstrap 是 Netty 用于启动 NIO 服务端的辅助启动类，用于降低开发难度
                 * */
                final ServerBootstrap b = new ServerBootstrap();
                b.group(bossGroup, workerGroup)
                        .channel(NioServerSocketChannel.class)
                        .option(ChannelOption.SO_BACKLOG, 1024)
                        .childHandler(new ChannelInitializer<SocketChannel>() {
                            @Override
                            protected void initChannel(SocketChannel ch) throws Exception {
                                System.out.println(Thread.currentThread().getName() + ",服务器初始化通道...");
                                /**
                                 * FixedLengthFrameDecoder(int frameLength)： frameLength：指定单条消息的长度
                                 * 为了测试结果明显，这里设置消息长度为 64 字节，之后在发送消息时，会让部分消息长度小于等于 64，然后部分消息长度大于 64，以观察区别
                                 * 实际应用中应该根据时间情况进行设置，比如 1024 字节
                                 */
                                ch.pipeline().addLast(new FixedLengthFrameDecoder(64));
                                ch.pipeline().addLast(new StringDecoder());
                                ch.pipeline().addLast(new FixedServerHandler());
                            }
                        });
    
                /**服务器启动辅助类配置完成后，调用 bind 方法绑定监听端口，调用 sync 方法同步等待绑定操作完成*/
                ChannelFuture f = b.bind(port).sync();
    
                System.out.println(Thread.currentThread().getName() + ",服务器开始监听端口，等待客户端连接.........");
    
                /**下面会进行阻塞，等待服务器连接关闭之后 main 方法退出，程序结束* */
                f.channel().closeFuture().sync();
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                /**优雅退出，释放线程池资源*/
                bossGroup.shutdownGracefully();
                workerGroup.shutdownGracefully();
            }
        }
    }

重点在第 53-54 行，在注释中已经说明。

**FixedServerHandler 内容如下：**

import io.netty.buffer.ByteBuf;
    import io.netty.buffer.Unpooled;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.ChannelInboundHandlerAdapter;
    import java.util.concurrent.atomic.AtomicInteger;
    /**
     * Created by Administrator on 2017/5/16.
     * ChannelInboundHandlerAdapter extends ChannelHandlerAdapter 用于对网络事件进行读写操作
     */
    public class FixedServerHandler extends ChannelInboundHandlerAdapter {
        /**
         * 因为多线程，所以使用原子操作类来进行计数
         */
        private static AtomicInteger atomicInteger = new AtomicInteger();
    
        /**
         * 收到客户端消息，自动触发
         *
         * @param ctx
         * @param msg
         * @throws Exception
         */
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            /**
             * FixedLengthFrameDecoder 自动对发送来的消息进行了解码，后续的 ChannelHandler 接收到的 msg 对象是完整的消息包
             * 第二个 ChannelHandler 是 StringDecoder，它将 ByteBuf 解码成字符串对象
             * 第三个 EchoServerHandler 接收到的 msg 对象就是解码后的字符串对象*/
            String body = (String) msg;
    
            /**
             * 使用 FixedLengthFrameDecoder 固定长度解码器时，通常对于消息实际长度不足构造器要求的长度时，会使用空格进行填充
             * 所以这里接收之后，使用 trime 去掉消息结尾多余的空格。
             */
            String receiveMsg = body.trim();
            System.out.println((atomicInteger.addAndGet(1)) + "--->" + Thread.currentThread().getName() + ",The server receive  order : " + receiveMsg);
    
            /**回复消息 ，body 中因为已经包含了结尾填充的空格，所以这里直接返回
             * 每次写的时候，同时刷新*/
            ByteBuf respByteBuf = Unpooled.copiedBuffer(body.getBytes());
            ctx.writeAndFlush(respByteBuf);
        }
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            System.out.println("-----客户端关闭:" + ctx.channel().remoteAddress());
            /**当发生异常时，关闭 ChannelHandlerContext，释放和它相关联的句柄等资源 */
            ctx.close();
        }
    }

重点在第 40 行，已在注释中说明。

# 客户端 #

**FixedClient 内容如下：**

import io.netty.bootstrap.Bootstrap;
    import io.netty.channel.ChannelFuture;
    import io.netty.channel.ChannelInitializer;
    import io.netty.channel.ChannelOption;
    import io.netty.channel.EventLoopGroup;
    import io.netty.channel.nio.NioEventLoopGroup;
    import io.netty.channel.socket.SocketChannel;
    import io.netty.channel.socket.nio.NioSocketChannel;
    import io.netty.handler.codec.FixedLengthFrameDecoder;
    import io.netty.handler.codec.string.StringDecoder;
    /**
     * Created by Administrator on 2017/5/16.
     * Fixed 客户端
     */
    public class FixedClient {
        /**
         * 使用 2 个线程模拟 2 个客户端
         *
         * @param args
         */
        public static void main(String[] args) {
            for (int i = 0; i < 2; i++) {
                new Thread(new MyThread()).start();
            }
        }
    
        static class MyThread implements Runnable {
    
            @Override
            public void run() {
                connect("192.168.1.20", 9898);
            }
    
            public void connect(String host, int port) {
                /**配置客户端 NIO 线程组/池*/
                EventLoopGroup group = new NioEventLoopGroup();
                try {
                    /**Bootstrap 与 ServerBootstrap 都继承(extends)于 AbstractBootstrap
                     * 创建客户端辅助启动类,并对其配置,与服务器稍微不同，这里的 Channel 设置为 NioSocketChannel
                     * 然后为其添加 Handler，这里直接使用匿名内部类，实现 initChannel 方法
                     * 作用是当创建 NioSocketChannel 成功后，在进行初始化时,将它的ChannelHandler设置到ChannelPipeline中，用于处理网络I/O事件*/
                    Bootstrap b = new Bootstrap();
                    b.group(group).channel(NioSocketChannel.class)
                            .option(ChannelOption.TCP_NODELAY, true)
                            .handler(new ChannelInitializer<SocketChannel>() {
                                @Override
                                public void initChannel(SocketChannel ch) throws Exception {
                                    System.out.println(Thread.currentThread().getName() + ",客户端初始化管道...");
                                    /**
                                     * 与服务器同理
                                     * FixedLengthFrameDecoder(int frameLength)： frameLength：指定单条消息的长度
                                     * 为了测试结果明显，这里设置消息长度为 64 字节，实际应用中应该根据时间情况进行设置，比如 1024 字节
                                     */
                                    ch.pipeline().addLast(new FixedLengthFrameDecoder(64));
                                    ch.pipeline().addLast(new StringDecoder());
                                    ch.pipeline().addLast(new FixedClientHandler());
                                }
                            });
    
                    /**connect：发起异步连接操作，调用同步方法 sync 等待连接成功*/
                    ChannelFuture channelFuture = b.connect(host, port).sync();
                    System.out.println(Thread.currentThread().getName() + ",客户端发起异步连接..........");
    
                    /**等待客户端链路关闭*/
                    channelFuture.channel().closeFuture().sync();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally {
                    /**优雅退出，释放NIO线程组*/
                    group.shutdownGracefully();
                }
            }
        }
    }

重点在第 56-61，82行，注释中已经说明。

# 运行结果 #

先运行服务器，在运行客户端，输出结果如下：

> 客户端控制台输出：
> 
> nioEventLoopGroup-3-1,客户端初始化管道...  
> nioEventLoopGroup-2-1,客户端初始化管道...  
> Thread-1,客户端发起异步连接..........  
> Thread-0,客户端发起异步连接..........  
> 1---nioEventLoopGroup-3-1,Server return Message：1-我是客户端 nioEventLoopGroup-3-1  
> 2---nioEventLoopGroup-2-1,Server return Message：1-我是客户端 nioEventLoopGroup-2-1  
> 3---nioEventLoopGroup-3-1,Server return Message：2-我是客户端 nioEventLoopGroup-3-1  
> 4---nioEventLoopGroup-2-1,Server return Message：2-我是客户端 nioEventLoopGroup-2-1  
> 5---nioEventLoopGroup-3-1,Server return Message：3-我是客户端 nioEventLoopGroup-3-1  
> 6---nioEventLoopGroup-2-1,Server return Message：3-我是客户端 nioEventLoopGroup-2-1  
> 7---nioEventLoopGroup-3-1,Server return Message：4-我是客户端 nioEventLoopGroup-3-1  
> 8---nioEventLoopGroup-2-1,Server return Message：4-我是客户端 nioEventLoopGroup-2-1  
> 9---nioEventLoopGroup-3-1,Server return Message：5-我是客户端 nioEventLoopGroup-3-1  
> 10---nioEventLoopGroup-2-1,Server return Message：5-我是客户端 nioEventLoopGroup-2-1  
> 11---nioEventLoopGroup-3-1,Server return Message：0123456789012345678901234567890123456789012345678901234567891234  
> 12---nioEventLoopGroup-2-1,Server return Message：0123456789012345678901234567890123456789012345678901234567891234  
> 13---nioEventLoopGroup-3-1,Server return Message：7-我是客户端 nioEventLoopGroup-3-1  
> 14---nioEventLoopGroup-2-1,Server return Message：7-我是客户端 nioEventLoopGroup-2-1  
> 15---nioEventLoopGroup-3-1,Server return Message：8-我是客户端 nioEventLoopGroup-3-1  
> 16---nioEventLoopGroup-2-1,Server return Message：8-我是客户端 nioEventLoopGroup-2-1  
> 17---nioEventLoopGroup-3-1,Server return Message：9-我是客户端 nioEventLoopGroup-3-11234567890abcdefghijklmno  
> 18---nioEventLoopGroup-2-1,Server return Message：9-我是客户端 nioEventLoopGroup-2-11234567890abcdefghijklmno  
> 19---nioEventLoopGroup-3-1,Server return Message：pqrstuvwxyz10-我是客户端 nioEventLoopGroup-3-1  
> 20---nioEventLoopGroup-2-1,Server return Message：pqrstuvwxyz10-我是客户端 nioEventLoopGroup-2-1
> 
> 服务器控制台输出：
> 
> main,服务器开始监听端口，等待客户端连接.........  
> nioEventLoopGroup-3-1,服务器初始化通道...  
> nioEventLoopGroup-3-2,服务器初始化通道...  
> 1--->nioEventLoopGroup-3-1,The server receive  order : 1-我是客户端 nioEventLoopGroup-3-1  
> 2--->nioEventLoopGroup-3-2,The server receive  order : 1-我是客户端 nioEventLoopGroup-2-1  
> 4--->nioEventLoopGroup-3-1,The server receive  order : 2-我是客户端 nioEventLoopGroup-3-1  
> 3--->nioEventLoopGroup-3-2,The server receive  order : 2-我是客户端 nioEventLoopGroup-2-1  
> 5--->nioEventLoopGroup-3-1,The server receive  order : 3-我是客户端 nioEventLoopGroup-3-1  
> 6--->nioEventLoopGroup-3-2,The server receive  order : 3-我是客户端 nioEventLoopGroup-2-1  
> 8--->nioEventLoopGroup-3-1,The server receive  order : 4-我是客户端 nioEventLoopGroup-3-1  
> 7--->nioEventLoopGroup-3-2,The server receive  order : 4-我是客户端 nioEventLoopGroup-2-1  
> 9--->nioEventLoopGroup-3-1,The server receive  order : 5-我是客户端 nioEventLoopGroup-3-1  
> 10--->nioEventLoopGroup-3-2,The server receive  order : 5-我是客户端 nioEventLoopGroup-2-1  
> 11--->nioEventLoopGroup-3-1,The server receive  order : 0123456789012345678901234567890123456789012345678901234567891234  
> 12--->nioEventLoopGroup-3-2,The server receive  order : 0123456789012345678901234567890123456789012345678901234567891234  
> 13--->nioEventLoopGroup-3-1,The server receive  order : 7-我是客户端 nioEventLoopGroup-3-1  
> 14--->nioEventLoopGroup-3-2,The server receive  order : 7-我是客户端 nioEventLoopGroup-2-1  
> 15--->nioEventLoopGroup-3-1,The server receive  order : 8-我是客户端 nioEventLoopGroup-3-1  
> 16--->nioEventLoopGroup-3-2,The server receive  order : 8-我是客户端 nioEventLoopGroup-2-1  
> 17--->nioEventLoopGroup-3-1,The server receive  order : 9-我是客户端 nioEventLoopGroup-3-11234567890abcdefghijklmno  
> 18--->nioEventLoopGroup-3-2,The server receive  order : 9-我是客户端 nioEventLoopGroup-2-11234567890abcdefghijklmno  
> 19--->nioEventLoopGroup-3-1,The server receive  order : pqrstuvwxyz10-我是客户端 nioEventLoopGroup-3-1  
> 20--->nioEventLoopGroup-3-2,The server receive  order : pqrstuvwxyz10-我是客户端 nioEventLoopGroup-2-1

从输出可以看出 FixedLengthFrameDecoder  完全按着约定的长度进行解码消息，如上所示对于小于等于 64 自己的消息接收正确，对于第 9  条消息长度大于 64 字节时，显然它还有一部分被发到了第 10 条消息中去了。

对于数组填补空元素，提供以下两种思路：

import java.io.UnsupportedEncodingException;
    /**
     * Created by Administrator on 2018/10/21 0021.
     */
    public class Text {
        public static void main(String[] args) throws InterruptedException {
            try {
                /**
                 * 方式一：循环遍历，对于不满的，直接添加空格补齐
                 */
                String s1 = "我是客户端-123";
                byte[] bytes1 = s1.getBytes("UTF-8");
                byte[] msgBytes1 = new byte[64];
                for (int i = 0; i < msgBytes1.length; i++) {
                    if (i < bytes1.length) {
                        msgBytes1[i] = bytes1[i];
                    } else {
                        /**32 表示空格，等价于：msgBytes1[i] = " ".getBytes()[0];*/
                        msgBytes1[i] = 32;
                    }
                }
                System.out.println(new String(msgBytes1, "UTF-8") + "," + msgBytes1.length);
    
                /**
                 * 方式二：使用 System.arraycopy 快速复制数组
                 */
                String s2 = "我是客户端-123";
                byte[] bytes2 = s2.getBytes("UTF-8");
                byte[] msgBytes2 = new byte[64];
                System.arraycopy(bytes2, 0, msgBytes2, 0, bytes2.length);
                System.out.println(new String(msgBytes2) + "," + msgBytes2.length);
            } catch (UnsupportedEncodingException e) {
                e.printStackTrace();
            }
        }
    }

> 控制台输出：
> 
> 我是客户端-123                                             ,64  
> 我是客户端-123                                             ,64

[DelimiterBasedFrameDecoder _ TCP]: https://blog.csdn.net/wangmx1993328/article/details/83956282
[20181111123055137.png]: /images/20220417/545dcc9d9ad147ecb84273cfd9dd7eee.png