Java NIO概念

Java NIO是什么？

Java NIO，全称为Java Non-blocking Input/Output或New IO，是Java平台从JDK 1.4版本开始引入的一套新的输入/输出API。它旨在提供一种更高效、可扩展性更强的IO操作方式，特别适合构建高性能的网络应用和进行大容量的数据传输。Java NIO的关键特性包括：

非阻塞（Non-blocking）: NIO支持非阻塞模式，允许在数据未准备就绪时，线程可以继续执行其他任务，而不是阻塞等待数据，从而提高了程序的效率和响应能力。
缓冲区（Buffers）: NIO的所有数据都是通过缓冲区（Buffer）处理的。缓冲区实质上是一个可以写入数据，然后再次读取数据的内存块。数据总是先读取到缓冲区或从缓冲区写入，这提高了处理速度并允许在处理大量数据时有效地复用内存。
通道（Channels）: 通道是NIO中用于读取和写入数据的抽象，它们类似于流，但是可以进行双向数据传输，并且总是与缓冲区一起使用。通道可以打开文件、网络套接字等，支持更灵活的数据传输。
选择器（Selectors）: 选择器是NIO中的一个强大组件，它允许单个线程管理多个通道，监控多个通道上的事件（如连接打开、数据可读等），实现了IO多路复用，大大提高了并发处理能力。
直接缓冲区（Direct Buffers）: NIO还支持直接缓冲区，可以直接在物理内存中分配空间，绕过JVM的堆内存，减少数据复制，提高数据传输速率。

Java NIO为什么非阻塞？

Java NIO之所以被认为是非阻塞的，主要是因为它采用了IO多路复用技术，特别是通过Selector（选择器）这一核心组件来实现的。以下是几个关键点解释Java NIO如何达到非阻塞效果：

IO多路复用模型：Java NIO基于Select/Poll/Epoll等机制，这些机制允许一个单独的线程管理多个通道（Channel）。在没有事件就绪（如数据可读、可写或新连接到达）时，线程不会阻塞等待，而是继续执行其他任务或检查其他通道的状态。当有事件发生时，Selector会通知应用程序，从而实现非阻塞式的通知机制。
非阻塞通道操作：在NIO中，通道可以被配置成非阻塞模式。这意味着当从通道读取数据或向通道写入数据时，如果没有数据可读或缓冲区没有足够的空间写入，操作会立即返回，而不是阻塞线程。应用程序可以根据操作的返回状态决定是否继续尝试读写或进行其他处理。
事件驱动编程：开发者可以注册对特定事件（如读、写、连接等）的兴趣到Selector上。当这些事件发生时，Selector通过选择操作（selector.select()）告知应用程序哪些通道已准备好进行IO操作。这样，应用程序可以针对性地处理这些就绪的通道，无需为每个通道或操作分配独立的线程，避免了不必要的线程上下文切换和阻塞。

总结来说，Java NIO通过Selector实现的IO多路复用模型和非阻塞通道操作，使得它能够在单个或少量线程中高效地管理大量并发连接，避免了传统阻塞IO模型中线程被挂起等待的问题，从而提高了系统在高并发环境下的性能和资源利用率。尽管在调用selector.select()方法时线程看似“等待”，但实际上它是非阻塞地等待事件发生，一旦有事件就绪便立刻继续执行，因此整体上被视为非阻塞IO。

Select/Poll/Epoll机制

Select、Poll、和Epoll都是操作系统提供的用于实现I/O多路复用的技术，主要用于解决单线程（或少量线程）处理大量并发I/O连接的问题，特别是在网络编程中非常有用。它们都允许程序同时监控多个文件描述符（如socket），并在至少一个描述符变得可读或可写时通知程序，而不是为每个连接创建一个新的线程或进程。下面是它们之间的一些关键区别：

Select

历史最悠久：Select是最原始的I/O多路复用函数，几乎在所有类UNIX系统中都可用，也包括Windows。
文件描述符限制：Select受限于FD_SETSIZE，这是一个编译时常量，默认值通常是1024，意味着它能同时监控的文件描述符数量有限。
效率问题：每次调用select时，都需要将用户态的文件描述符集合复制到内核态，然后内核检查每个文件描述符的状态，即使大多数都没有变化，这导致了较大的性能开销。
水平触发：Select默认使用水平触发模式，即如果一个文件描述符已经就绪并且没有被处理，那么下次调用select时还会报告这个文件描述符。

Poll

改进的结构：Poll改进了Select的缺点，使用链表结构存储文件描述符，因此不再受FD_SETSIZE限制，理论上可以支持更多文件描述符。
性能提升：虽然Poll避免了文件描述符集合大小的限制，但它仍然需要遍历整个文件描述符列表来查找就绪的描述符，这在大量连接下效率不高。
模式相似：Poll的工作模式与Select相似，也是在每次调用时检查所有文件描述符的状态。

Epoll

Linux特有：Epoll是Linux特有的I/O多路复用技术，设计上更加高效。
事件驱动：Epoll采用事件驱动模型，仅当文件描述符状态发生变化时才通知应用程序，这减少了不必要的遍历。
高效检查：Epoll通过维护一个内核事件表来跟踪每个文件描述符的状态，避免了重复的遍历操作，性能远超Select和Poll。
两种触发模式：Epoll支持水平触发（LT）和边缘触发（ET）两种模式。LT模式下，只要文件描述符就绪，就会持续通知；而ET模式下，仅当状态从非就绪变为就绪时才通知一次，要求应用程序正确处理，以避免丢失事件。
内存高效：Epoll通过在内核中保存事件集合，减少了用户态与内核态之间的数据复制。

什么是I/O多路复用？

I/O多路复用（I/O Multiplexing）是一种在单个线程或进程中同时管理多个I/O操作的技术，特别适用于需要同时处理多个输入输出通道的场景，如网络服务器处理多个客户端连接。这项技术允许程序监视多个文件描述符（例如，网络套接字、管道、常规文件等），等待其中的一个或多个变为“就绪”状态（即可以进行读取或写入操作），而无需为每个描述符分配单独的线程或进程。

I/O多路复用的核心优势在于它减少了系统中线程或进程的数量，从而降低了资源消耗和上下文切换的开销，提高了系统在高并发环境下的性能和可扩展性。在使用I/O多路复用时，程序首先会注册关注的文件描述符及其感兴趣的事件（如可读、可写、异常等）到一个中心控制器（如select()函数中的fd_set，或epoll_create()创建的epoll实例）。随后，程序调用相应的多路复用函数（如select()、poll()、或epoll_wait()）进入等待状态。当有任何一个文件描述符上的事件发生时，这些函数将返回，并告知应用程序哪些描述符已就绪，随后程序可以针对这些就绪的描述符进行读写操作，从而实现了并发处理多个I/O请求的目的。

I/O多路复用和NIO有啥关系？

I/O多路复用（Multiplexing）和Java中的NIO（New Input/Output）是密切相关的概念，但它们分别代表了操作系统层面和编程语言层面的不同技术。

I/O多路复用：这是操作系统提供的一种机制，允许单个进程或线程同时监控多个I/O通道（如文件描述符、网络套接字等），并能知晓何时某个通道准备好进行读取或写入操作。常见的I/O多路复用技术包括select、poll和epoll（Linux特有）。这种机制对于编写高性能的服务器程序特别重要，因为它能减少线程上下文切换的开销，并有效管理大量并发连接。

Java NIO： Java NIO（Non-blocking I/O）是Java平台的一部分，它引入了一系列新的API来支持非阻塞式的I/O操作，以及利用操作系统提供的I/O多路复用能力。Java NIO中的核心组件之一是Selector（选择器），它允许一个线程管理多个通道（Channel），并能够监听这些通道上的多种事件（如连接就绪、读就绪、写就绪等）。实质上，Selector背后使用的就是操作系统提供的I/O多路复用机制（在Linux上可能是epoll，在其他系统上可能是select或poll）。

关系总结：

技术层次：I/O多路复用是操作系统层面上的概念和技术，而NIO是Java语言级别的API和编程模型。
实现结合：Java NIO通过Selector实现了对I/O多路复用的支持，使得开发者能够用更少的线程处理更多的并发连接，提高了应用的性能和可扩展性。
非阻塞特性：NIO不仅利用了多路复用，还强调非阻塞操作，即在读写操作不可立即完成时，不会阻塞当前线程，而是立即返回，之后可以通过轮询或事件通知来得知操作是否完成。

因此，可以说Java NIO是对I/O多路复用机制的一种高级封装和利用，它使开发者能够以更简洁、更面向对象的方式编写高性能的网络服务程序。

NIO为何不阻塞？

很多人应该对“Java NIO 是非阻塞的 I/O”这一信条熟记于心，但其中的有些人可能经过实践之后却产生这样的疑惑：Java NIO 明明是非阻塞的 I/O，但 Java NIO 中无论是 Channel 还是 Selector 的方法却是阻塞的，其中的一个被称为设置 Channel 为非阻塞的方法 XXXChannel.configureBlocking(false) 看起来并没有所言的拥有“将 Channel 变为非阻塞的”的作用，那究竟凭什么说 Java NIO 是非阻塞的 I/O 呢？ “Java NIO 是非阻塞的 I/O”，这一论断实际上是谬论，Java NIO 中的 N，并不是指的是 Non-blocking（非阻塞）的意思，而是 New（新）的意思。Java NIO 只是相对于原来所谓 Java BIO 的一个升级，并不能就说它就是一个非阻塞的 I/O。不过，相对于 Java BIO，Java NIO 确实有一些非阻塞的特性。具体来说，Java NIO 的非阻塞，是相对于连接的非阻塞，而不是指方法调用时的非阻塞。

reactor模式

Reactor 模式就是基于建立连接与具体服务之间线程分离的模式。在 Reactor 模式中，会有一个线程负责与所有客户端建立连接，这个线程通常称之为 Reactor。然后在建立连接之后，Reactor 线程会使用其它线程（可以有多个）来处理与每一个客户端之间的数据传输，这个（些）线程通常称之为 Handler。由于服务端需要与多个客户端通信，它的通信是一对多的关系，所以它需要使用 Reactor 模式。对客户端，它只需要与服务端通信，它的通信是一对一的关系，所以它不需要使用 Reactor 模式。也就是说，对客户端来讲，它不需要进行建立连接与传输数据之间的线程分离。

Windows支持Epoll吗？

Windows操作系统不支持Linux中的epoll机制。epoll是Linux系统内核提供的一种高效的I/O多路复用技术，特别适合处理大量并发的文件描述符（如网络套接字）。

Windows平台上有与epoll功能类似的替代方案，即I/O完成端口（I/O Completion Ports, IOCP）。IOCP是Windows下用于实现高性能异步I/O处理的机制，它允许应用程序以高效的方式管理多个I/O操作，尤其是在面对大量并发连接时表现出色。与epoll一样，IOCP也能让单个线程（或少量线程）管理大量并发的输入输出操作，而无需为每个操作创建单独的线程，从而提高了系统效率和可扩展性。

虽然两者都用于处理并发I/O，但它们的API和内部实现机制有所不同，因此在跨平台开发时，需要根据目标操作系统选择相应的I/O多路复用技术。在涉及跨平台的网络编程框架中，如Netty，会根据运行时的平台自动选择最适合的I/O模型（在Linux上使用epoll，在Windows上使用IOCP）。

文件描述符是什么？

文件描述符（File Descriptor，简称FD）是计算机科学中的一个概念，特别是在Unix/Linux操作系统及其衍生系统中广泛使用。它是一个非负整数，作为操作系统内核用来标识已经打开的文件或输入输出资源（如管道、套接字等）的一种抽象。文件描述符对于每个进程是唯一的，并且每个进程都维护着一张文件描述符表，该表映射文件描述符到内核中对应的打开文件结构。

文件描述符的主要作用和特性包括：

访问标识：文件描述符是进程用来访问文件的唯一标识符。当进程通过系统调用（如open()、socket()）打开一个文件或建立一个网络连接时，内核会返回一个文件描述符给进程。
I/O操作：所有涉及输入输出的操作，如读取（read()）、写入（write()）、以及控制操作（如lseek()），都需要通过文件描述符来指定要操作的文件或资源。
默认描述符：每个进程在启动时都会自动拥有三个标准文件描述符：0（标准输入stdin），1（标准输出stdout），和2（标准错误stderr）。
资源管理：文件描述符帮助操作系统高效管理打开的文件和资源，包括跟踪打开的文件状态、权限控制以及资源分配等。
可重用性：在同一个进程中，可以多次打开同一个文件，每个打开实例都会获得一个不同的文件描述符。此外，不同进程也可以各自拥有指向同一文件的文件描述符。
限制：尽管早期Unix系统中文件描述符的数量限制较小（例如，最多19或20），现代系统通常支持更大的数量，可以达到数千甚至更多，具体限制取决于系统配置和资源。

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.util.Iterator;

public class NioServer {
    public static void main(String[] args) throws IOException {
        // 创建ServerSocketChannel，用于监听客户端的连接请求。
        ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
        serverSocketChannel.configureBlocking(false); // 设置为非阻塞模式，意味着调用如accept()时不阻塞，如果没有连接立即可用则立即返回。
        serverSocketChannel.bind(new InetSocketAddress(8080)); // 绑定端口，，开始监听客户端的连接。

        // 创建Selector，用于管理多个通道的I/O事件。
        Selector selector = Selector.open();

        // 注册ServerSocketChannel到Selector，监听ACCEPT事件
        serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);

        System.out.println("Server started, listening on port 8080...");
        //进入一个无限循环，持续监听和处理客户端的连接和数据读取。
        while (true) {
            selector.select(); // 阻塞直到至少有一个通道就绪（即至少有一个事件发生）
            //这些键表示已发生特定事件的通道。
            Iterator<SelectionKey> keyIterator = selector.selectedKeys().iterator();
            while (keyIterator.hasNext()) {
                SelectionKey key = keyIterator.next();
                //获取当前迭代的键，然后从集合中移除它，以避免重复处理。
                keyIterator.remove();
                //键代表一个新连接请求，接受该连接，并将客户端的SocketChannel注册到Selector，关注OP_READ事件，准备读取数据。
                if (key.isAcceptable()) {
                    ServerSocketChannel server = (ServerSocketChannel) key.channel();
                    SocketChannel client = server.accept(); // 接受来自客户端的连接请求，创建一个SocketChannel实例。
                    client.configureBlocking(false);//将客户端的SocketChannel也设置为非阻塞模式。
                    client.register(selector, SelectionKey.OP_READ); // 注册读事件，将客户端通道注册到Selector，关注读事件。
                    System.out.println("Accepted connection from " + client);
                } else if (key.isReadable()) {//当客户端有数据可读时，读取数据并处理。
                    SocketChannel client = (SocketChannel) key.channel();
                    ByteBuffer buffer = ByteBuffer.allocate(1024);
                    int bytesRead = client.read(buffer); // 读取数据，从客户端通道读取数据到ByteBuffer中。
                    if (bytesRead > 0) {
                        buffer.flip();//使用buffer.flip()切换到读模式，输出读取到的数据。
                        System.out.println("Message received: " + new String(buffer.array(), 0, bytesRead));
                        buffer.clear();//清空缓冲区buffer.clear()，准备下一次读取。
                        ByteBuffer response = ByteBuffer.wrap("Hello from server".getBytes());
                        client.write(response); // 写入响应，准备并发送响应数据到客户端：client.write(response)。
                    } else {
                        client.close(); // 关闭连接
                    }
                }
            }
        }
    }
}

客户端：

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SocketChannel;

public class NioClient {
    public static void main(String[] args) throws IOException {
        //初始化一个SocketChannel实例，这是客户端用来与服务器通信的通道。
        SocketChannel socketChannel = SocketChannel.open();
        socketChannel.configureBlocking(false); // 设置为非阻塞模式，将SocketChannel设置为非阻塞模式，意味着读写操作不会阻塞，即使没有数据可读或没有足够的空间可写也会立即返回。
        boolean isConnected = socketChannel.connect(new InetSocketAddress("localhost", 8080));//尝试连接到指定地址（本例中为本地主机的8080端口）的服务器。由于是非阻塞模式，connect可能不会立即完成，因此返回一个布尔值指示连接是否立即成功。

        if (!isConnected) {
            //使用while循环调用socketChannel.finishConnect()，直到连接完成。这用于处理非阻塞连接时的延迟情况，确保连接成功建立。
            while (!socketChannel.finishConnect()) {
                // 可以在这里添加重试逻辑或处理连接延迟的情况
            }
        }
        //创建一个ByteBuffer，并用要发送的消息填充它。这里的消息是"Hello, Server!"。
        ByteBuffer buffer = ByteBuffer.wrap("Hello, Server!".getBytes());
        socketChannel.write(buffer); // 发送消息，将缓冲区中的数据写入通道，发送给服务器。

        buffer.clear();//在读取响应前清空缓冲区，准备接收数据。
        int bytesRead = socketChannel.read(buffer); // 读取响应，从通道读取服务器的响应到缓冲区。由于是非阻塞读，若没有数据可读，read会立即返回0。
        if (bytesRead > 0) {
            buffer.flip();//如果读取到了数据（bytesRead > 0），则翻转缓冲区(buffer.flip())以准备读取数据，然后输出接收到的消息内容。
            System.out.println("Response from server: " + new String(buffer.array(), 0, bytesRead));
        }

        socketChannel.close();//完成通信后，关闭SocketChannel释放资源。
    }
}

再看一个Netty的例子

package com.allen.netty_demo;
 
import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.oio.OioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.oio.OioServerSocketChannel;
 
import java.net.InetSocketAddress;
import java.nio.charset.Charset;
 
/**
 * @author ：jhys
 * @date ：Created in 2021/7/15 11:23
 * @Description ：
 */
public class NettyOioServer {
    public void server(int port) throws Exception {
        //创建共享ByteBuf：使用Unpooled.unreleasableBuffer创建了一个不可释放的ByteBuf实例，内容为"Hi!\r\n"，字符集为UTF-8。这个缓冲区将用于向每个连接的客户端发送消息。
        final ByteBuf buf = Unpooled.unreleasableBuffer(Unpooled.copiedBuffer("Hi!\r\n", Charset.forName("UTF-8")));
        //创建OioEventLoopGroup：EventLoopGroup group = new OioEventLoopGroup(); 创建了一个基于OIO的事件循环组，用于处理I/O操作和事件。
        EventLoopGroup group = new OioEventLoopGroup();
 
        try {
            //配置并启动服务器： - 使用ServerBootstrap引导类配置服务器。 - bootstrap.group(group) 指定使用的事件循环组。 - .channel(OioServerSocketChannel.class) 指定服务器通道类型为OIO的OioServerSocketChannel。 - .localAddress(new InetSocketAddress(port)) 设置服务器监听的本地地址和端口。 - .childHandler(...) 配置一个ChannelInitializer，用于初始化每个新接受的连接的管道（ChannelPipeline）。 - 在initChannel方法中，向管道添加了一个ChannelInboundHandlerAdapter实例。 - 重写了channelActive方法，当客户端连接激活时，通过ctx.writeAndFlush(buf.duplicate()).addListener(ChannelFutureListener.CLOSE);发送缓冲区中的消息给客户端，并在消息发送完毕后关闭连接。
            ServerBootstrap bootstrap = new ServerBootstrap();
            bootstrap.group(group)                                    //2
                    .channel(OioServerSocketChannel.class)
                    .localAddress(new InetSocketAddress(port))
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            socketChannel.pipeline().addLast(new ChannelInboundHandlerAdapter() {
                                @Override
                                public void channelActive(ChannelHandlerContext ctx) throws Exception {
                                    ctx.writeAndFlush(buf.duplicate()).addListener(ChannelFutureListener.CLOSE);//5
                                }
                            });
                        }
                    });
            //绑定并同步等待：ChannelFuture future = bootstrap.bind().sync(); 绑定服务器到指定端口，并通过sync()方法同步等待直到绑定完成。
            ChannelFuture future  = bootstrap.bind().sync();
            //等待服务器关闭：future.channel().closeFuture().sync(); 等待服务器关闭的未来，意味着程序会在此阻塞，直到服务器关闭。
            future.channel().closeFuture().sync();
        } finally {
            //优雅关闭资源：在finally块中，使用group.shutdownGracefully().sync();优雅地关闭事件循环组，确保所有资源被正确释放。
            group.shutdownGracefully().sync();        
        }
    }
}

Netty作为客户端连接一个Http Restful接口的例子

netty即使没有netty对应的服务端，也可以使用。

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.http.DefaultFullHttpRequest;
import io.netty.handler.codec.http.FullHttpResponse;
import io.netty.handler.codec.http.HttpHeaderNames;
import io.netty.handler.codec.http.HttpMethod;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.HttpRequestEncoder;
import io.netty.handler.codec.http.HttpResponseDecoder;
import io.netty.handler.codec.http.HttpVersion;

import java.net.URI;
import java.nio.charset.StandardCharsets;

public class NettyHttpClient {

    public static void main(String[] args) throws Exception {
        String host = "localhost";
        int port = 8080; // Spring MVC服务端口
        String uriPath = "/api/your-endpoint"; // 替换为你的API路径

        EventLoopGroup workerGroup = new NioEventLoopGroup();

        try {
            Bootstrap b = new Bootstrap();
            b.group(workerGroup);
            b.channel(NioSocketChannel.class);
            b.handler(new ChannelInitializer<Channel>() {
                @Override
                protected void initChannel(Channel ch) throws Exception {
                    ch.pipeline()
                     .addLast(new HttpResponseDecoder())
                     .addLast(new HttpRequestEncoder())
                     .addLast(new HttpObjectAggregator(65536)) // 适应较大的HTTP响应
                     .addLast(new SimpleClientHandler(uriPath));
                }
            });

            ChannelFuture f = b.connect(host, port).sync();
            f.addListener((ChannelFutureListener) future -> {
                if (future.isSuccess()) {
                    System.out.println("Connected to " + host + ":" + port);
                } else {
                    System.err.println("Failed to connect");
                    future.cause().printStackTrace();
                }
            });

            f.channel().closeFuture().sync();
        } finally {
            workerGroup.shutdownGracefully();
        }
    }

    static class SimpleClientHandler extends ChannelInboundHandlerAdapter {

        private final String uriPath;
        private boolean readingChunks;

        public SimpleClientHandler(String uriPath) {
            this.uriPath = uriPath;
        }

        @Override
        public void channelActive(ChannelHandlerContext ctx) {
            URI uri = URI.create("http://" + ctx.channel().remoteAddress() + uriPath);
            DefaultFullHttpRequest request = new DefaultFullHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.GET, uri.getRawPath());
            request.headers().set(HttpHeaderNames.HOST, uri.getHost());
            request.headers().set(HttpHeaderNames.CONNECTION, HttpHeaderValues.CLOSE);
            ctx.writeAndFlush(request);
        }

        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) {
            if (msg instanceof FullHttpResponse) {
                FullHttpResponse response = (FullHttpResponse) msg;
                System.out.println("Response received: " + response.status() + ", Content: " + response.content().toString(StandardCharsets.UTF_8));
                ctx.close();
            }
        }

        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
            cause.printStackTrace();
            ctx.close();
        }
    }
}