Netty框架之深入了解NIO核心组件

Posted 2022-12-07 木兮君

前言

从今天开始，小编开始学习Netty框架，Netty作为底层网络通信框架可以说是无处不在。比如 Duboo、Zookeeper、Elasticsearch、Jboss 等底层都是依赖了它。但很少人会在工作中直接接触到Netty，原因：第一是底层（封装好了使用即可），第二是难。但正因为这样，小编觉得大家更应该主动去掌握它，因为能驾驭Netty，是技术硬实力的体现（可以装，可以吹）。

IO概念

在计算机系统中I/O就是输入（Input）和输出(Output)的意思，针对不同的操作对象，可以划分为磁盘I/O模型，网络I/O模型，内存映射I/O, Direct I/O、数据库I/O等，只要具有输入输出类型的交互系统都可以认为是I/O系统，也可以说I/O是整个操作系统数据交换与人机交互的通道，这个概念与选用的开发语言没有关系，是一个通用的概念。

IO演进

现在系统都有可能有大量文件上传下载操作，大量数据库操作等等，而这些操作都依赖于系统的I/O性能，也就造成了现在系统的瓶颈往往都是由于I/O性能造成的。因此，为了解决I/O性能问题，java也逐步演变出了三代IO模型分别是 BIO、NIO、AIO。

BIO 同步阻塞式 （Blocking I/O）
在java1.4之前 这种传输的实现 只能通过inputStream和OutputStream 实现。这是一种阻塞式IO模型，应对连接数不多的文件系统还好，如果应对的是成千上万的网络服务，这种阻塞式模型就会造成大量的线程占用，造成服务器无法承载更高的并发。

NIO 同步非阻塞式（Non Blocking I/O）
为解决这个问题 java1.4之后引入了NIO ，它通过双向管道进行通信，并且支持以非阻塞式的方式进行，就解决了网络传输导致线程占用的问题。Netty其在底层就是采用这种通信模型。

AIO 同步非阻塞式（Asynchronous I/O）
NIO的非阻塞的实现是依赖选择器 对管道状态进行轮循实现，如果同时进行的管道较多，性能必会受影响，所以java1.7引入了 异步非阻塞式IO，通过异步回调的方式代替选择器。

AIO对比NIO的实现改变在windows下是很明显，在linux系统中不明显。现大部分JAVA系统都是通过linux部署，所以AIO直正被应用的并不广泛。所以我们接下学的学习重点更关注到BIO与NIO的对比。

BIO与NIO模型区别

两组模型最大的别区在于阻塞与非阻塞，而所谓的阻塞是什么呢？而非阻塞又是如何解决的呢？
在阻塞模型中客户端与服务端建立连接后，就会按顺序发生三个事件

1. 客户端把数据写入流中。（阻塞）
2. 通过网络通道(TPC/IP)传输到服务端。（阻塞）
3. 服务端读取。
   这个过程中服务端的线程会一直处阻塞等待，直到数据发送过来后执行第3步。如果在第1步客户端迟迟不写入数据，或者第2步网络传输延迟太高，都会导致服务端线程阻塞时间更长。所以更多的并发，就意味着需要更多的线程来支撑。
   下面是BIO和NIO的简易模型，对于上面三步骤的不同考虑：
   BIO简易模型：
   
   BIO:BIO模型里是通1对1线程来等待第1、2步的完成。

NIO简易模型

NIO:NIO里是指派了选择器（Selector）来检查，是否满足执行第3步的条件，满足了就会通知线程来执行第3步，并处理业务。这样1、2步的延迟就与用于处理业务线程无关。

上面小编画了Channel和Buffer两大组件，接下来小编介绍一下这两大组件。

NIO基础组件

在BIO API中是通过InputStream 与outPutStream 两个流进行输入输出。而NIO 使用一个双向通信的管道代替了它俩。管道(Channel)必须依赖缓冲区（Buffer）实现通信，管道对比流多了一些如：非阻塞、堆外内存映射、零拷贝等特性。因为Channel必须依赖Buffer，那首先介绍一下Buffer。

缓存区（Buffer）

缓冲区就是一个数据容器内部维护了一个数组来存储。Buffer缓冲区并不支持存储任何数据，只能存储一些基本类型，就连字符串也是不能直接存储的。平常用到的最多的就是byte数组。

Buffer 内部结构

在Buffer内部维护了一个数组，同时我们只需要关注四个属性即可：

• capacity：容量， 即内部数组的大小，这个值一但声明就不允许改变。
• position：位置，当前读写位置，默认是0每次读或写一位就会加1。
• mark：标记位置，以便后续调用 reset 将position回到标记位。
• limit：限制，即能够进行读写的最大值，它必须小于或等于capacity。

有了capacity做容量限制为什么还要有limit，原因往Buffer中写数据的时候 不一定会写满，而limit就是用来标记写到了哪个位置，读取的时候就不会超标。如果读取超标就会报：BufferUnderflowException同样写入超标也会报：BufferOverflowException。

当然还有address等属性这个不重要

Buffer核心使用

allocate:声明一个指定大小的Buffer,position为0，limit为容量值。
wrap:基于数组包装一个Buffer，position为0，limit为容量值。

比方说我们使用allocate指定一个为6

put为向数组里面填值，get向数组里面取值，因为position只会向前，且不能超过limit的大小，那当填满后怎么取值呢？这就需要flip操作。
flip操作：为读取写入做好准备，将position 置为0，limit置为原position值

当然除了flip操作为还有其他方式，如clear操作。

clear操作为读取写入做好准备，将position 置为0，limit置为capacity值

注：clear不会实际清空数据

上面小编写到mark操作，他需要和reset一块使用，如果前面没有mark操作，用reset就会报错。
mark操作添加标记，以便后续调用 reset 将position回到标记，常用场景如：替换某一段内容

上面 mark、position、limit、capacity它们有以下规则：mark<= position<= limit<= capacity (当然)。

还要几个较重要的方法，小编就不画图了：
rewind与clear类似将position设为0
remaining 主要是查看还有几个位置可以读取或写入 （limit - positon）
hasRemaining查看是否还可以读取或写入（position < limit）

注意： flip 以及clear与rewind都可以将mark设置为初始值 -1

代码示例：

public class BufferTest 
    @Test
    public void allocateAndExceptionBufferTest() 
        //allocate方法
        IntBuffer intBuffer = IntBuffer.allocate(6);
        System.out.println("capacity: " + intBuffer.capacity() + ", position:" + intBuffer.position() +
                ", limit:" + intBuffer.limit());
        //put方法
        for (int i = 0; i < 6; i++) 
            intBuffer.put(i);
        
        //BufferOverflowException异常
        //写超标
        try 
            intBuffer.put(6);
         catch (Exception exception) 
            System.out.println(exception.getClass().getName());
        
        try 
            //BufferUnderflowException异常
            //读超标
            intBuffer.get();
         catch (Exception exception) 
            System.out.println(exception.getClass().getName());
        
        System.out.println("=================");
    

    @Test
    public void flipAndClearBufferTest() 

        IntBuffer intBuffer = IntBuffer.allocate(6);

        //put方法
        for (int i = 0; i < 4; i++) 
            intBuffer.put(i);
        
        //flip 与clear的主要区别在于limit的位置
        intBuffer.flip();
        System.out.println("capacity: " + intBuffer.capacity() + ", position:" + intBuffer.position() +
                ",after flip limit:" + intBuffer.limit());
        intBuffer.clear();
        System.out.println("capacity: " + intBuffer.capacity() + ", position:" + intBuffer.position() +
                ",after clear limit:" + intBuffer.limit());
        System.out.println("=================");
    

    @Test
    public void markBufferTest() 

        IntBuffer intBuffer = IntBuffer.allocate(6);
        try 
            //没有mark 直接reset 报错
            intBuffer.reset();
         catch (Exception exception) 
            System.out.println(exception.getClass().getName());
        
        //put方法
        for (int i = 0; i < 4; i++) 
            if (i == 1) 
                intBuffer.mark();
            
            intBuffer.put(i);
        
        intBuffer.reset();
        System.out.println("capacity: " + intBuffer.capacity() + ",after reset position:" + intBuffer.position() +
                ", limit:" + intBuffer.limit());

        System.out.println("mark area's value:" + intBuffer.get());
    

测试结果

capacity: 6, position:0,after flip limit:4
capacity: 6, position:0,after clear limit:6
=================
capacity: 6, position:0, limit:6
java.nio.BufferOverflowException
java.nio.BufferUnderflowException
=================
java.nio.InvalidMarkException
capacity: 6,after reset position:1, limit:6
mark area's value:1

以上Buffer小编讲解就告一段落了。

管道（Channel）

管道用于连接文件、网络Socket等。它可同时执行读取和写入这两个I/O 操作，固称双向管道，它有连接和关闭两个状态，在创建管道时处于打开状态，一但关闭 在调用I/O操作就会ClosedChannelException 。通过管道的isOpen 方法可判断其是否处于打开状态。

FileChannel

固名思议它就是用于操作文件的，除常规操作外它还支持以下特性：

• 支持对文件的指定区域进行读写
• 堆外内存映射，进行大文件读写时，可直接映射到JVM声明内存之外，从面提升读写效率。
• 零拷贝技术，通过 transferFrom 或transferTo 直接将数据传输到某个通道，极大提高性能。
• 锁定文件指定区域，以阻止其它程序员进行访问

代码示例
小编只是写了个简单示例，如果是大文件大家想一下怎么写比较好。

public class ChannelTest 

    private static final String TEXT_FILE="C:\\\\workSpaces\\\\lecture-netty\\\\src\\\\test\\\\java\\\\resource\\\\本质思考力20201114.text";
    @Test
    public void fileChannelTest() throws Exception 
        File file = new File(TEXT_FILE);
        FileChannel channel = new RandomAccessFile(file,"rw").getChannel();
        //声明大小
        ByteBuffer buffer= ByteBuffer.allocate(1024);
        channel.read(buffer);
        System.out.println(new String(buffer.array()));
        channel.write(ByteBuffer.wrap("\\\\n 我可以再次写入text中".getBytes()));
    

测试结果：

写完之后：

DatagramChannel

UDP套接字管道，udp 是一个无连接协议，DatagramChannel就是为这个协议提供服务，以接收客户端发来的消息。
代码示例：
小编只是写了个简单示例，（千万别用于生产环境）。

@Test
    public void datagramChannelTest() throws IOException 
        DatagramChannel channel = DatagramChannel.open();
        channel.bind(new InetSocketAddress(8777));
        ByteBuffer buffer = ByteBuffer.allocate(8192);
        while (true) 
        // 接收消息，如果客户端没有消息，则当前会阻塞等待channel.receive(buffer);
            channel.receive(buffer);
            int position = buffer.position();
            byte[] bytes = new byte[position];
            buffer.flip();
            for (int i = 0; i < position; i++) 
                bytes[i] = buffer.get();
            
            String msg = new String(bytes);
            System.out.println(msg);
            if ("exit".equals(msg)) 
                break;
            
            buffer.clear();
        
        channel.close();
    

    @Test
    public void datagramSocketTest() throws IOException 
        Scanner input = new Scanner(System.in);
        DatagramSocket ds = new DatagramSocket();
        InetAddress inet = InetAddress.getByName("127.0.0.1");
        while (true) 
            String next = input.next();
            byte[] date = next.getBytes();
            //创建InetAdress对象，封装自己的IP地址
            DatagramPacket dp = new DatagramPacket(date, date.length, inet, 8777);
            //创建DatagramSocket对象，数据包的发送和接收对象
            //调用ds对象的方法send，发送数据包
            ds.send(dp);
            if ("exit".equalsIgnoreCase(next)) 
                break;
            
        

        ds.close();
    

ServerSocketChannel

TCP套接字管道TCP是一个有连接协议，须建立连接后才能通信。这就需要下面两个管道：ServerSocketChannel ：用于与客户端建立连接，SocketChannel ：用于和客户端进行消息读写。
代码示例
注意事项如上。

    @Test
    public void socketChannelTest() throws IOException 
        // 1.打开TCP服务管道
        ServerSocketChannel channel = ServerSocketChannel.open();
        // 2.绑定端口
        channel.bind(new InetSocketAddress(8080));
        // 3.接受客户端发送的连接请求，(如果没有则阻塞)
        while (true) 
            SocketChannel socketChannel = channel.accept();
            // 使用子线程处理请求
            new Thread(() -> handle(socketChannel)).start();
        
    

    private void handle(SocketChannel socketChannel) 
        try 
            ByteBuffer buffer = ByteBuffer.allocate(1024);
            // 1.读取客户端发来的消息(如果没有则阻塞)
            socketChannel.read(buffer);
            // 返回消息
            socketChannel.write(ByteBuffer.wrap("返回消息".getBytes()));
            // 关闭管道
            socketChannel.close();
         catch (IOException e) 
            e.printStackTrace();
        
    
	@Test
    public void socketChannelTest() throws IOException 
        SocketChannel socketChannel = SocketChannel.open();
        socketChannel.connect(new InetSocketAddress(8080));
        ByteBuffer buffer = ByteBuffer.allocate(1024);
        buffer.put("hello world".getBytes());
        buffer.flip();
        socketChannel.write(buffer);

        buffer.clear();
        socketChannel.read(buffer);
        System.out.println(new String(buffer.array()));
        socketChannel.close();
    

总结

今天所介绍的内容相对比较简单，希望各位大佬和小编多多交流，积极评论，Netty新的旅程中一起成长。加油！