netty学习之路二，特殊字符处理拆包黏包问题

秒速五厘米 2022-03-09 06:44 356阅读 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 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(Unpooled.copiedBuffer(result.getBytes()));//.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 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(Unpooled.copiedBuffer("中国$$china$$".getBytes()));
            f.channel().writeAndFlush(Unpooled.copiedBuffer("ddd$$".getBytes()));
    
    
            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("收到的响应为:"+response);
            }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();
        }
    }