io.netty学习（一）Netty入门

前言

Java原生API之痛

Netty的优势

非阻塞 I/O

丰富的协议

异步和事件驱动

精心设计的API

丰富的缓冲实现

高效的网络传输

Netty 核心概念

核心组件

传输服务

协议支持

Netty简单应用

总结

前言

关于Netty的学习，最近看了不少有关视频和书籍，也收获不少，希望把我知道的分享给你们，一起加油，一起成长。前面我们对 Java IO、BIO、NIO、 AIO进行了分析，相关文章链接如下：

io.netty学习使用汇总

本篇文章我们就开始对 Netty来进行深入分析，首先我们来了解一下 JAVA NIO 、AIO的不足之处。

Java原生API之痛

虽然JAVA NIO 和 JAVA AIO框架提供了多路复用IO/异步IO的支持，但是并没有提供上层“信息格式”的良好封装。用这些API实现一款真正的网络应用则并非易事。

JAVA NIO 和 JAVA AIO并没有提供断连重连、网络闪断、半包读写、失败缓存、网络拥塞和异常码流等的处理，这些都需要开发者自己来补齐相关的工作。

AIO在实践中，并没有比NIO更好。AIO在不同的平台有不同的实现，windows系统下使用的是一种异步IO技术：IOCP；Linux下由于没有这种异步 IO 技术，所以使用的是epoll 对异步 IO 进行模拟。所以 AIO 在 Linux 下的性能并不理想。AIO 也没有提供对 UDP 的支持。

综上，在实际的大型互联网项目中，Java 原生的 API 应用并不广泛，取而代之的是一款第三方Java 框架，这就是Netty。

Netty的优势

Netty 提供 异步的、事件驱动的网络应用程序框架和工具，用以快速开发高性能、高可靠性的网络服务器和客户端程序。

非阻塞 I/O

Netty 是基于 Java NIO API 实现的网络应用框架，使用它可以快速简单的开发网络应用程序，如服务器和客户端程序。Netty 大大简化了网络程序开发的过程，如 TCP 和 UDP 的 Socket 服务的开发。

由于是基于 NIO 的 API，因此，Netty 可以提供非阻塞的 I/O操作，极大的提升了性能。同时，Netty 内部封装了 Java NIO API 的复杂性，并提供了线程池的处理，使得开发 NIO 的应用变得极其简单。

丰富的协议

Netty 提供了简单、易用的 API ,但这并不意味着应用程序会有难维护和性能低的问题。Netty 是一个精心设计的框架，它从许多协议的实现中吸收了很多的经验，如 FTP 、SMTP、 HTTP、许多二进制和基于文本的传统协议。

Netty 支持丰富的网络协议，如TCP、 UDP、 HTTP、 HTTP/2、 WebSocket、 SSL/TLS等，这些协议实现开箱即用，因此，Netty 开发者能够在不失灵活的前提下来实现开发的简易性、高性能和稳定性。

异步和事件驱动

Netty 是异步事件驱动的框架，该框架体现为所有的I/O操作都是异步的，所有的I/O调用会立即返回，并不保证调用成功与否，但是调用会返回ChannelFuture。Netty 会通过 ChannelFuture通知调用是成功了还是失败了，抑或是取消了。

同时，Netty 是基于事件驱动的，调用者并不能立即获得结果，而是通过事件监听机制，用户可以方便地主动获取或者通过通知机制获得I/O操作的结果。

当Future对象刚刚创建时，处于非完成状态，调用者可以通过返回的ChannelFuture来获取操作执行的状态，再通过注册监听函数来执行完成后的操作，常见有如下操作：

通过isDone方法来判断当前操作是否完成。
通过isSuccess方法来判断已完成的当前操作是否成功。
通过getCause方法来获取已完成的当前操作失败的原因。
通过isCancelled方法来判断已完成的当前操作是否被取消。
通过addListener方法来注册监听器，当操作已完成(isDone方法返回完成)，将会通知指定的监听器；如果future对象已完成，则理解通知指定的监听器。

例如：下面的代码中绑定端口是异步操作，当绑定操作处理完，将会调用相应的监听器处理逻辑。

serverBootstrap.bind(port).addListener(future -> {
    if(future.isSuccess()){
        System.out.println("端口绑定成功！");
    }else {
        System.out.println("端口绑定失败！");
    }
});

相比传统的阻塞 I/O，Netty 异步处理的好处是不会造成线程阻塞，线程在 I/O操作期间可以执行其他的程序，在高并发情形下会更稳定并拥有更高的吞吐量。

精心设计的API

Netty 从开始就为用户提供了体验最好的API及实现设计。

例如，在用户数较小的时候可能会选择传统的阻塞API，毕竟与 Java NIO 相比使用阻塞 API 将会更加容易一些。然而，当业务量呈指数增长并且服务器需要同时处理成千上万的客户连接，便会遇到问题。这种情况下可能会尝试使用 Java NIO，但是复杂的 NIO Selector 编程接口又会耗费大量的时间并最终会阻碍快速开发。

Netty 提供了一个叫作 channel的统一的异步I/O编程接口，这个编程接口抽象了所有点对点的通信操作。也就是说，如果应用是基于Netty 的某一种传输实现，那么同样的，应用也可以运行在 Netty 的另一种传输实现上。Channel常见的子接口有：

丰富的缓冲实现

Netty 使用自建的缓存 API，而不是使用 Java NIO 的 ByteBuffer 来表示一个连续的字节序列。与 ByteBuffer 相比，这种方式拥有明显的优势。

Netty 使用新的缓冲类型 ByteBuf ,并且被设计为可从底层解决 ByteBuffer 问题，同时还满足日常网络应用开发需要的缓冲类型。

Netty 重要有以下特性：

允许使用自定义的缓冲类型。
复合缓冲类型中内置透明的零拷贝实现。
开箱即用动态缓冲类型，具有像 StringBuffer 一样的动态缓冲能力。
不再需要调用flip()方法。
正常情况下具有比ByteBuffer更快的响应速度。

高效的网络传输

Java 原生的序列化主要存在以下几个弊端：

无法跨语言。
序列化后码流太大。
序列化后性能太低。

业界有非常多的框架用于解决上述问题，如 Google Protobuf 、JBoss Marshalling、Facebook Thrift等。针对这些框架，Netty 都提供了相应的包将这些框架集成到应用中。同时，Netty 本身也提供了众多的编解码工具，方便开发者使用。开发者可以基于 Netty 来开发高效的网络传输应用，例如：高性能的消息中间件 Apache RocketMQ、高性能RPC框架Apache Dubbo等。

Netty 核心概念

从上述的架构图可以看出，Netty 主要由三大块组成：

核心组件
传输服务
协议

核心组件

核心组件包括：事件模型、字节缓冲区和通信API

事件模型

Netty 是基于异步事件驱动的，该框架体现为所有的I/O操作都是异步的，调用者并不能立即获得结果，而是通过事件监听机制，用户可以方便地主动获取或者通过通知机制获得I/O操作的结果。

Netty 将所有的事件按照它们与入站或出站数据流的相关性进行了分类。

可能由入站数据或者相关的状态更改而触发的事件包括以下几项：

连接已被激活或者连接失活。
数据读取。
用户事件。
错误事件。

出站事件是未来将会触发的某个动作的操作结果，包括以下动作：

打开或者关闭到远程节点的连接。
将数据写到或者冲刷到套接字。

每个事件都可以被分发到ChannelHandler类中的某个用户实现的方法。

字节缓冲区

Netty 使用了区别于Java ByteBuffer 的新的缓冲类型ByteBuf,ByteBuf提供了丰富的特性。

通信API

Netty 的通信API都被抽象到Channel里，以统一的异步I/O编程接口来满足所有点对点的通信操作。

传输服务

Netty 内置了一些开箱即用的传输服务。因为并不是它们所有的传输都支持每一种协议，所以必须选择一个和应用程序所使用的协议相兼容的传输。以下是Netty提供的所有的传输。

NIO

io.netty.channel.socket.nio包用于支持NIO。该包下面的实现是使用java.nio.channels包作为基础（基于选择器的方式）。

epoll

io.netty.channel.epoll包用于支持由 JNI 驱动的 epoll 和非阻塞 IO。

需要注意的是，这个epoll传输只能在 Linux 上获得支持。epoll同时提供多种特性，如：SO_REUSEPORT 等，比 NIO传输更快，而且是完全非阻塞的。

OIO

io.netty.channel.socket.oio包用于支持使用java.net包作为基础的阻塞I/O。

本地

io.netty.channel.local包用于支持在 VM 内部通过管道进行通信的本地传输。

内嵌

io.netty.channel.embedded包作为内嵌传输，允许使用ChannelHandler而又不需要一个真正的基于网络的传输。

协议支持

Netty简单应用

引入Maven依赖

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

服务端的管道处理器

public class NettyServerHandler extends ChannelInboundHandlerAdapter {

    //读取数据实际(这里我们可以读取客户端发送的消息)
    /*
    1. ChannelHandlerContext ctx:上下文对象, 含有 管道pipeline , 通道channel, 地址
    2. Object msg: 就是客户端发送的数据 默认Object
     */
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        System.out.println("server ctx =" + ctx);
        Channel channel = ctx.channel();
        //将 msg 转成一个 ByteBuf
        //ByteBuf 是 Netty 提供的，不是 NIO 的 ByteBuffer.
        ByteBuf buf = (ByteBuf) msg;
        System.out.println("客户端发送消息是:" + buf.toString(CharsetUtil.UTF_8));
        System.out.println("客户端地址:" + channel.remoteAddress());
    }


    //数据读取完毕
    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        //writeAndFlush 是 write + flush
        //将数据写入到缓存，并刷新
        //一般讲，我们对这个发送的数据进行编码
        ctx.writeAndFlush(Unpooled.copiedBuffer("公司最近账户没啥钱，再等几天吧！", CharsetUtil.UTF_8));
    }

    //处理异常, 一般是需要关闭通道
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        ctx.close();
    }
}

NettyServerHandler继承自ChannelInboundHandlerAdapter，这个类实现了ChannelInboundHandler接口。ChannelInboundHandler提供了许多事件处理的接口方法。

这里覆盖了channelRead()事件处理方法。每当从客户端收到新的数据时，这个方法会在收到消息时被调用。

channelReadComplete()事件处理方法是数据读取完毕时被调用，通过调用ChannelHandlerContext的writeAndFlush()方法，把消息写入管道，并最终发送给客户端。

exceptionCaught()事件处理方法是，当出现Throwable对象时才会被调用。

服务端主程序

public class NettyServer {

    public static void main(String[] args) throws Exception {
        //创建BossGroup 和 WorkerGroup
        //说明
        //1. 创建两个线程组 bossGroup 和 workerGroup
        //2. bossGroup 只是处理连接请求 , 真正的和客户端业务处理，会交给 workerGroup完成
        //3. 两个都是无限循环
        //4. bossGroup 和 workerGroup 含有的子线程(NioEventLoop)的个数
        //   默认实际 cpu核数 * 2
        //
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup(); //8
        try {
            //创建服务器端的启动对象，配置参数
            ServerBootstrap bootstrap = new ServerBootstrap();
            //使用链式编程来进行设置
            bootstrap.group(bossGroup, workerGroup) //设置两个线程组
                    .channel(NioServerSocketChannel.class) //bossGroup使用NioSocketChannel 作为服务器的通道实现
                    .option(ChannelOption.SO_BACKLOG, 128) // 设置线程队列得到连接个数 option主要是针对boss线程组，
                    .childOption(ChannelOption.SO_KEEPALIVE, true) //设置保持活动连接状态 child主要是针对worker线程组
                    .childHandler(new ChannelInitializer<SocketChannel>() {//workerGroup使用 SocketChannel创建一个通道初始化对象                                                                                                                        (匿名对象)
                        //给pipeline 设置处理器
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            //可以使用一个集合管理 SocketChannel， 再推送消息时，可以将业务加入到各个channel 对应的 NIOEventLoop 的                                     taskQueue 或者 scheduleTaskQueue
                            ch.pipeline().addLast(new NettyServerHandler());
                        }
                    }); // 给我们的workerGroup 的 EventLoop 对应的管道设置处理器

            System.out.println(".....服务器 is ready...");
            //绑定一个端口并且同步, 生成了一个 ChannelFuture 对象
            //启动服务器(并绑定端口)
            ChannelFuture cf = bootstrap.bind(7788).sync();
            //给cf 注册监听器，监控我们关心的事件
            cf.addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(ChannelFuture future) throws Exception {
                    if (cf.isSuccess()) {
                        System.out.println("服务已启动,端口号为7788...");
                    } else {
                        System.out.println("服务启动失败...");
                    }
                }
            });
            //对关闭通道进行监听
            cf.channel().closeFuture().sync();
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}

NioEventLoopGroup是用来处理I/O操作的多线程事件循环器。Netty 提供了许多不同的EventLoopGroup的实现来处理不同的传输。

上面的服务端应用中，有两个NioEventLoopGroup被使用。第一个叫作bossGroup，用来接收进来的连接。第二个叫作workerGroup，用来处理已经被接收的连接，一旦 bossGroup接收连接，就会把连接的信息注册到workerGroup上。

ServerBootstrap是一个NIO服务的引导启动类。可以在这个服务中直接使用Channel。

group方法用于设置EventLoopGroup。
通过Channel方法，可以指定新连接进来的Channel类型为NioServerSocketChannel类。
childHandler用于指定ChannelHandler，也就是前面实现的NettyServerHandler。
可以通过option设置指定的Channel来实现NioServerSocketChannel的配置参数。
childOption主要设置SocketChannel的子Channel的选项。
bind用于绑定端口启动服务。

客户端管道处理器

public class NettyClientHandler extends ChannelInboundHandlerAdapter {

    //当通道就绪就会触发该方法
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("client ctx =" + ctx);
        ctx.writeAndFlush(Unpooled.copiedBuffer("老板，工资什么时候发给我啊？", CharsetUtil.UTF_8));
    }

    //当通道有读取事件时，会触发
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        ByteBuf buf = (ByteBuf) msg;
        System.out.println("服务器回复的消息:" + buf.toString(CharsetUtil.UTF_8));
        System.out.println("服务器的地址： "+ ctx.channel().remoteAddress());
    }

    //处理异常, 一般是需要关闭通道
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.close();
    }
}

channelRead方法中将接收到的消息转化为字符串，方便在控制台上打印出来。

channelRead接收到的消息类型为ByteBuf，ByteBuf提供了转为字符串的方便方法。

客户端主程序

public class NettyClient {

    public static void main(String[] args) throws Exception {
        //客户端需要一个事件循环组
        EventLoopGroup group = new NioEventLoopGroup();
        try {
            //创建客户端启动对象
            //注意客户端使用的不是 ServerBootstrap 而是 Bootstrap
            Bootstrap bootstrap = new Bootstrap();
            //设置相关参数
            bootstrap.group(group) //设置线程组
                    .channel(NioSocketChannel.class) // 设置客户端通道的实现类(反射)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ch.pipeline().addLast(new NettyClientHandler()); //加入自己的处理器
                        }
                    });
            System.out.println("客户端 ok..");
            //启动客户端去连接服务器端
            //关于 ChannelFuture 要分析，涉及到netty的异步模型
            ChannelFuture channelFuture = bootstrap.connect("127.0.0.1", 7788).sync();
            //给关闭通道进行监听
            channelFuture.channel().closeFuture().sync();
        } finally {
            group.shutdownGracefully();
        }
    }
}

客户端只需要一个NioEventLoopGroup就可以了。

测试运行

分别启动服务器 NettyServer 和客户端 NettyClient程序

服务端控制台输出内容：

.....服务器 is ready...
服务已启动,端口号为7788...
server ctx =ChannelHandlerContext(NettyServerHandler#0, [id: 0xa1b2233c, L:/127.0.0.1:7788 - R:/127.0.0.1:63239])
客户端发送消息是:老板，工资什么时候发给我啊？
客户端地址:/127.0.0.1:63239

客户端控制台输出内容：

客户端 ok..
client ctx =ChannelHandlerContext(NettyClientHandler#0, [id: 0x21d6f98e, L:/127.0.0.1:63239 - R:/127.0.0.1:7788])
服务器回复的消息:公司最近账户没啥钱，再等几天吧！
服务器的地址： /127.0.0.1:7788

至此，一个简单的基于Netty开发的服务端和客户端就完成了。