Netty详解（三）：Netty 入门应用

朱雀 2022-01-29 05:09 297阅读 0赞

# 1. Netty服务端开发 #

## TimeServer.java ##

package com.basic.netty.bio;
    
    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;
    
    public class TimeServer { 
        public void bind(int port)throws Exception{ 
            //配置服务端的NIO线程组，包包含了一组NIO线程，专门用于网络事件的处理，
            //实际上它们就是Reactor线程组。
            //这里创建了两个，一个用于服务端接受客户端的连接，
            //另一个用于进行SocketChannel的网络读写。
            EventLoopGroup bossGroup = new NioEventLoopGroup();
            EventLoopGroup workerGroup=new NioEventLoopGroup();
            try{ 
                //Netty用于启动NIO服务端的辅助启动类，目的是降低服务端的开发复杂度
                ServerBootstrap b = new ServerBootstrap();
                //将两个NIO线程组当作入参传递到ServerBootstrap中
                b.group(bossGroup,workerGroup)
                //功能对应于JDK NIO类库中的ServerSocketChannel类
                .channel(NioServerSocketChannel.class)
                //配置TCP参数，这里将backlog设置为1024
                .option(ChannelOption.SO_BACKLOG,1024)
                //绑定I/O事件的处理类ChildChannelHandler，它
                //的作用类似于Reactor模式中的Handler类，主要用于处理网络I/O事件，例如记录日志、对消息进行编解码等。
                .childHandler(new ChildChannelHandler());
                //绑定端口，同步等待成功
                ChannelFuture f=b.bind(port).sync();
                //调用同步阻塞方法sync 等待绑定操作完成，完成后Netty会返回一个ChannelFuture，
                //它的功能类似于JDK 的 java.util.concurrent.Future，主要用于异步操作的通知回调。
                //等待服务端链路关闭之后main函数才退出。
                f.channel().closeFuture().sync();
            }finally{ 
                //优雅退出，翻放线程池资源
                bossGroup.shutdownGracefully();
                workerGroup.shutdownGracefully();
            }
    
        }
        private class ChildChannelHandler extends ChannelInitializer<SocketChannel>{ 
            @Override
            protected void initChannel(SocketChannel arg0) throws Exception { 
                 arg0.pipeline().addLast(new TimeServerHandler());
            }
        }
        public static void main(String[] args) throws Exception{ 
            int port=8080;
            new TimeServer().bind(port);
        }
    }

## TimeServerHandler.java ##

package com.basic.netty.bio;
    
    import io.netty.buffer.ByteBuf;
    import io.netty.buffer.Unpooled;
    import io.netty.channel.ChannelHandlerAdapter;
    import io.netty.channel.ChannelHandlerContext;
    
    //从ChannelHandlerAdapter继承，用于对网络事件进行读写操作，通常只需要关注channelRead和exceptionCaught方法
    public class TimeServerHandler extends ChannelHandlerAdapter{ 
        //channelRead() 该方法在接受到数据的时候自动调用（会存在半包问题）
        @Override
        public void channelRead(ChannelHandlerContext ctx,Object msg)throws Exception{ 
            //类型转换，将msg转换为Netty的ByteBuf，类似JDK的java.nio.ByteBuffer对象
            ByteBuf buf=(ByteBuf)msg;
            //获取缓冲区可读字节数，创建byte数组
            byte[] req=new byte[buf.readableBytes()];
            //将缓冲区字节复制到新建的数组中
            buf.readBytes(req);
            //获取请求消息
            String body=new String(req,"UTF-8");
            System.out.println("The time server receive order : "+body);
            String currentTime="QUERY TIME ORDER".equalsIgnoreCase(body)?new java.util.Date(System.currentTimeMillis()).toString():"BAD ORDER";
            ByteBuf resp =Unpooled.copiedBuffer(currentTime.getBytes());
            ctx.write(resp);
        }
        @Override
        public void channelReadComplete(ChannelHandlerContext ctx)throws Exception{ 
            /*将消息发送队列中的消息写入到SocketChannel中发送给对方。 从性能角度考虑，为了防止频繁地唤醒Selector进行消息发送， Netty的write方法并不直接将消息写入SocketChannel中， 调用write方法只是把待发送的消息放到发送缓冲数组中， 再通过调用flush方法，将发送的缓冲区的消息全部写到SocketChannel中 */
            ctx.flush();
        }
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx,Throwable cause){ 
            //发生异常时，关闭ChannelHandlerContext
            ctx.close();
        }
    }

# 2. Netty客户端开发 #

## TimeClient.java ##

import io.netty.bootstrap.Bootstrap;
    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.channel.ChannelFuture;
    import io.netty.channel.ChannelInitializer;
    public class TimeClient { 
        public void connect(int port,String host) throws Exception{ 
            EventLoopGroup group=new NioEventLoopGroup();
            try{ 
                Bootstrap b=new Bootstrap();
                //Channel需要设置为NioSocketChannel，然后为其添加Handler
                b.group(group).channel(NioSocketChannel.class)
                .option(ChannelOption.TCP_NODELAY,true)
                .handler(new ChannelInitializer<SocketChannel>(){ 
                    //为了简单直接创建匿名内部类，实现initChannel方法
                    //其作用是当创建NioSocketChannel成功之后，在进行初始化时，
                    //将它的ChannelHandler设置到ChannelPipeline中，用于处理网络I/O事件
                    @Override
                    public void initChannel(SocketChannel ch) throws Exception{ 
                        ch.pipeline().addLast(new TimeClientHandler());
                    }
                });
                //发起异步连接，然后调用同步方法等待连接成功
                ChannelFuture f=b.connect(host,port).sync();
                //当客户端连接关闭之后，客户端主函数退出，退出前释放NIO线程组的资源
                f.channel().closeFuture().sync();
            }finally{ 
    
            }
        }
        public static void main(String[] args) throws Exception { 
            int port=8080;
            new TimeClient().connect(port, "127.0.0.1");
        }
    }

## TimeClientHandler.java ##

import java.util.logging.Logger;
    
    import io.netty.buffer.ByteBuf;
    import io.netty.buffer.Unpooled;
    import io.netty.channel.ChannelHandlerAdapter;
    import io.netty.channel.ChannelHandlerContext;
    
    public class TimeClientHandler extends ChannelHandlerAdapter{ 
        private static final Logger logger=Logger.getLogger(TimeClientHandler.class.getName());
    
        private final ByteBuf firstMessage;
    
        public TimeClientHandler(){ 
            byte[] req="QUERY TIME ORDER".getBytes();
            firstMessage=Unpooled.buffer(req.length);
            firstMessage.writeBytes(req);
        }
        /** * 当客户端和服务器TCP链路建立成功后，NIO线程会调用channelActive方法 */
        @Override
        public void channelActive(ChannelHandlerContext ctx){ 
            //发送查询时间的指令给服务端
            ctx.writeAndFlush(firstMessage);
        }
        /** * 当服务端返回应答消息时调用 */
        @Override
        public void channelRead(ChannelHandlerContext ctx,Object msg)throws Exception{ 
            ByteBuf buf=(ByteBuf)msg;
            byte[] req=new byte[buf.readableBytes()];
            buf.readBytes(req);
            String body=new String(req,"UTF-8");
            System.out.println("Now is : " + body);
        }
        /** * 当发生异常时 */
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx,Throwable cause){ 
            logger.warning("Unexpected exception from downstrea : " + cause.getMessage());
            ctx.close();
        }
    }

上述例程仍没有考虑读半包的处理，但进行性能或者压力测试将不能正确工作。

# 3. 总结 #

*  当channel上面有数据到来时会触发channelRead事件，当数据到来时，eventLoop被唤醒继而调用channelRead方法处理数据。
 *  当Channel上一旦没有更多数据要从底层传输中读取，就会触发channelReadComplete（）。可能是以下两种情况，read到0个字节或者是read到的字节数小于buffer的容量，满足以上条件就会调用channelReadComplete方法。