自定义分隔符解码器04

Posted 2021-11-27 lazyli

分隔符和定长解码器
1、消息长度固定，累计读取到长度总和为定长LEN的报文后，就认为读取到了一个完整的消息；将计数置位，重新开始读取下一个数据报。
2、将回车换行符作为消息结束符，例如FTP协议，这种方式在文本协议中应用比较广泛
3、将特殊的分隔符作为消息的结束标志，回车换行符就是一种特殊的结束分隔符
4、通过在消息头中定义长度字段来标识消息的总长度。
DelimiterBasedFrameDecoder：可以自动完成以分隔符结束标志的消息的解码，

服务端

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, 100)
                .handler(new LoggingHandler(LogLevel.INFO))
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
                        //自定义特殊符号,以"$_"为结束符
                        ByteBuf delimiter = Unpooled.copiedBuffer("$_".getBytes());
                        ch.pipeline().addLast(new DelimiterBasedFrameDecoder(1024,delimiter));
                        ch.pipeline().addLast(new StringDecoder());
                        ch.pipeline().addLast(new EchoServerHandler());
                    }
                });
            //绑定端口，同步等待成功
            ChannelFuture f = b.bind(port).sync();
            //等待服务端监听端口关闭
            f.channel().closeFuture().sync();
        } catch (Exception e) {
        }finally {
            //释放资源
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
    
    public static void main(String[] args) throws Exception {
        int port = 8088;
        new EchoServer().bind(port);
    }
}

服务端处理类

public class EchoServerHandler extends ChannelHandlerAdapter {
    private int counter;
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        //获取客户端消息(msg就是客户端发过来的消息)
        String body = (String)msg;
        System.out.println("This is "+ (++counter) + 
                " times receive client:" + "[" +body + "]");
        body = body + "$_";
        ByteBuf echo = Unpooled.copiedBuffer(body.getBytes());
        //向客户端发送消息
        System.out.println("准备发送");
        //Thread.sleep(5000);
        ctx.writeAndFlush(echo);
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.close();//发生异常，关闭链路
    }
}

客户端

public class EchoClient {
    
    public void connect(String host,int port) throws Exception{
        //配置客户端NIO线程组
        EventLoopGroup group = new NioEventLoopGroup();
        try {
            Bootstrap b = new Bootstrap();
            b.group(group).channel(NioSocketChannel.class)
                .option(ChannelOption.TCP_NODELAY, true)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
                        ByteBuf delimiter = Unpooled.copiedBuffer("$_".getBytes());
                        ch.pipeline().addLast(new DelimiterBasedFrameDecoder(1024,delimiter));
                        ch.pipeline().addLast(new StringDecoder());
                        ch.pipeline().addLast(new EchoClientHandler());
                    }
                });
            
            //发起异步连接操作
            ChannelFuture f = b.connect(host,port);
            //等待客户端链路关闭
            f.channel().closeFuture().sync();
        } finally {
            //释放NIO线程组
            group.shutdownGracefully();
        }
    }
    
    public static void main(String[] args) throws Exception {
        String host = "localhost";
        int port = 8088;
        new EchoClient().connect(host, port);
    }
}

客户端处理类

public class EchoClientHandler extends ChannelHandlerAdapter {

    private int counter;
    
    static final String ECHO_REQ = "Hi,Lilinfeng. welcome to Netty.$_";
    
    public EchoClientHandler(){
        
    }
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        for(int i = 0 ;i < 10 ; i++){
            //向服务端发送信息
            ctx.writeAndFlush(Unpooled.copiedBuffer(ECHO_REQ.getBytes()));
        }
    }
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        //打印从服务端接收的消息
        System.out.println("This is " + (++counter) + 
                " times receive server:[" + msg + "]");
    }
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        ctx.flush();
    }
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.close();
    }
}