[netty]--分隔符解码器DelimiterBasedFrameDecoder和定长解码器FixedLengthFrameDecoder

桃扇骨 2022-07-11 13:42 238阅读 0赞

TCP以字节流的方式进行数据传输，上层应用协议为了对消息进行区分，往往采用如下4种方式。

1.  消息长度固定：累计读取到固定长度为LENGTH之后就认为读取到了一个完整的消息。然后将计数器复位，重新开始读下一个数据报文。
2.  回车换行符作为消息结束符：在文本协议中应用比较广泛。
3.  将特殊的分隔符作为消息的结束标志，回车换行符就是一种特殊的结束分隔符。
4.  通过在消息头中定义长度字段来标示消息的总长度。

netty中针对这四种场景均有对应的解码器作为解决方案。

本问我们继续介绍DelimiterBasedFrameDecoder和FixedLengthFrameDecoder分别来解决以特殊分隔符作为消息的结束标志的解码和定长消息的解码。他们均能解决TCP导致的黏包读半包问题。

# 1.DelimiterBasedFrameDecoder应用开发 #

我们以Echo服务器来演示DelimiterBasedFrameDecoder自动完成以分隔符作为码流结束标识消息结束的解码。

EchoServer收到EchoClient的请求消息后，将其打印出来，然后将原始消息返回客户端。消息以”$\_” 作为分隔符。

## 服务端 ##

服务端主要加了DelimiterBasedFrameDecoder，并在构造器中配置消息的最大长度是1024，并设置行分隔符是”$\_”。

package netty.quanwei.p5;
    
    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;
    
    /** * Created by louyuting on 16/11/27. */
    public class EchoServer { 
        private final int port;//定义服务器端监听的端口
        /** 构造函数中传入参数 **/
        public EchoServer(int port){
            this.port = port;
        }
    
        /** 启动服务器 **/
        public void start() throws Exception{
            //县城组
            EventLoopGroup boss = new NioEventLoopGroup();
            EventLoopGroup worker = new NioEventLoopGroup();
    
            //创建一个serverbootstrap实例
            ServerBootstrap serverBootstrap = new ServerBootstrap();
    
            try {
                serverBootstrap.group(boss, worker)
                        .channel(NioServerSocketChannel.class)//指定使用一个NIO传输Channel
                        .option(ChannelOption.SO_BACKLOG, 100)
                        .childHandler(new ChannelInitializer<SocketChannel>() {
                            //在channel的ChannelPipeline中加入EchoServerHandler到最后
                            @Override
                            protected void initChannel(SocketChannel channel) throws Exception {
                                ByteBuf delimiter = Unpooled.copiedBuffer("$_".getBytes());
                                channel.pipeline().addLast(new DelimiterBasedFrameDecoder(1024,delimiter));
    
                                channel.pipeline().addLast(new StringDecoder());
    
                                channel.pipeline().addLast(new EchoServerHandler());
                            }
                        });
                //异步的绑定服务器,sync()一直等到绑定完成.
                ChannelFuture future = serverBootstrap.bind(this.port).sync();
                System.out.println(EchoServer.class.getName()+" started and listen on '"+ future.channel().localAddress());
                future.channel().closeFuture().sync();//获得这个channel的CloseFuture,阻塞当前线程直到关闭操作完成
            } finally {
                boss.shutdownGracefully().sync();//关闭group,释放所有的资源
                worker.shutdownGracefully().sync();//关闭group,释放所有的资源
            }
        }
    
    
        /** * main * @param args * @throws Exception */
        public static void main(String[] args) throws Exception {
            new EchoServer(8000).start();
        }
    
    }

handler：

package netty.quanwei.p5;
    
    import io.netty.buffer.ByteBuf;
    import io.netty.buffer.Unpooled;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.ChannelInboundHandlerAdapter;
    
    /** * * Created by louyuting on 16/11/27. * 服务器端向客户端传输数据...... * */
    public class EchoServerHandler extends ChannelInboundHandlerAdapter{ 
    
        private int counter=0;
        /** * 每次收到消息的时候被调用; * @param ctx * @param msg * @throws Exception */
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            String body = (String)msg;
    
            System.out.println("this is:"+ (++counter) +" time." + " Server received: " + body);
    
            body = body + "$_";
    
            ByteBuf echo = Unpooled.copiedBuffer(body.getBytes());
            ctx.writeAndFlush(echo);
        }
        /** * 在读操作异常被抛出时被调用 * @param ctx * @param cause * @throws Exception */
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            cause.printStackTrace();//打印异常的堆栈跟踪信息
            ctx.close();//关闭这个channel
        }
    }

handler 中直接把收到的消息echo回客户端，并在服务端做了一个计数，用来验证TCP的黏包是否被解决。

## 客户端的实现： ##

客户端和服务端基本一致的思路：

package netty.quanwei.p5;
    
    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;
    
    import java.net.InetSocketAddress;
    
    /** * Created by louyuting on 16/11/27. */
    public class EchoClient { 
        private final String host;
        private final int port;//定义服务器端监听的端口
        /** 构造函数中传入参数 **/
        public EchoClient(String host, int port){
            this.host = host;
            this.port = port;
        }
    
        /** 启动服务器 **/
        public void start() throws Exception{
            EventLoopGroup group = new NioEventLoopGroup();
            //创建一个client 的bootstrap实例
            Bootstrap clientBootstrap = new Bootstrap();
    
            try {
                clientBootstrap.group(group)
                        .channel(NioSocketChannel.class)//指定使用一个NIO传输Channel
                        .remoteAddress(new InetSocketAddress(host, port))//设置远端服务器的host和端口
                        .option(ChannelOption.TCP_NODELAY, true)
                        .handler(new ChannelInitializer<SocketChannel>() {
                            //在channel的ChannelPipeline中加入EchoClientHandler到最后
                            @Override
                            protected void initChannel(SocketChannel channel) throws Exception {
                                ByteBuf delimiter = Unpooled.copiedBuffer("$_".getBytes());
                                channel.pipeline().addLast(new DelimiterBasedFrameDecoder(1024,delimiter));
    
                                channel.pipeline().addLast(new StringDecoder());
                                channel.pipeline().addLast(new EchoClientHandler());
                            }
                        });
                ChannelFuture f = clientBootstrap.connect().sync();//连接到远端,一直等到连接完成
                f.channel().closeFuture().sync();//一直阻塞到channel关闭
            } finally {
                group.shutdownGracefully().sync();//关闭group,释放所有的资源
            }
        }
    
    
        /** * main * @param args * @throws Exception */
        public static void main(String[] args) throws Exception {
            new EchoClient("127.0.0.1", 8000).start();
        }
    
    }

handler: 客户端的handler中，在建立链路时发送10个请求给服务端

package netty.quanwei.p5;
    
    import io.netty.buffer.Unpooled;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.SimpleChannelInboundHandler;
    
    /** * * Created by louyuting on 16/12/1. * */
    public class EchoClientHandler extends SimpleChannelInboundHandler<String>{ 
    
        private int counter=0;
    
        private static final String REQ = "LOUYYUTING netty. $_";
    
        /** * 当收到连接成功的通知,发送一条消息. * @param ctx * @throws Exception */
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
    
            for(int i=0; i<10; i++){
                ctx.writeAndFlush( Unpooled.copiedBuffer(REQ.getBytes()) );
            }
        }
    
        /** * 每当收到数据时这个方法会被调用.打印收到的消息日志 * @param channelHandlerContext * @param msg * @throws Exception */
        @Override
        protected void channelRead0(ChannelHandlerContext channelHandlerContext, String msg) throws Exception {
            System.out.println("client received: " + "counter:" + (++counter) + " msg:"+msg);
        }
    
        /** * 异常发生时,记录错误日志,关闭channel * @param ctx * @param cause * @throws Exception */
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            cause.printStackTrace();//打印堆栈的错误日志
            ctx.close();
        }
    }

## 运行结果： ##

服务端：

this is:1 time. Server received: LOUYYUTING netty. 
    this is:2 time. Server received: LOUYYUTING netty. 
    this is:3 time. Server received: LOUYYUTING netty. 
    this is:4 time. Server received: LOUYYUTING netty. 
    this is:5 time. Server received: LOUYYUTING netty. 
    this is:6 time. Server received: LOUYYUTING netty. 
    this is:7 time. Server received: LOUYYUTING netty. 
    this is:8 time. Server received: LOUYYUTING netty. 
    this is:9 time. Server received: LOUYYUTING netty. 
    this is:10 time. Server received: LOUYYUTING netty.

服务端成功获取到10次请求。

客户端：

client received: counter:1  msg:LOUYYUTING netty. 
    client received: counter:2  msg:LOUYYUTING netty. 
    client received: counter:3  msg:LOUYYUTING netty. 
    client received: counter:4  msg:LOUYYUTING netty. 
    client received: counter:5  msg:LOUYYUTING netty. 
    client received: counter:6  msg:LOUYYUTING netty. 
    client received: counter:7  msg:LOUYYUTING netty. 
    client received: counter:8  msg:LOUYYUTING netty. 
    client received: counter:9  msg:LOUYYUTING netty. 
    client received: counter:10  msg:LOUYYUTING netty.

客户端也成功接收到了10次服务器的回应。

# 2.FixedLengthFrameDecoder应用开发 #

FixedLengthFrameDecoder是一个固定长度解码器，它可以按照指定的长度对消息进行解码，开发者不需要关心TCP的黏包和拆包问题，非常实用，下面也是通过实例讲解：

在第一节中的Echo服务器中做一点小小改动来实现：  
仅仅只把解码器DelimiterBasedFrameDecoder换成FixedLengthFrameDecoder。

源码如下：  
服务端：

package netty.quanwei.p5_2;
    
    import io.netty.bootstrap.ServerBootstrap;
    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.FixedLengthFrameDecoder;
    import io.netty.handler.codec.string.StringDecoder;
    
    /** * Created by louyuting on 16/11/27. */
    public class EchoServer { 
        private final int port;//定义服务器端监听的端口
        /** 构造函数中传入参数 **/
        public EchoServer(int port){
            this.port = port;
        }
    
        /** 启动服务器 **/
        public void start() throws Exception{
            //县城组
            EventLoopGroup boss = new NioEventLoopGroup();
            EventLoopGroup worker = new NioEventLoopGroup();
    
            //创建一个serverbootstrap实例
            ServerBootstrap serverBootstrap = new ServerBootstrap();
    
            try {
                serverBootstrap.group(boss, worker)
                        .channel(NioServerSocketChannel.class)//指定使用一个NIO传输Channel
                        .option(ChannelOption.SO_BACKLOG, 100)
                        .childHandler(new ChannelInitializer<SocketChannel>() {
                            //在channel的ChannelPipeline中加入EchoServerHandler到最后
                            @Override
                            protected void initChannel(SocketChannel channel) throws Exception {
                                channel.pipeline().addLast(new FixedLengthFrameDecoder(10));
    
                                channel.pipeline().addLast(new StringDecoder());
    
                                channel.pipeline().addLast(new EchoServerHandler());
                            }
                        });
                //异步的绑定服务器,sync()一直等到绑定完成.
                ChannelFuture future = serverBootstrap.bind(this.port).sync();
                System.out.println(EchoServer.class.getName()+" started and listen on '"+ future.channel().localAddress());
                future.channel().closeFuture().sync();//获得这个channel的CloseFuture,阻塞当前线程直到关闭操作完成
            } finally {
                boss.shutdownGracefully().sync();//关闭group,释放所有的资源
                worker.shutdownGracefully().sync();//关闭group,释放所有的资源
            }
        }
    
    
        /** * main * @param args * @throws Exception */
        public static void main(String[] args) throws Exception {
            new EchoServer(8000).start();
        }
    
    }

handler类：

package netty.quanwei.p5_2;
    
    import io.netty.buffer.ByteBuf;
    import io.netty.buffer.Unpooled;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.ChannelInboundHandlerAdapter;
    
    /** * * Created by louyuting on 16/11/27. * 服务器端向客户端传输数据...... * */
    public class EchoServerHandler extends ChannelInboundHandlerAdapter{ 
    
        private int counter=0;
        /** * 每次收到消息的时候被调用; * @param ctx * @param msg * @throws Exception */
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            String body = (String)msg;
    
            System.out.println("this is:"+ (++counter) +" time." + " Server received: " + body);
    
            ByteBuf echo = Unpooled.copiedBuffer(body.getBytes());
    
            ctx.writeAndFlush(echo);
        }
    
    
        /** * 在读操作异常被抛出时被调用 * @param ctx * @param cause * @throws Exception */
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            cause.printStackTrace();//打印异常的堆栈跟踪信息
            ctx.close();//关闭这个channel
        }
    }

客户端：

package netty.quanwei.p5_2;
    
    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;
    
    import java.net.InetSocketAddress;
    
    /** * Created by louyuting on 16/11/27. */
    public class EchoClient { 
        private final String host;
        private final int port;//定义服务器端监听的端口
        /** 构造函数中传入参数 **/
        public EchoClient(String host, int port){
            this.host = host;
            this.port = port;
        }
    
        /** 启动服务器 **/
        public void start() throws Exception{
            EventLoopGroup group = new NioEventLoopGroup();
            //创建一个client 的bootstrap实例
            Bootstrap clientBootstrap = new Bootstrap();
    
            try {
                clientBootstrap.group(group)
                        .channel(NioSocketChannel.class)//指定使用一个NIO传输Channel
                        .remoteAddress(new InetSocketAddress(host, port))//设置远端服务器的host和端口
                        .option(ChannelOption.TCP_NODELAY, true)
                        .handler(new ChannelInitializer<SocketChannel>() {
                            //在channel的ChannelPipeline中加入EchoClientHandler到最后
                            @Override
                            protected void initChannel(SocketChannel channel) throws Exception {
                                channel.pipeline().addLast(new FixedLengthFrameDecoder(10));
    
    
                                channel.pipeline().addLast(new StringDecoder());
                                channel.pipeline().addLast(new EchoClientHandler());
                            }
                        });
                ChannelFuture f = clientBootstrap.connect().sync();//连接到远端,一直等到连接完成
                f.channel().closeFuture().sync();//一直阻塞到channel关闭
            } finally {
                group.shutdownGracefully().sync();//关闭group,释放所有的资源
            }
        }
    
    
        /** * main * @param args * @throws Exception */
        public static void main(String[] args) throws Exception {
            new EchoClient("127.0.0.1", 8000).start();
        }
    
    }

handler类：

package netty.quanwei.p5_2;
    
    import io.netty.buffer.Unpooled;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.SimpleChannelInboundHandler;
    
    /** * * Created by louyuting on 16/12/1. * */
    public class EchoClientHandler extends SimpleChannelInboundHandler<String>{ 
    
        private int counter=0;
    
        private static final String REQ = "LOUYUTING.";
    
        /** * 当收到连接成功的通知,发送一条消息. * @param ctx * @throws Exception */
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
    
            for(int i=0; i<10; i++){
                ctx.writeAndFlush( Unpooled.copiedBuffer(REQ.getBytes()) );
            }
        }
    
        /** * 每当收到数据时这个方法会被调用.打印收到的消息日志 * @param channelHandlerContext * @param msg * @throws Exception */
        @Override
        protected void channelRead0(ChannelHandlerContext channelHandlerContext, String msg) throws Exception {
            System.out.println("client received: " + "counter:" + (++counter) + " msg:"+msg);
        }
    
        /** * 异常发生时,记录错误日志,关闭channel * @param ctx * @param cause * @throws Exception */
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            cause.printStackTrace();//打印堆栈的错误日志
            ctx.close();
        }
    
    }

### 运行结果： ###

server:

this is:1 time. Server received: LOUYUTING.
    this is:2 time. Server received: LOUYUTING.
    this is:3 time. Server received: LOUYUTING.
    this is:4 time. Server received: LOUYUTING.
    this is:5 time. Server received: LOUYUTING.
    this is:6 time. Server received: LOUYUTING.
    this is:7 time. Server received: LOUYUTING.
    this is:8 time. Server received: LOUYUTING.
    this is:9 time. Server received: LOUYUTING.
    this is:10 time. Server received: LOUYUTING.

client端

client received: counter:1  msg:LOUYUTING.
    client received: counter:2  msg:LOUYUTING.
    client received: counter:3  msg:LOUYUTING.
    client received: counter:4  msg:LOUYUTING.
    client received: counter:5  msg:LOUYUTING.
    client received: counter:6  msg:LOUYUTING.
    client received: counter:7  msg:LOUYUTING.
    client received: counter:8  msg:LOUYUTING.
    client received: counter:9  msg:LOUYUTING.
    client received: counter:10  msg:LOUYUTING.

我们上面定义的协议定长是10个字节，所以每次发送的都是以10个字节为一个完整的数据包。

所以FixedLengthFrameDecoder解码器无论一次接收到多少字节的数据包，都会按照构造函数中设定的字节数进行解码。如果数据字节数不到指定的定长（半包），FixedLengthFrameDecoder会缓存半包消息等待下个包到达后进行拼包。

上面两节的完整代码的github地址：  
DelimiterBasedFrameDecoder：[https://github.com/leetcode-hust/leetcode/tree/master/louyuting/src/netty/quanwei/p5][https_github.com_leetcode-hust_leetcode_tree_master_louyuting_src_netty_quanwei_p5]

FixedLengthFrameDecoder：[https://github.com/leetcode-hust/leetcode/tree/master/louyuting/src/netty/quanwei/p5\_2][https_github.com_leetcode-hust_leetcode_tree_master_louyuting_src_netty_quanwei_p5_2]

[https_github.com_leetcode-hust_leetcode_tree_master_louyuting_src_netty_quanwei_p5]: https://github.com/leetcode-hust/leetcode/tree/master/louyuting/src/netty/quanwei/p5
[https_github.com_leetcode-hust_leetcode_tree_master_louyuting_src_netty_quanwei_p5_2]: https://github.com/leetcode-hust/leetcode/tree/master/louyuting/src/netty/quanwei/p5_2