概述

NIO中的N可以理解为 Non-blocking,一种同步非阻塞的I/O模型,在Java 1.4 中引入

NIO新增了 ChannelSelectorBuffer 等抽象概念,支持面向缓冲、基于通道的I/O操作方法

NIO 提供了与传统 BIO 模型中的 Socket 和 ServerSocket 相对应的 SocketChannel 和 ServerSocketChannel 两种不同的套接字通道实现

NIO 这两种通道都支持阻塞和非阻塞两种模式。阻塞模式使用就像传统中的支持一样,比较简单,但是性能和可靠性都不好;非阻塞模式正好与之相反

对于低负载、低并发的应用程序,可以使用同步阻塞 I/O 来提升开发效率和更好的维护性;对于高负载、高并发的(网络)应用,应使用 NIO 的非阻塞模式来开发

架构图

netty-第 2 页.drawio.png

说明

  1. Channel:可以理解为通道
  2. Selector:可以理解为事件选择器
  3. Buffer:可以理解为数据缓冲流

代码示例

客户端

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
package org.example.nio.example;

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.SocketChannel;
import java.util.Iterator;
import java.util.Set;

/**
* NIO客户端
*/
public class NIOClient {

// 通道管理器(Selector)
private static Selector selector;

public static void main(String[] args) throws IOException {
// 创建通道管理器(Selector)
selector = Selector.open();
// 创建通道SocketChannel
SocketChannel channel = SocketChannel.open();
// 将通道设置为非阻塞
channel.configureBlocking(false);
// 客户端连接服务器,其实方法执行并没有实现连接,需要在handleConnect方法中调channel.finishConnect()才能完成连接
channel.connect(new InetSocketAddress("127.0.0.1", 9090));
/**
* 将通道(Channel)注册到通道管理器(Selector),并为该通道注册selectionKey.OP_CONNECT
* 注册该事件后,当事件到达的时候,selector.select()会返回,
* 如果事件没有到达selector.select()会一直阻塞。
*/
channel.register(selector, SelectionKey.OP_CONNECT);
// 循环处理
while (true) {
/*
* 选择一组可以进行I/O操作的事件,放在selector中,客户端的该方法不会阻塞,
* selector的wakeup方法被调用,方法返回,而对于客户端来说,通道一直是被选中的
* 这里和服务端的方法不一样,查看api注释可以知道,当至少一个通道被选中时。
*/
selector.select();
// 获取监听事件
Set selectionKeys = selector.selectedKeys();
Iterator iterator = selectionKeys.iterator();
// 迭代处理
while (iterator.hasNext()) {
// 获取事件
SelectionKey key = iterator.next();
// 移除事件,避免重复处理
iterator.remove();
// 检查是否是一个就绪的已经连接服务端成功事件
if (key.isConnectable()) {
handleConnect(key);
} else if (key.isReadable()) {// 检查套接字是否已经准备好读数据
handleRead(key);
}
}
}
}

/**
* 处理客户端连接服务端成功事件
*/
private static void handleConnect(SelectionKey key) throws IOException {
// 获取与服务端建立连接的通道
SocketChannel channel = (SocketChannel) key.channel();
if (channel.isConnectionPending()) {
// channel.finishConnect()才能完成连接
channel.finishConnect();
}
channel.configureBlocking(false);
// 数据写入通道
String msg = "Hello Server!";
channel.write(ByteBuffer.wrap(msg.getBytes()));
// 通道注册到选择器,并且这个通道只对读事件感兴趣
channel.register(selector, SelectionKey.OP_READ);
}

/**
* 监听到读事件,读取客户端发送过来的消息
*/
private static void handleRead(SelectionKey key) throws IOException {
SocketChannel channel = (SocketChannel) key.channel();
// 从通道读取数据到缓冲区
ByteBuffer buffer = ByteBuffer.allocate(128);
channel.read(buffer);
// 输出服务端响应发送过来的消息
byte[] data = buffer.array();
String msg = new String(data).trim();
System.out.println("服务端发来的消息:" + msg);
}
}

服务端

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
package org.example.nio.example;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
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;
import java.util.Set;

/**
* NIO服务端
*/
public class NIOServer {

// 通道管理器(Selector)
private static Selector selector;

public static void main(String[] args) throws IOException {
// 创建通道管理器(Selector)
selector = Selector.open();
// 创建通道ServerSocketChannel
ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
// 将通道设置为非阻塞
serverSocketChannel.configureBlocking(false);
// 将ServerSocketChannel对应的ServerSocket绑定到指定端口(port)
ServerSocket serverSocket = serverSocketChannel.socket();
serverSocket.bind(new InetSocketAddress(9090));
/**
* 将通道(Channel)注册到通道管理器(Selector),并为该通道注册selectionKey.OP_ACCEPT事件
* 注册该事件后,当事件到达的时候,selector.select()会返回,
* 如果事件没有到达selector.select()会一直阻塞。
*/
serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
// 循环处理
while (true) {
// 当注册事件到达时,方法返回,否则该方法会一直阻塞
selector.select();
// 获取监听事件
Set selectionKeys = selector.selectedKeys();
Iterator iterator = selectionKeys.iterator();
// 迭代处理
while (iterator.hasNext()) {
// 获取事件
SelectionKey key = iterator.next();
// 移除事件,避免重复处理
iterator.remove();
// 检查是否是一个就绪的可以被接受的客户端请求连接
if (key.isAcceptable()) {
handleAccept(key);
} else if (key.isReadable()) {// 检查套接字是否已经准备好读数据
handleRead(key);
}
}
}
}

/**
* 处理客户端连接成功事件
*/
private static void handleAccept(SelectionKey key) throws IOException {
// 获取客户端连接通道
ServerSocketChannel server = (ServerSocketChannel) key.channel();
SocketChannel socketChannel = server.accept();
socketChannel.configureBlocking(false);
// 信息通过通道发送给客户端
String msg = "Hello Client!";
socketChannel.write(ByteBuffer.wrap(msg.getBytes()));
// 给通道设置读事件,客户端监听到读事件后,进行读取操作
socketChannel.register(selector, SelectionKey.OP_READ);
}


/**
* 监听到读事件,读取客户端发送过来的消息
*/
private static void handleRead(SelectionKey key) throws IOException {
SocketChannel channel = (SocketChannel) key.channel();
// 从通道读取数据到缓冲区
ByteBuffer buffer = ByteBuffer.allocate(128);
channel.read(buffer);
// 输出客户端发送过来的消息
byte[] data = buffer.array();
String msg = new String(data).trim();
System.out.println("server received msg from client:" + msg);
}
}

通道channel

工作模式

netty-第 7 页.drawio.png

通道有点类似于IO中的流,但不同的是,同一个通道既可以读也可以写,而任意一个流要么只能读要么只能写。但通道和流一样都是需要基于物理文件的,而每个流或者通道都通过文件指针操作文件,这里说的通道是双向的也是有前提的,那就是通道基于随机访问文件RandomAccessFile的可读可写文件指针。

基本的通道类型有:

FileChannel 是基于文件的通道;

SocketChannel 和 ServerSocketChannel 用于网络 TCP 套接字数据报读写;

DatagramChannel 是用于网络 UDP 套接字数据报读写;

通道不能单独存在,它永远需要绑定一个缓存区,所有的数据只会存在于缓存区中,无论你是写或是读,必然是缓存区通过通道到达磁盘文件,或是磁盘文件通过通道到达缓存区。即缓存区是数据的起点,也是终点。

缓冲区Buffer

缓冲区(Buffer):一个用于特定基本数据类型的容器。由java.nio包定义的,所有缓冲区都是Buffer抽象类的子类。

Java NIO 中的 Buffer 主要用于和 NIO 通道进行交互,数据是从通道读入到缓冲区的,然后从缓冲区中写入到通道中的。

Buffer 就像一个数组,可以保存多个相同类型的数据。根据数据类型的不同(boolean)除外,有以下 Buffer 常用子类:ByteBuffer、CharBuffer、ShortBuffer、IntBuffer、LongBuffer、FloatBuffer、DoubleBuffer。

上述 Buffer 类他们都是通过相似的方法进行管理数据的,只是各自管理的数据类型不同而已。

缓冲区的基本属性

  • 容量(capacity):表示 Buffer 最大数据容量,缓冲区容量不能为负,并且一旦创建不能更改。
  • 限制(limit):第一个不应该读取或写入的数据的索引,即位于 limit 后的数据不可读写。缓冲区的限制不能为负,并且不能大于容量。
  • 位置(position):下一个要读取或写入的数据的索引。缓冲区的位置不能为负,并且不能大于其限制。
  • 标记(mark)和重置(reset):标记是一个索引,通过Buffer中的mark()方法指定Buffer中的一个特定的position,之后可以通过调用reset()方法恢复到这个position。

简而言之:0 <= mark <= position <= limit <= capacity。

Selector(选择器)

Selector 被称为选择器 ,当然你也可以翻译为多路复用器 。它是Java NIO 核心组件中的一个,用于检查一个或多个 Channel(通道)的状态是否处于连接就绪、接受就绪、可读就绪、可写就绪。

如此可以实现单线程管理多个 channels,也就是可以管理多个网络连接。

netty-第 8 页.drawio.png

使用 Selector 的好处在于: 相比传统方式使用多个线程来管理 IO,Selector 使用了更少的线程就可以处理更多通道,减少多线程之间频繁进行上下文切换带来的性能损耗。

创建一个选择器一般是通过 Selector 的工厂方法,Selector.open :

1
Selector selector = Selector.open();

而一个通道想要注册到某个选择器中,必须调整模式为非阻塞模式,例如:

1
2
3
4
5
6
//创建一个 TCP 套接字通道
SocketChannel channel = SocketChannel.open();
//调整通道为非阻塞模式
channel.configureBlocking(false);
//向选择器注册一个通道
SelectionKey key = channel.register(selector, SelectionKey.OP_READ);

以上代码是注册一个通道到选择器中的最简单版本,支持注册选择器的通道都有一个 register 方法,该方法就是用于注册当前实例通道到指定选择器的。

该方法的第一个参数就是目标选择器,第二个参数其实是一个二进制掩码,它指明当前选择器感兴趣当前通道的哪些事件。以枚举类型提供了以下几种取值:

1
2
3
4
int OP_READ = 1 << 0;
int OP_WRITE = 1 << 2;
int OP_CONNECT = 1 << 3;
int OP_ACCEPT = 1 << 4;

register 方法会返回一个 SelectionKey 实例,该实例代表的就是选择器与通道的一个关联关系。你可以调用它的 selector 方法返回当前相关联的选择器实例,也可以调用它的 channel 方法返回当前关联关系中的通道实例。

除此之外,SelectionKey 的 readyOps 方法将返回当前选择感兴趣当前通道中事件中准备就绪的事件集合,依然返回的一个整型数值,也就是一个二进制掩码。

1
int readySet = selectionKey.readyOps();

假如 readySet 的值为 13,二进制 「0000 1101」,从后向前数,第一位为 1,第三位为 1,第四位为 1,那么说明选择器关联的通道,读就绪、写就绪,连接就绪。

所以,当我们注册一个通道到选择器之后,就可以通过返回的 SelectionKey 实例监听该通道的各种事件。

当然,一旦某个选择器中注册了多个通道,我们不可能一个一个的记录它们注册时返回的 SelectionKey 实例来监听通道事件,选择器应当有方法返回所有注册成功的通道相关的 SelectionKey 实例。

1
Set<SelectionKey> keys = selector.selectedKeys();

selectedKeys 方法会返回选择器中注册成功的所有通道的 SelectionKey 实例集合。我们通过这个集合的 SelectionKey 实例,可以得到所有通道的事件就绪情况并进行相应的处理操作。

总结

优点

1个线程就行就能处理所有连接,这个线程不停循环遍历就行了。

缺点

单线程不停循环发起系统调用,一样会耗尽 CPU 资源。

NIO的瓶颈

在于需要不停的调起系统调用,每个链接我们都要调系统调用询问是否有过来数据,我们要是明确的知道哪个连接有数据包过来呢,就不用挨个遍历寻找了。