使用DelimiterBasedFrameDecoder进行消息分隔

港控/mmm° 2022-03-29 12:26 260阅读 0赞

在使用Netty进行TCP消息传输时，为了上层协议能够对消息正确区分，避免粘包和拆包导致的问题，一般可以通过消息定长、将回车换行符作为消息结束符、将特殊的分隔符作为消息的结束标志或者在消息头中定义长度字段来标识消息的总长度。其中常用的通过分隔符作为消息的结束标志就涉及到Netty的DelimiterBasedFrameDecoder类,服务端如下：

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.string.StringDecoder;
    import io.netty.handler.logging.LogLevel;
    import io.netty.handler.logging.LoggingHandler;
    
    public class EchoServer
    {
        public void bind(int port)throws Exception{
            //配置服务端的NIO线程组
            EventLoopGroup bossGroup=new NioEventLoopGroup();
            EventLoopGroup workerGroup=new NioEventLoopGroup();
            try
            {
                ServerBootstrap b=new ServerBootstrap();
                b.group(bossGroup, workerGroup)
                .channel(NioServerSocketChannel.class)
                .option(ChannelOption.SO_BACKLOG, 1024)
                //控制台输出服务端运行日志
                .handler(new LoggingHandler(LogLevel.INFO))
                //编写服务端接收和发送消息的具体逻辑
                .childHandler(new ChildChannleHandler());
                //绑定启动端口，同步等待成功
                ChannelFuture f=b.bind(port).sync();
                //等待服务端监听端口关闭
                f.channel().closeFuture().sync();
            }
            catch (Exception e)
            {
                e.printStackTrace();
            }
            finally{
                //释放线程资源
                bossGroup.shutdownGracefully();
                workerGroup.shutdownGracefully();
            }
        }
        //服务端接收到客户端的消息时会先执行该类的initChannel()方法进行channel的初始化操作
        private class ChildChannleHandler extends ChannelInitializer<SocketChannel>{
            @Override
            protected void initChannel(SocketChannel arg0)
                throws Exception
            {
                //创建分隔符缓冲对象，使用"$_"作为分隔符
                ByteBuf delimiter=Unpooled.copiedBuffer("$_".getBytes());
                //创建DelimiterBasedFrameDecoder对象，将其加入到ChannelPipeline
                //参数1024表示单条消息的最大长度，当达到该长度仍然没有找到分隔符就抛出TooLongFrame异常，第二个参数就是分隔符
                //由于DelimiterBasedFrameDecoder自动对请求消息进行了解码，下面的ChannelHandler接收到的msg对象就是完整的消息包
                arg0.pipeline().addLast(new DelimiterBasedFrameDecoder(1024, delimiter));
                //StringDecoder解码器将ByteBuf解码成字符串对象，这样在ChannelHandlerAdapter中读取消息时就不需要通过ByteBuf获取了
                arg0.pipeline().addLast(new StringDecoder());
                //对网络事件进行读写操作的类
                arg0.pipeline().addLast(new EchoServerHandler());
            }
        }
        
        public static void main(String[] args)throws Exception
        {
            int port =8888;
            if (args!=null&&args.length>0)
            {
                port=Integer.valueOf(args[0]);
            }
            new EchoServer().bind(port);
        }
    }

服务端消息读写操作：

import io.netty.buffer.ByteBuf;
    import io.netty.buffer.Unpooled;
    import io.netty.channel.ChannelHandlerAdapter;
    import io.netty.channel.ChannelHandlerContext;
    //网络I/O事件读写操作
    public class EchoServerHandler extends ChannelHandlerAdapter
    {
        int counter=0;
        
        //接收客户端发送的消息并返回响应
        @Override
        public void channelRead(ChannelHandlerContext ctx,Object msg)throws Exception{
            //获取String类型的请求消息（StringDecoder已经对消息进行解码）
            String body=(String)msg;
            System.out.println("This is "+ ++counter+"times receive client : ["+body+"]");
            //由于设置了DelimiterBasedFrameDecoder过滤掉了分隔符"$_"，   因此需要将返回消息尾部拼接上分隔符
            body+="$_";
            //将接收到的消息再放到ByteBuf中重新发送给客户端
            ByteBuf buf=Unpooled.copiedBuffer(body.getBytes());
            //把待发送的消息放到发送缓冲数组中，并把缓冲区中的消息全部写入SockChannel发送给客户端
            ctx.writeAndFlush(buf);
        }
        
        //发生异常时关闭ChannelHandlerContext，释放和ChannelHandlerContext相关联的句柄等资源
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx,Throwable cause){
            cause.printStackTrace();
            ctx.close();
        }
    }

客户端：

import io.netty.bootstrap.Bootstrap;
    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.NioSocketChannel;
    import io.netty.handler.codec.DelimiterBasedFrameDecoder;
    import io.netty.handler.codec.string.StringDecoder;
    
    //客户端
    public class EchoClient
    {
        public void connect(int port,String host)throws Exception{
            //创建客户端进行I/O读写的线程组
            EventLoopGroup g=new NioEventLoopGroup();
            try
            {
                //创建客户端启动辅助类Bootstrap
                Bootstrap b=new Bootstrap();
                b.group(g)
                //设置Channel
                .channel(NioSocketChannel.class)
                //配置Channel
                .option(ChannelOption.TCP_NODELAY, true)
                //添加处理类，这里为了方便直接使用了匿名内部类
                .handler(new ChannelInitializer<SocketChannel>()
                {
                    //当创建NioSocketChannel成功后，将ChannelHandler设置到ChannelPipeline中处理网络I/O事件
                    @Override
                    protected void initChannel(SocketChannel arg0)
                        throws Exception
                    {
                        //与服务端相同，需要配置一系列的ChannelHandler
                        ByteBuf delimiter=Unpooled.copiedBuffer("$_".getBytes());
                        arg0.pipeline().addLast(new DelimiterBasedFrameDecoder(1024,delimiter));
                        arg0.pipeline().addLast(new StringDecoder());
                        //客户端的处理类加入ChannelPipeline
                        arg0.pipeline().addLast(new EchoClientHandler());
                    }
                });
                //调用connect方法发起异步连接，并调用同步方法等待连接成功
                ChannelFuture f=b.connect(host, port).sync();
                //f.channel().writeAndFlush(Unpooled.wrappedBuffer("111$_".getBytes()));
                //等待客户端连接关闭
                f.channel().closeFuture().sync();
            }
            catch (Exception e)
            {
                e.printStackTrace();
            }
            finally{
                //释放线程组
                g.shutdownGracefully();
            }
            
        }
        public static void main(String[] args)throws Exception
        {
            int port=8888;
            if (args!=null&&args.length>0)
            {
                port=Integer.valueOf(args[0]);
            }
            new EchoClient().connect(port, "127.0.0.1");
        }
    }

客户端网络I/O事件处理：

import io.netty.buffer.Unpooled;
    import io.netty.channel.ChannelHandlerAdapter;
    import io.netty.channel.ChannelHandlerContext;
    
    //客户端读写网络I/O事件类
    public class EchoClientHandler extends ChannelHandlerAdapter
    {
        int counter;
        
        //发送到服务端的消息，注意结尾的分隔符一定要和服务端配置的分隔符一致，否则服务端ChannelInitializer.initChannel()方法虽然能够调用，但是DelimiterBasedFrameDecoder无法找到分隔符，不会调用读取消息的channelRead方法
        static final String ECHO_REQ="Hi,Welcome to Netty.$_";
        
        public EchoClientHandler(){
            
        }
        
        //客户端发送消息的方法
        @Override
        public void channelActive(ChannelHandlerContext ctx)throws Exception{
            for (int i = 0; i < 10; i++ )
            {
                //Unpooled.copiedBuffer()方法是深克隆，也可以使用Unpooled.buffer()写入消息发送
                ctx.writeAndFlush(Unpooled.copiedBuffer(ECHO_REQ.getBytes()));
            }
        }
        
        //读取服务端发送的消息
        @Override
        public void channelRead(ChannelHandlerContext ctx,Object msg)throws Exception{
            String body=(String)msg;
            System.out.println("This is "+ ++counter+" times receive server:["+body+"]");
        }
        
        @Override
        public void channelReadComplete(ChannelHandlerContext ctx)throws Exception{
            //将消息发送队列中的消息写入到SocketChannel中发送给对方，channelActive使用了writeAndFlush这里可以不重写
            ctx.flush();
        }
        
        //异常处理，关闭ChannelHandlerContext
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx,Throwable cause){
            cause.printStackTrace();
            ctx.close();
        }
    }

启动服务端：

![20190108170013725.png][]

启动客户端发送消息：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2Rvbmd5dXh1MzQyNzE5_size_16_color_FFFFFF_t_70][]

参考书籍《Netty权威指南》第五章

[20190108170013725.png]: /images/20220329/e2c1ff14a3cb47408c2b148ddd6b1195.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2Rvbmd5dXh1MzQyNzE5_size_16_color_FFFFFF_t_70]: /images/20220329/9fbe0feffe2a4a9cb0662b3dd4dc1169.png