文章/答案/技术大牛

发布

社区首页 >专栏 >【Netty】NIO编程的利器

【Netty】NIO编程的利器

周三不加班

发布于 2019-09-04 02:05:47

41400

代码可运行

文章被收录于专栏：程序猿杂货铺程序猿杂货铺

运行总次数：0

代码可运行

今天换换口味，由于本人工作中马上要用到Netty这个东西，所以这几天也是开始学习，此学习过程应该会是一个完整的系列，初步的目标是先会用，之后有机会再深入。鉴于笔者之前也从未使用过Netty,所以有什么疏漏错误的，希望大家指正，先行感谢！

学习Netty之前需要先回顾一下java中的IO编程和NIO编程，因为根据我的理解Netty就是对JDK中的NIO编程做了一层很好的封装，可以让开发者专注于业务逻辑的开发。

官方的定义是这样的：

Netty is an asynchronous event-driven network application framework

for rapid development of maintainable high performance protocol servers & clients

大致意思是说：

Netty是一个异步事件驱动的网络应用程序框架

用于快速开发可维护的高性能协议服务器和客户端。

IO编程模型

回顾一个Java Socket编程的简单例子，应用场景为客户度每隔2秒钟向服务端发送一个消息,

服务端代码如下：

import java.io.IOException;
import java.io.InputStream;
import java.net.ServerSocket;
import java.net.Socket;

/**
 * Created by zhoudl on 2018/10/3.
 * IO编程 服务端
 */
public class IOServer {

    public static void main(String[] args) {
        try {
            server(8080);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 1.首先创建了一个serverSocket来监听 某个 端口，然后创建一个线程
     * 2.线程里面不断调用阻塞方法 serversocket.accept();获取新的连接
     * 3.当获取到新的连接之后，给每条连接创建一个新的线程，这个线程负责从该连接中读取数据
     * 4.然后读取数据是以字节流的方式
     *
     * @param port
     * @throws IOException
     */
    public static void server(int port) throws IOException {
        // 1
        ServerSocket serverSocket = new ServerSocket(port);
        new Thread(() -> {
            while (true) {
                try {
                    // 2.调用阻塞方法获取新的连接
                    Socket socket = serverSocket.accept();
                    // 3.每个连接都交给一个新的线程去处理
                    new Thread (() -> {
                        int length;
                        byte[] bytes = new byte[1024];
                        try {
                            InputStream inputStream = socket.getInputStream();
                            // 4.根据字节流方式获取数据
                            while ((length = inputStream.read(bytes)) != -1) {
                                System.out.println(new String(bytes, 0, length));
                            }
                        } catch (IOException e) {
                            e.printStackTrace();
                        }
                    }).start();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }

        }).start();
    }
}

服务端创建过程分析

1.首先创建了一个serverSocket来监听某个端口，然后创建一个线程
2.线程里面不断调用阻塞方法 serversocket.accept()获取新的连接
3.当获取到新的连接之后，给每条连接创建一个新的线程，这个线程负责从该连接中读取数据
4.然后读取数据是以字节流的方式

客户端代码如下：

package top.zhoudl.io;

import java.io.IOException;
import java.net.Socket;
import java.util.Date;

/**
 * Created by zhoudl on 2018/10/3.
 * IO 编程 客户端
 */
public class IOClient {

    public static void client(int port,String address){

        new Thread(() -> {
            try {
                // 建立连接
                Socket socket = new Socket(address,port);
                while (true) {
                    try {
                        socket.getOutputStream()
                            .write((new Date() + ": hello world").getBytes());
                        // 每隔2秒向服务器发送一条 hello world
                        Thread.sleep(2000);
                    } catch (Exception e) {
                    }
                }
            } catch (IOException e) {
                e.printStackTrace();
            }

        }).start();
    }

    public static void main(String[] args) {
        client(8080,"127.0.0.1");
    }
}

O 编程模型在客户端较少的情况下运行良好，但是对于客户端比较多的业务来说，单机服务端可能需要支撑成千上万的连接，IO 模型可能就不太合适了，我们来分析一下原因，根据刚才举例的IO编程模型来说，传统的 IO 模型中，每个连接创建成功之后都需要一个线程来维护，每个线程包含一个 while 死循环，那么 1w 个连接对应 1w 个线程，继而 1w 个 while 死循环，这就带来如下几个问题：

线程资源受限：同一时刻有大量的线程处于阻塞状态是非常严重的资源浪费
线程切换效率低下：单机 CPU 核数固定，线程爆炸之后操作系统频繁进行线程切换，应用性能急剧下降
字节流读取数据：效率低下

NIO编程模型

NIO编程模型可以解决的问题

线程资源受限问题

NIO 编程模型中，新来一个连接不再创建一个新的线程，而是可以把这条连接直接绑定到某个固定的线程，然后这条连接所有的读写都由这个线程来负责。 IO 模型中，一个连接来了，会创建一个线程，对应一个 while 死循环，死循环的目的就是不断监测这条连接上是否有数据可以读，大多数情况下，1w 个连接里面同一时刻只有少量的连接有数据可读，因此，很多个 while 死循环都白白浪费掉了，因为读不出啥数据。而在 NIO 模型中，这么多 while 死循环变成一个死循环，这个死循环由一个线程控制，那么他又是如何做到一个线程，一个 while 死循环就能监测1w个连接是否有数据可读的呢？这就是 NIO 模型中 selector 的作用，一条连接来了之后，现在不创建一个 while 死循环去监听是否有数据可读了，而是直接把这个新的连接注册到 selector 上，然后，通过检查这个 selector，就可以批量监测出有数据可读的连接，进而读取数据。这就是 NIO 模型解决线程资源受限的方案，实际开发过程中，我们会开多个线程，每个线程都管理着一批连接，相对于 IO 模型中一个线程管理一条连接，消耗的线程资源大幅减少。

线程切换效率问题

与此同时，NIO中线程切换效率也会变得高很多，因为线程数量减少，对应所需切换次数也就减少了。

读写效率问题

IO编程使用字节流读取数据，而NIO编程使用的是字节块，NIO模型维护了一个缓冲区，可以按块从这个缓冲区中读取数据。这就好比小时候嗑瓜子（当然我长大之后还是喜欢这样），一般会先嗑好很多之后，然后一次性吃到嘴里，这种快感比一个一个塞嘴里咬爽了不少。对于程序来讲，效率自然高了很多。接下来演示一下使用JDK原生的API实现NIO编程，可能比较辣眼睛，非礼勿视（说实话，下面这段代码我也是拷过来的，因为JDK原生的NIO编程用起来实在是恐怖，恐怖如斯，倒吸凉气，有木有？） package top.zhoudl.NIO; import java.io.IOException; import java.net.InetSocketAddress; import java.nio.ByteBuffer; import java.nio.channels.SelectionKey; import java.nio.channels.Selector; import java.nio.channels.ServerSocketChannel; import java.nio.channels.SocketChannel; import java.nio.charset.Charset; import java.util.Iterator; import java.util.Set; /** * Created by zhoudl on 2018/10/3. */ public class NIOServer { public static void main(String[] args) throws IOException { Selector serverSelector = Selector.open(); Selector clientSelector = Selector.open(); new Thread(() -> { try { // 对应IO编程中服务端启动 ServerSocketChannel listenerChannel = ServerSocketChannel.open(); listenerChannel.socket().bind(new InetSocketAddress(8000)); listenerChannel.configureBlocking(false); listenerChannel.register(serverSelector, SelectionKey.OP_ACCEPT); while (true) { // 监测是否有新的连接，这里的1指的是阻塞的时间为 1ms if (serverSelector.select(1) > 0) { Set<SelectionKey> set = serverSelector.selectedKeys(); Iterator<SelectionKey> keyIterator = set.iterator(); while (keyIterator.hasNext()) { SelectionKey key = keyIterator.next(); if (key.isAcceptable()) { try { // (1) 每来一个新连接，不需要创建一个线程，而是直接注册到clientSelector SocketChannel clientChannel = ((ServerSocketChannel) key.channel()).accept(); clientChannel.configureBlocking(false); clientChannel.register(clientSelector, SelectionKey.OP_READ); } finally { keyIterator.remove(); } } } } } } catch (IOException ignored) { } }).start(); new Thread(() -> { try { while (true) { // (2) 批量轮询是否有哪些连接有数据可读，这里的1指的是阻塞的时间为 1ms if (clientSelector.select(1) > 0) { Set<SelectionKey> set = clientSelector.selectedKeys(); Iterator<SelectionKey> keyIterator = set.iterator(); while (keyIterator.hasNext()) { SelectionKey key = keyIterator.next(); if (key.isReadable()) { try { SocketChannel clientChannel = (SocketChannel) key.channel(); ByteBuffer byteBuffer = ByteBuffer.allocate(1024); // (3) 读取数据以块为单位批量读取 clientChannel.read(byteBuffer); byteBuffer.flip(); System.out.println(Charset.defaultCharset().newDecoder().decode(byteBuffer) .toString()); } finally { keyIterator.remove(); key.interestOps(SelectionKey.OP_READ); } } } } } } catch (IOException ignored) { } }).start(); } } 简单分析下过程：

NIO 模型中通常会有两个线程，每个线程绑定一个轮询器 selector ，在我们这个例子中serverSelector负责轮询是否有新的连接，clientSelector负责轮询连接是否有数据可读
服务端监测到新的连接之后，不再创建一个新的线程，而是直接将新连接绑定到clientSelector上，这样就不用 IO 模型中 1w 个 while 循环在死等，参见(1)
clientSelector被一个 while 死循环包裹着，如果在某一时刻有多条连接有数据可读，那么通过 clientSelector.select(1)方法可以轮询出来，进而批量处理，参见(2)
数据的读写以内存块为单位，参见(3)

总结

根据大佬的讲解，这是大佬的原话：

JDK 的 NIO 编程需要了解很多的概念，编程复杂，对 NIO 入门非常不友好，编程模型不友好，ByteBuffer 的 Api 简直反人类
对 NIO 编程来说，一个比较合适的线程模型能充分发挥它的优势，而 JDK 没有给你实现，你需要自己实现，就连简单的自定义协议拆包都要你自己实现
JDK 的 NIO 底层由 epoll 实现，该实现饱受诟病的空轮询 bug 会导致 cpu 飙升 100%
项目庞大之后，自行实现的 NIO 很容易出现各类 bug，维护成本较高

接下来进入正题，开始学习Netty编程

Netty编程模型

Netty是什么？

Netty 何方神圣？根据我的理解就是它封装了JDK原生的NIO编程，可以让我们用的更爽。 Netty 是一个异步事件驱动的网络应用框架，用于快速开发可维护的高性能服务器和客户端。

优点

Netty底层模型可以随意切换，只需改动几个参数而已；
Netty自带拆包解包，异常机制检测等，让我们只专注于业务逻辑开发；
Netty 解决了 JDK 的很多包括空轮询在内的 Bug；
Netty 底层对线程，selector 做了很多细小的优化，精心设计的 reactor 线程模型做到非常高效的并发处理；
自带各种协议栈：处理任何一种通用协议都几乎不用亲自动手；
Netty 已经历各大 RPC 框架，消息中间件，分布式通信中间件线上的广泛验证，如dubbo、elecsticsearch等底层通信都使用了Netty。

使用Netty编程

引入Maven依赖 <dependency> <groupId>io.netty</groupId> <artifactId>netty-all</artifactId> <version>4.1.6.Final</version> </dependency>

服务端

package top.zhoudl.netty;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.string.StringDecoder;

/**
 * Created by zhoudl on 2018/10/3.
 */
public class NettyServer {

    public static void server (int port) {

        ServerBootstrap serverBootstrap = new ServerBootstrap();
        // boss 对应 IOServer.java 中的接受新连接线程，主要负责创建新连接
        NioEventLoopGroup boss = new NioEventLoopGroup();
        // worker 对应 IOClient.java 中的负责读取数据的线程，主要用于读取数据以及业务逻辑处理
        NioEventLoopGroup worker = new NioEventLoopGroup();
        serverBootstrap.group(boss,worker)
                .channel(NioServerSocketChannel.class)
                .childHandler(new ChannelInitializer<Channel>() {
                    @Override
                    protected void initChannel(Channel channel) throws Exception {
                        channel.pipeline().addLast(new StringDecoder());
                        channel.pipeline().addLast(new SimpleChannelInboundHandler<String>() {
                            @Override
                            protected void channelRead0(ChannelHandlerContext ctx, String msg) {
                                System.out.println(msg);
                            }
                        });
                    }
                })
                .bind(8080);
    }

    public static void main(String[] args) {
        server（8080）；
    }
}

客户端

package top.zhoudl.netty;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringEncoder;
import java.util.Date;
/**
 * Created by zhoudl on 2018/10/3.
 */
public class NettyClient {
    public static void client(String hots,int port) throws InterruptedException {
        Bootstrap bootstrap = new Bootstrap();
        NioEventLoopGroup group = new NioEventLoopGroup();
        // 设置线程组
        bootstrap.group(group)
                // 设置线程模型
                .channel(NioSocketChannel.class)
                // 设置连接读写处理逻辑
                .handler(new ChannelInitializer<Channel>() {
                    @Override
                    protected void initChannel(Channel channel) {
                        channel.pipeline().addLast(new StringEncoder());
                    }
                });
        // 连接指定地址的指定端口
        ChannelFuture connect = bootstrap.connect(hots, port);
        // 判断是否连接成功
        // 实际开发中 在此处肯定是要添加连接重试的逻辑的
        connect.addListener(future -> {
                    if (future.isSuccess()) {
                        System.out.println("连接成功!");
                    } else {
                        System.err.println("连接失败!");
                    }
                });
        Channel channel = connect.channel();
        while (true) {
            channel.writeAndFlush(new Date() + ": hello world!");
            Thread.sleep(2000);
        }
    }
    public static void main(String[] args) throws InterruptedException {
        client("127.0.0.1",8080);
    }
}

总结

生活是好的，峰回路转，柳暗花明，前面总会有另一番不同的风光。

使用了Netty进行NIO编程之后，是不是觉得爽了很多，虽然现在你还不太了解上述代码具体是做什么用的，但我想聪明的你对比着前边的或者是自己已经熟知的Java Socket编程也能猜个一二。所谓学习，便是大胆猜测，小心验证，当最终的结果证实你的猜想之后，你会发现，疯起来，整个世界都是你的！（相应的，沉默下来，整个世界都与你无关）

服务端启动过程分析

最小参数启动举例：

public class NettyServer {
    public static void main(String[] args) {
        NioEventLoopGroup bossGroup = new NioEventLoopGroup();
        NioEventLoopGroup workerGroup = new NioEventLoopGroup();
        ServerBootstrap serverBootstrap = new ServerBootstrap();
        serverBootstrap
                .group(bossGroup, workerGroup)
                .channel(NioServerSocketChannel.class)
                .childHandler(new ChannelInitializer<NioSocketChannel>() {
                    protected void initChannel(NioSocketChannel ch) {
                    }
                });
        serverBootstrap.bind(8080);
    }
}

创建了两个线程NioEventLoopGroup ,我们暂且定义名字为 bossGroup 和 workerGroup ,和传统的IO编程相比，他们俩可以理解为 IO编程中的两个线程组，前者表示监听端口，使用accept()方法建立连接，后者表示每一个连接的读写数据线程组，bossGroup负责接活，workerGroup则负责真正的干活（这也是起这个名字的源头，bossGroup就好比老板，负责接活签合同，而workerGroup则表示下海干活的员工）；
ServerBootstrap是一个启动类，负责引导我们对服务端的操作；
通过 ServerBootstrap 的.group()方法将两大线程组设置进去，也就是说这个引导类至此线程模型就确定了；
制定服务端的线程模型为NIO：.channel(NioServerSocketChannel.class) ,如果要设置成Bio的话对应 OioServerSocketChannel.class
调用 childHandler() 方法，给引导类创建一个ChannelInitializer,这个类是 Netty 对 NIO 类型的连接的抽象，而我们前面NioServerSocketChannel也是对 NIO 类型的连接的抽象，NioServerSocketChannel和NioSocketChannel的概念可以和 BIO 编程模型中的ServerSocket以及Socket两个概念对应上。

总结一点

启动一个Netty服务端，最少需要3个过程

指定线程模型
指定IO模型
连接读写处理逻辑

客户端启动流程概述

学完服务端启动流程之后，客户端启动流程就比较简单了，代码直接参考上述NettyClient.java Netty提供的API个人觉得很优秀，而且大多都简介明了，无论是客户端还是服务端。客户端启动流程大致概括如下：

创建启动类Bootstrap，创建线程NioEventLoopGroup;
利用 Bootstrap的 group()方法为Bootstrap设置线程组，也就是指定线程模型;
指定IO模型：.channel(NioSocketChannel.class) 指定为NIO模型；
处理读写逻辑：.handler(new ChannelInitializer<Channel>() 指定一个handler ;
配置好线程模型、IO 模型、业务处理逻辑之后，开始建立连接，connect()方法的第一个参数为IP或者域名，第二个参数为连接端口号。此外，connect()方法返回的是一个Future，了解Java的应该会知道这是个一个异步的，通过 addListener方法可以监听到连接是否成功，进而打印出连接信息。

客户端启动过程中会有个失败重连的问题：

鉴于失败重连和第一次获取连的逻辑基本分毫不差，所以此处我们很自然的想到了可以使用递归，所以地NettyClient.java中的connect()方法做出封装

private static void connect(Bootstrap bootstrap, String host, int port, int retryCounts) {
    bootstrap.connect(host, port).addListener(future -> {
        if (future.isSuccess()) {
            System.out.println("success!");
        } else if (retryCounts == 0) {
            System.err.println("The number of retries has been used up, giving up the connection！");
        } else {
            // 第几次重新建立连接
            int number = (MAX_RETRY - retryCounts) + 1;
            // 本次重连的时间间隔
            int delay = 1 << order;
            System.err.println(new Date() + ":Connection failed，Reconnected for the"  + number +  "time");
            bootstrap.config().group().schedule(() -> connect(bootstrap, host, port, retryCounts - 1), delay, TimeUnit
                    .SECONDS);
        }
    });
}