netty学习之路三，定长方式处理拆包黏包问题

ゞ浴缸里的玫瑰 2022-03-09 07:26 215阅读 0赞

### 定长方式处理拆包黏包问题 ###

*  server
 *  ServerHandler
 *  Client
 *  ClientHandler

> netty除了提供特殊字符拆包之外，还提供了一种定长方式处理拆包的方法。但是，这种方法每次发送的消息必须是固定长度的，如果长度不够，就会阻塞。所以，和特殊字符拆包相比，定长方式处理拆包，优势不大。

# server #

public class Server {
        public static void main(String[] args) throws Exception{
    
            EventLoopGroup bossGroup = new NioEventLoopGroup(); // 负责处理来自客户端的链接
            EventLoopGroup workerGroup = new NioEventLoopGroup();//负责处理任务
    
            ServerBootstrap bootstrap = new ServerBootstrap();//配置类
            bootstrap.group(bossGroup,workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .option(ChannelOption.SO_SNDBUF, 32*1024)
                    .option(ChannelOption.SO_RCVBUF, 32*1024)
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            //ByteBuf buf = Unpooled.copiedBuffer("$$".getBytes());
                            //socketChannel.pipeline().addLast(new DelimiterBasedFrameDecoder(1024, buf));
                            socketChannel.pipeline().addLast(new FixedLengthFrameDecoder(5));
                            socketChannel.pipeline().addLast(new StringDecoder());
                            socketChannel.pipeline().addLast(new StringEncoder());
                            socketChannel.pipeline().addLast(new ServerHandler());//添加处理器
                        }
                    })
                    .option(ChannelOption.SO_BACKLOG, 128)
                    .childOption(ChannelOption.SO_KEEPALIVE, true);
    
            ChannelFuture future = bootstrap.bind(8888).sync();
            future.channel().closeFuture().sync();//阻塞线程
    
            workerGroup.shutdownGracefully();
            bossGroup.shutdownGracefully();
    
        }
    }

# ServerHandler #

public class ServerHandler extends ChannelHandlerAdapter {
    
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            System.out.println("channelActive is invoked");
        }
    
        /**
         * 收到数据时调用
         * @param ctx
         * @param msg
         * @throws Exception
         */
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            try{
                System.out.println("server channelRead");
                String result = (String)msg;
                result += "$$";
                System.out.println(result);
                ctx.writeAndFlush(result);//.addListener(ChannelFutureListener.CLOSE);
            }finally {
                ReferenceCountUtil.release(msg);
            }
        }
    
        /**
         * 抛出异常时调用
         * @param ctx
         * @param cause
         * @throws Exception
         */
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            cause.printStackTrace();
            ctx.close();
        }
    }

# Client #

public class Client {
        public static void main(String[] args) throws Exception{
            EventLoopGroup workerGroup = new NioEventLoopGroup();//负责处理任务
    
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(workerGroup)
                    .channel(NioSocketChannel.class)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            //ByteBuf buf = Unpooled.copiedBuffer("$$".getBytes());
                            //socketChannel.pipeline().addLast(new DelimiterBasedFrameDecoder(1024, buf));
                            socketChannel.pipeline().addLast(new FixedLengthFrameDecoder(5));
                            socketChannel.pipeline().addLast(new StringDecoder());
                            socketChannel.pipeline().addLast(new StringEncoder());
                            socketChannel.pipeline().addLast(new ClientHandler());
                        }
                    })
                    .option(ChannelOption.SO_BACKLOG, 128)
                    .option(ChannelOption.SO_KEEPALIVE, true);
    
            ChannelFuture f = bootstrap.connect("127.0.0.1",8888).sync();
            f.channel().writeAndFlush("7777");
            f.channel().writeAndFlush("ddd$$");
    
    
            f.channel().closeFuture().sync();
            workerGroup.shutdownGracefully();
    
        }

# ClientHandler #

public class ClientHandler extends ChannelHandlerAdapter {
    
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            System.out.println("client channelActive");
        }
    
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            try {
                System.out.println("client channelRead");
                //String response = (String)msg;
                System.out.println("收到的响应为:"+msg);
            }finally {
                ReferenceCountUtil.release(msg);
            }
        }
    
        @Override
        public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
            System.out.println("client channelReadComplete");
        }
    
        @Override
        public void channelInactive(ChannelHandlerContext ctx) throws Exception {
            System.out.println("client channelInactive");
        }
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            cause.printStackTrace();
            ctx.close();
        }
    
    }