使用Netty框架的原因和入门（八）

超、凢脫俗 2022-10-29 12:29 144阅读 0赞

今天我们分析使用Netty框架的原因和入门

**一、为什么要用**  **Netty**

1 、虽然  JAVA NIO  框架提供了 多路复用  IO  的支持，但是并没有提供上层“信息格式” 的良好封装。例如前两者并没有提供针对 Protocol Buffer 、 JSON  这些信息格式的封装，但是Netty 框架提供了这些数据格式封装（基于责任链模式的编码和解码功能）；

2 、 NIO  的类库和  API  相当复杂，使用它来开发，需要非常熟练地掌握  Selector 、ByteBuffer、 ServerSocketChannel、 SocketChannel  等，需要很多额外的编程技能来辅助使用  NIO, 例如，因 为 NIO  涉及了  Reactor  线程模型，所以必须必须对多线程和网络编程非常熟悉才能写出高质量的 NIO  程序

3 、要编写一个可靠的、易维护的、高性能的  NIO  服务器应用。除了框架本身要兼容实 现各类操作系统的实现外。更重要的是它应该还要处理很多上层特有服务，例如：客户端的 权限、还有上面提到的信息格式封装、简单的数据读取，断连重连，半包读写，心跳等等，这些 Netty  框架都提供了响应的支持。

4 、 JAVA NIO  框架存在一个  poll/epoll bug ： Selector doesn ’ t block on Selector.select(timeout)，不能  block  意味着  CPU  的使用率会变成  100% （这是底层  JNI  的问题，

上层要处理这个异常实际上也好办）。当然这个  bug  只有在  Linux  内核上才能重现。 这个问题在 JDK 1.7  版本中还没有被完全解决，但是  Netty  已经将这个  bug  进行了处理。 这个 Bug  与操作系统机制有关系的，JDK 虽然仅仅是一个兼容各个操作系统平台的软件， 但在 JDK5  和  JDK6  最初的版本中（严格意义上来将， JDK  部分版本都是），这个问题并没有 解决，而将这个帽子抛给了操作系统方，这也就是这个 bug  最终一直到  2013  年才最终修复 的原因(JDK7  和  JDK8  之间 ) 。 Netty的性能很高，按照  Facebook  公司开发小组的测试表明， Netty  最高能达到接近百万的吞吐：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70][]

**二、为什么不用**  **Netty5**

Netty5  已经停止开发了。

**三、为什么**  **Netty**  **使用**  **NIO**  **而不是**  **AIO** **？**

Netty  不看重  Windows  上的使用，在  Linux  系统上， AIO  的底层实现仍使用  EPOLL ，没有 很好实现 AIO ，因此在性能上没有明显的优势，而且被  JDK  封装了一层不容易深度优化。 AIO 还有个缺点是接收数据需要预先分配缓存 ,  而不是  NIO  那种需要接收时才需要分配 缓存,  所以对连接数量非常大但流量小的情况 ,  内存浪费很多。 据说 Linux  上  AIO  不够成熟，处理回调结果速度跟不上处理需求，有点像外卖员太少， 顾客太多，供不应求，造成处理速度有瓶颈。

作者原话：

Not faster than NIO (epoll) on unix systems (which is true)

There is no daragram suppport

Unnecessary threading model (too much abstraction without usage)

**四、第一个**  **Netty**  **程序**

**1、EventLoop(Group)**  **、** **Channel**

Channel  是  Java NIO  的一个基本构造。 它代表一个到实体（如一个硬件设备、一个文件、一个网络套接字或者一个能够执行一 个或者多个不同的 I/O  操作的程序组件）的开放连接，如读操作和写操作 目前，可以把 Channel  看作是传入（入站）或者传出（出站）数据的载体。因此，它 可以被打开或者被关闭，连接或者断开连接。 EventLoop 暂时可以看成一个线程、 EventLoopGroup  自然就可以看成线程组。

**2、事件和**  **ChannelHandler** **、** **ChannelPipeline**

Netty  使用不同的事件来通知我们状态的改变或者是操作的状态。这使得我们能够基于 已经发生的事件来触发适当的动作。 Netty 事件是按照它们与入站或出站数据流的相关性进行分类的。 可能由入站数据或者相关的状态更改而触发的事件包括： 连接已被激活或者连接失活；数据读取；用户事件；错误事件。 出站事件是未来将会触发的某个动作的操作结果，这些动作包括： 打开或者关闭到远程节点的连接；将数据写到或者冲刷到套接字。 每个事件都可以被分发给 ChannelHandler  类中的某个用户实现的方法。

Netty  提供了大量预定义的可以开箱即用的  ChannelHandler  实现，包括用于各种协议 （如 HTTP  和  SSL/TLS ）的  ChannelHandler 。

**3、ChannelFuture**

Netty  中所有的  I/O  操作都是异步的。 JDK 预置了  interface java.util.concurrent.Future ， Future  提供了一种在操作完成时通知 应用程序的方式。这个对象可以看作是一个异步操作的结果的占位符；它将在未来的某个时刻完成，并提供对其结果的访问。但是其所提供的实现，只允许手动检查对应的操作是否已经完成，或者一直阻塞直到它完成。这是非常繁琐的，所以 Netty  提供了它自己的实现—— ChannelFuture，用于在执行异步操作的时候使用。

**每个**  **Netty**  **的出站**  **I/O**  **操作都将返回一个**  **ChannelFuture** **。**

**4、引用Netty相关jar包**

<dependency>
                <groupId>io.netty</groupId>
                <artifactId>netty-all</artifactId>
                <version>4.1.28.Final</version>
            </dependency>

**5、Netty 服务端代码：**

public class EchoServer  {
    
        private final int port;
    
        public EchoServer(int port) {
            this.port = port;
        }
    
        public static void main(String[] args) throws InterruptedException {
            int port = 9999;
            EchoServer echoServer = new EchoServer(port);
            System.out.println("服务器即将启动");
            echoServer.start();
            System.out.println("服务器关闭");
        }
    
        public void start() throws InterruptedException {
            final EchoServerHandler serverHandler = new EchoServerHandler();
            /**线程组*/
            EventLoopGroup group  = new NioEventLoopGroup();
            try {
                /**服务端启动必备*/
                ServerBootstrap b = new ServerBootstrap();
                b.group(group)
                        .channel(NioServerSocketChannel.class)/**指定使用NIO的通信模式*/
                .localAddress(new InetSocketAddress(port))/**指定监听端口*/
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
                        ch.pipeline().addLast(serverHandler);
                    }
                });
                ChannelFuture f = b.bind().sync();/**异步绑定到服务器，sync()会阻塞到完成*/
                f.channel().closeFuture().sync();/**阻塞当前线程，直到服务器的ServerChannel被关闭*/
            } finally {
                group.shutdownGracefully().sync();
            }
        }
    }

服务端事件类处理：继承 ChannelInboundHandlerAdapter

@ChannelHandler.Sharable
    public class EchoServerHandler extends ChannelInboundHandlerAdapter {
    
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            ByteBuf in = (ByteBuf)msg;
            System.out.println("Server accept: "+in.toString(CharsetUtil.UTF_8));
    //
            ctx.writeAndFlush(in);
            //ctx.close();
        }
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            cause.printStackTrace();
            ctx.close();
        }
    }

**6、Netty 客户端代码：**

ublic class EchoClient {
    
        private final int port;
        private final String host;
    
        public EchoClient(int port, String host) {
            this.port = port;
            this.host = host;
        }
        public void start() throws InterruptedException {
            /**线程组*/
            EventLoopGroup group  = new NioEventLoopGroup();
            try {
                /**客户端启动必备*/
                Bootstrap b = new Bootstrap();
                b.group(group)
                        .channel(NioSocketChannel.class)/**指定使用NIO的通信模式*/
    
                        .remoteAddress(new InetSocketAddress(host,port))/**指定服务器的IP地址和端口*/
                        .handler(new ChannelInitializer<SocketChannel>() {
                            @Override
                            protected void initChannel(SocketChannel ch) throws Exception {
                                ch.pipeline().addLast(new EchoClientHandler());
    
                            }
                        });
                ChannelFuture f = b.connect().sync();/**异步连接到服务器，sync()会阻塞到完成*/
                f.channel().closeFuture().sync();/**阻塞当前线程，直到客户端的Channel被关闭*/
            } finally {
                group.shutdownGracefully().sync();
            }
        }
    
        public static void main(String[] args) throws InterruptedException {
            new EchoClient(9999,"127.0.0.1").start();
        }
    }

客户端事件处理：继承 SimpleChannelInboundHandler<ByteBuf>

public class EchoClientHandler extends SimpleChannelInboundHandler<ByteBuf> {
    
        /**读取到网络数据后进行业务处理*/
        @Override
        protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception {
            System.out.println("client Accept"+msg.toString(CharsetUtil.UTF_8));
            //ctx.close();
            ctx.close();
        }
    
        /**channel活跃后，做业务处理*/
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            ctx.writeAndFlush(Unpooled.copiedBuffer(
                    "Hello,Netty",CharsetUtil.UTF_8));
        }
    }

**7、Netty 程序执行结果：客户端执行结果**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 1][]

服务端执行结果：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 2][]

到此，Netty 的入门使用分析完成，下篇我们分析Netty 各组件的原理，敬请期待！

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70]: /images/20221024/67511f3fc300471bb565b50bc9933bc3.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 1]: /images/20221024/7e42aa2f465f4e1b8e82155312359d08.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 2]: /images/20221024/a8061387ec2e47d4aebcbfd67fce205c.png