基于netty4.x开发时间服务器

在写代码之前 先了解下Reactor模型:

Reactor单线程模型就是指所有的IO操作都在同一个NIO线程上面完成的，也就是IO处理线程是单线程的。NIO线程的职责是： 

（1）作为NIO服务端，接收客户端的TCP连接；

（2）作为NIO客户端，向服务端发起TCP连接；

（3）读取通信对端的请求或者应答消息；

（4）向通信对端发送消息请求或者应答消息。

模型图如下:

Reactor模式使用的是同步非阻塞IO(NIO)，所有的IO操作都不会导致阻塞，理论上一个线程可以独立的处理所有的IO操作(selector会主动去轮询哪些IO操作就绪)。从架构层次看，一个NIO线程确实可以完成其承担的职责，比如上图的Acceptor类接收客户端的TCP请求消息，当链路建立成功之后，通过Dispatch将对应的ByteBuffer转发到指定的handler上，进行消息的处理。

对于一些小容量的应用场景下，可以使用单线程模型，但是对于高负载、大并发的应用场景却不适合，主要原因如下： 
（1）一个NIO线程处理成千上万的链路，性能无法支撑，即使CPU的负荷达到100%；

（2）当NIO线程负载过重，处理性能就会变慢，导致大量客户端连接超时然后重发请求，导致更多堆积未处理的请求，成为性能瓶颈。

（3）可靠性低，只有一个NIO线程，万一线程假死或则进入死循环，就完全不可用了，这是不能接受的。

Reactor多线程模型与单线程模型最大的区别在于，IO处理线程不再是一个线程，而是一组NIO处理线程。

模型图:

Reactor多线程模型的特点如下： 
（1）有一个专门的NIO线程—-Acceptor线程用于监听服务端，接收客户端的TCP连接请求。

（2）网络IO操作—-读写等操作由一个专门的线程池负责，线程池可以使用JDK标准的线程池实现，包含一个任务队列和N个可用的线程，这些NIO线程就负责读取、解码、编码、发送。

（3）一个NIO线程可以同时处理N个链路，但是一个链路只对应一个NIO线程。

 (4)虽然采用了多线程，虽然仍旧是单个selector线程，但是请求的处理大头交给了线程池异步执行。

Reactor多线程模型可以满足绝大多数的场景，除了一些个别的特殊场景：比如一个NIO线程负责处理客户所有的连接请求，但是如果连接请求中包含认证的需求（安全认证），在百万级别的场景下，就存在性能问题了，因为认证本身就要消耗CPU，为了解决这种情景下的性能问题，产生了第三种线程模型：Reactor主从线程模型。

主从Reactor线程模型的特点是：服务端用于接收客户端连接的不再是一个单独的NIO线程，而是一个独立的NIO的线程池。Acceptor接收到客户端TCP连接请求并处理完成后（可能包含接入认证），再将新创建的SocketChannel注册到IO线程池（sub reactor）的某个IO处理线程上并处理编解码和读写工作。Acceptor线程池仅负责客户端的连接与认证，一旦链路连接成功，就将链路注册到后端的sub Reactor的IO线程池中。 利用主从Reactor模型可以解决服务端监听线程无法有效处理所有客户连接的性能不足问题，这也是netty推荐使用的线程模型。

模型图:

TimeServer类(服务端)

public class TimeServer {

    public void bind(int port)throws Exception{
        /**NioEventLoopGroup 是线程组  它包含了一组NIO线程 专门处理网络事件 实际上它们就是Reactor线程组
         *Reactor线程是多面手，负责多路分离套接字，Accept新连接，并分派请求到处理器链中。
         */

        /**
         *用于服务端接受客户端的连接
         */
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        /**
         * 用于进行SocketChannel的网络读写
         */
        EventLoopGroup workerGroup = new NioEventLoopGroup();

        try{
            /**
             * 用于启动NIO服务端的辅助启动类 降低服务端的开发难度
             */
            ServerBootstrap b = new ServerBootstrap();
            /**
             *1：将两个NIO线程组当作参数传到ServerBootstrap中
             * 2:设置创建的Channel 为NioServerSocketChannel 它对应于JDK NIO类库中的ServerSocketChannel
             * 3：配置NioServerSocketChannel的Tcp参数   BACKLOG用于构造服务端套接字ServerSocket对象，标识当服务器请求处理线程全满时，用于临时存放已完成三次握手的请求的队列的最大长度。
             * 4：绑定I/O事件的处理类ChildChannelHandler  它的作用模式类似于Reactor模式中的handler类 主要处理网络I/O事件 列如记录日志 对消息进行编解码等
             */

            b.group(bossGroup,workerGroup).channel(NioServerSocketChannel.class).option(ChannelOption.SO_BACKLOG,1024).childHandler(new
                    ChildChannelHandler());
            /**
             * 服务端启动辅助类配置完成后 调用它的bind方法绑定监听端口 随后调用它的同步阻塞方法sync等待绑定操作完成 完成后Netty会返回一个ChannelFuture 类似于jdk中
             * 的java.util.concurrent.Future包  主要用于异步操作的通知回调
             */
            ChannelFuture f = b.bind(port).sync();
            /**
             * 进行阻塞 等待服务端链路关闭之后 main函数才退出
             */
            f.channel().closeFuture().sync();
        }finally {
            /**
             * 优雅退出 相关资源
             */
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }

    }

    private class ChildChannelHandler extends ChannelInitializer<SocketChannel>{
        @Override
        protected void initChannel(SocketChannel ch){
            ch.pipeline().addLast(new TimeServerHandler());
        }
    }

    public static void main(String[] args) throws Exception {
        int port  = 8080;
        if(args!=null&&args.length>0){
            try {
                port = Integer.valueOf(args[0]);
            }catch (NumberFormatException e){
                e.printStackTrace();
            }
        }
        new TimeServer().bind(port);
    }
    
}

TimeServerHandler类(服务端处理类)

public class TimeServerHandler extends ChannelInboundHandlerAdapter {
    /**
     * TimeServerHandler继承ChannelInboundHandlerAdapter 用于对网络事件进行读写操作 通常我们只关注channelRead方法和exceptionCaught方法
     * @param ctx
     * @param msg
     * @throws Exception
     */
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        /**
         * 做类型转换 将msg转化为netty的ByteBuf对象
         */
        ByteBuf buf = (ByteBuf) msg;
        /**
         * 获取缓冲区可读的字节数 根据可读字节数创建byte数组
         */
        byte[] req = new byte[buf.readableBytes()];
        /**
         * 将缓冲区的字节数复制到新建的byte数组
         */
        buf.readBytes(req);
        /**
         *对请求消息进行判断 如果是"QUERY TIME ORDER" 则创建应答消息  通过ChannelHandlerContext的write方法异步发送应答消息给客户端
         */
        String body = new String(req,"UTF-8");
        System.out.println("The time server receive order: "+body);
        String currentTime = "QUERY TIME ORDER".equalsIgnoreCase(body)?new Date(
                System.currentTimeMillis()).toString():"BAD ORDER";
                ByteBuf resp = Unpooled.copiedBuffer(currentTime.getBytes());
                ctx.write(resp);
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        /**
         * 将消息发送队列中的消息写入到SocketChannel中发送给对方  通过调用write方法将消息发送到缓冲数组中 再调用flush方法将
         * 缓冲区的消息全部写入SocketChannel
         */
        ctx.flush();

    }

    /**
     * 发生异常时关闭资源 打印异常信息
     * @param ctx
     * @param cause
     * @throws Exception
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.close();
    }
}

TimeClient类(客户端)

public class TimeClient {

    public void connect(int port,String host){
        /**
         * 创建客户端处理的I/O读写的NioEventLoopGroup线程组
         */
        EventLoopGroup group = new NioEventLoopGroup();
        try{
            /**
             * 创建客户端辅助启动类
             */
            Bootstrap b = new Bootstrap();
            /**
             * 配置
             */
            b.group(group).channel(NioSocketChannel.class).option(ChannelOption.TCP_NODELAY,true).handler(
                    /**
                     * 再初始化化的时候将它的ChannelHandler设置到ChannelPipeline中 用于处理网络I/O事件
                     */
                    new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch){
                            ch.pipeline().addLast(new TimeClientHandler());
                        }
                    }
            );
            /**
             * 发起异步连接操作
             */
            ChannelFuture f = b.connect(host,port);
            /**
             * 等待客户端链路关闭
             */
            f.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }finally {
            group.shutdownGracefully();
        }

    }
    public static void main(String[] args){

        int port = 8080;
        if(args!=null&&args.length>0){
            try {
                port=Integer.valueOf(args[0]);
            }catch (NumberFormatException e){
                e.printStackTrace();
            }
        }
        new TimeClient().connect(port,"127.0.0.1");
    }

}

TimeClientHandler(客户端处理类)

public class TimeClientHandler extends ChannelInboundHandlerAdapter {
    private final ByteBuf fristMessage;

    public TimeClientHandler() {
        byte[] req = "QUERY TIME ORDER".getBytes();
        fristMessage = Unpooled.buffer(req.length);
        fristMessage.writeBytes(req);
    }

    /**
     * 当客户端和服务端的TCP链路建立成功后 Netty的NIO线程会调用channelActive方法 发送查询事件的指令给服务端 调用writeAndFlush方法
     * 将请求发送给服务端
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {

        ctx.writeAndFlush(fristMessage);
    }

    /**
     * 党服务端返回应答消息 channelRead 方法被调用
     * @param ctx
     * @param msg
     * @throws Exception
     */
    @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);

    }

    /**
     * 发生异常  释放客户端资源
     * @param ctx
     * @param cause
     * @throws Exception
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.close();
    }
}

运行结果:

github项目地址:https://github.com/INGUCoder/learning/tree/master/Netty%E5%AD%A6%E4%B9%A0

来源：华为云社区  作者：INGUCoder