NIO知识总结三
感兴趣的朋友可以去我的语雀平台进行查看更多的知识。
https://www.yuque.com/ambition-bcpii/muziteng
5. Selector
5.1 简介
5.1.1 Selector 和 Channel 关系
Selector 一般称为选择器 ，也可以翻译为多路复用器 。它是 Java NIO 核心组件中的一个，用于检查一个或多个 NIO Channel（通道）
的状态是否处于可读、可写。如此可以实现单线程管理多个 channels,也就是可以管理多个网络链接。
使用 Selector 的好处在于： 使用更少的线程来就可以来处理通道了， 相比使用多个线程，避免了线程上下文切换带来的开销。
5.1.2 可选择通道(SelectableChannel)
不是所有的 Channel 都可以被 Selector 复用的。比方说，FileChannel 就不能被选择器复用。判断一个 Channel 能被 Selector 复用，有
一个前提：判断他是否继承了一个抽象类 SelectableChannel。如果继承了 SelectableChannel，则可以被复用，否则不能。
一个通道可以被注册到多个选择器上，但对每个选择器而言只能被注册一次。通道和选择器之间的关系，使用注册的方式完成。
SelectableChannel 可以被注册到 Selector 对象上，在注册的时候，需要指定通道的哪些操作，是 Selector 感兴趣的。
5.1.3 Channel 注册到 Selector
使用 Channel.register（Selector sel，int ops方法，将一个通道注册到一个选择器时。第一个参数，指定通道要注册的选择器。
第二个参数指定选择器需要查询的通道操作。
可以供选择器查询的通道操作，从类型来分，包括以下四种：
- 可读 :
SelectionKey.OP_READ - 可写 :
SelectionKey.OP_WRITE - 连接 :
SelectionKey.OP_CONNECT - 接收 :
SelectionKey.OP_ACCEPT
如果 Selector 对通道的多操作类型感兴趣，可以用“位或”操作符来实现或者加减运算实现
选择器查询的不是通道的操作，而是通道的某个操作的一种就绪状态。什么是操作的就绪状态？一旦通道具备完成某个操作的条件，表示
该通道的某个操作已经就绪，就可以被 Selector 查询到，程序可以对通道进行对应的操作。比方说，某个 SocketChannel 通道可以连接
到一个服务器，则处于“连接就绪”(OP_CONNECT)。再比方说，一个 ServerSocketChannel 服务器通道准备好接收新进入的连接，则处于
“接收就绪”（OP_ACCEPT）状态。还比方说，一个有数据可读的通道，可以说是“读就绪”(OP_READ)。一个等待写数据的通道可以说是“写
就绪”(OP_WRITE)。
5.1.4 选择键(SelectionKey)
-
Channel 注册到后，并且一旦通道处于某种就绪的状态，就可以被选择器查询到。这个工作，使用选择器 Selector 的 select（）方法
完成。select 方法的作用，对感兴趣的通道操作，进行就绪状态的查询。
-
Selector 可以不断的查询 Channel 中发生的操作的就绪状态。并且挑选感兴趣的操作就绪状态。一旦通道有操作的就绪状态达成，
并且是 Selector 感兴趣的操作，就会被 Selector 选中，放入选择键集合中。
-
一个选择键，首先是包含了注册在 Selector 的通道操作的类型，比方说SelectionKey.OP_READ。也包含了特定的通道与特定的选择
器之间的注册关系。
5.2 使用方法
5.2.1 创建与注册
// 1. 获取 selector
Selector selector = Selector.open();
// 2. 获取通道
ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
// 3. 设置非阻塞
serverSocketChannel.configureBlocking(false);
// 4. 绑定连接
serverSocketChannel.bind(new InetSocketAddress(9999));
// 5. 进行注册 将通道注册到选择器上,并制定监听事件为：“接收”事件
serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
注意：与 Selector 一起使用时，Channel 必须处于非阻塞模式下，否则将抛出异常。
一个通道，并没有一定要支持所有的四种操作。比如服务器通道ServerSocketChannel 支持 Accept 接受操作，而 SocketChannel
客户端通道则不支持。可以通过通道上的 validOps() 方法，来获取特定通道下所有支持的操作集合。
5.2.2 轮询查询就绪操作
-
通过 Selector 的 select（）方法，可以查询出已经就绪的通道操作，这些就绪的状态集合，包存在一个元素是 SelectionKey 对象的
Set 集合中
-
下面是 Selector 几个重载的查询 select()方法：
select():阻塞到至少有一个通道在你注册的事件上就绪了。select(long timeout)：和 select()一样，但最长阻塞事件为 timeout 毫秒。selectNow():非阻塞，只要有通道就绪就立刻返回。
select()方法返回的 int 值，表示有多少通道已经就绪，更准确的说，是自前一次 select方法以来到这一次 select 方法之间的时间段
上，有多少通道变成就绪状态。
一旦调用 select()方法，并且返回值不为 0 时，在 Selector 中有一个 selectedKeys()方法，用来访问已选择键集合，迭代集合的每一个
选择键元素，根据就绪操作的类型，完成对应的操作。
事件发生后，要么处理，要么取消（cancel），不能什么都不做，否则下次该事件仍会触发
Iterator<SelectionKey> iter = selector.selectedKeys().iterator();
while (iter.hasNext()) {
SelectionKey key = iter.next();
if (key.isReadable()) {
// readable
} else if (key.isAcceptable()) {
// acceptable
} else if (key.isWritable()) {
// writable
} else if (key.isConnectable()) {
// connectable
}
iter.remove();
}
5.2.3 停止选择的办法
选择器执行选择的过程，系统底层会依次询问每个通道是否已经就绪，这个过程可能会造成调用线程进入阻塞状态,那么我们有以下三种方
式可以唤醒在 select（）方法中阻塞的线程。
wakeup()方法 ：通过调用 Selector 对象的 wakeup（）方法让处在阻塞状态的 select()方法立刻返回
该方法使得选择器上的第一个还没有返回的选择操作立即返回。如果当前没有进行中的选择操作，那么下一次对 select()方法的一次调用
将立即返回
close()方法 ：通过 close（）方法关闭 Selector。
该方法使得任何一个在选择操作中阻塞的线程都被唤醒（类似 wakeup（）），同时使得注册到该 Selector 的所有 Channel 被注销，所
有的键将被取消，但是 Channel本身并不会关闭。
5.3 NIO 编程步骤
- 创建 Selector 选择器
- 创建 ServerSocketChannel 通道，并绑定监听端口
- 设置 Channel 通道是非阻塞模式
- 把 Channel 注册到 Socketor 选择器上，监听连接事件
- 调用 Selector 的 select 方法（循环调用），监测通道的就绪状况
- 调用 selectKeys 方法获取就绪 channel 集合
- 遍历就绪 channel 集合，判断就绪事件类型，实现具体的业务操作
- 根据业务，决定是否需要再次注册监听事件，重复执行第三步操作
5.4 处理 accept 事件
客户端代码为
public class Client {
public static void main(String[] args) {
try (Socket socket = new Socket("localhost", 8080)) {
System.out.println(socket);
socket.getOutputStream().write("world".getBytes());
System.in.read();
} catch (IOException e) {
e.printStackTrace();
}
}
}
服务器端代码为
@Slf4j
public class ChannelDemo6 {
public static void main(String[] args) {
try (ServerSocketChannel channel = ServerSocketChannel.open()) {
channel.bind(new InetSocketAddress(8080));
Selector selector = Selector.open();
channel.configureBlocking(false);
channel.register(selector, SelectionKey.OP_ACCEPT);
while (true) {
int count = selector.select();
// int count = selector.selectNow();
log.debug("select count: {}", count);
// if(count <= 0) {
// continue;
// }
// 获取所有事件
Set<SelectionKey> keys = selector.selectedKeys();
// 遍历所有事件，逐一处理
Iterator<SelectionKey> iter = keys.iterator();
while (iter.hasNext()) {
SelectionKey key = iter.next();
// 判断事件类型
if (key.isAcceptable()) {
ServerSocketChannel c = (ServerSocketChannel) key.channel();
// 必须处理
SocketChannel sc = c.accept();
log.debug("{}", sc);
}
// 处理完毕，必须将事件移除
iter.remove();
}
}
} catch (IOException e) {
e.printStackTrace();
}
}
}
事件发生后，要么处理，要么取消（cancel），不能什么都不做，否则下次该事件仍会触发，这是因为 nio 底层使用的是水平触发
5.5 处理 read 事件
public static void main(String[] args) throws IOException {
// 1. 创建服务器
ServerSocketChannel ssc = ServerSocketChannel.open();
// 设置非阻塞模式
ssc.configureBlocking(false);
// 2. 创建 Selector 管理多个 channel
Selector selector = Selector.open();
// 3. 建立 selector 与 channel 的联系（注册）
// SelectionKey 就是将来事件发生后，通过它可以知道事件和哪了Channel的事件有联系
SelectionKey sscKey = ssc.register(selector, 0, null);
// key 只关注 accept 事件
sscKey.interestOps(SelectionKey.OP_ACCEPT);
log.debug("register key:{}", sscKey);
ssc.bind(new InetSocketAddress(8080));
while (true) {
// 4. select 方法 没有事件发生，线程阻塞，有事件发生，线程才恢复运行
// select 在事件未处理时，不会阻塞，事件发生后，要么处理，要么取消，不能置之不理
selector.select();
// 5. 处理事件 selectedKeys 内部包含了所有发生的事件
Iterator<SelectionKey> iter = selector.selectedKeys().iterator();
while (iter.hasNext()) {
SelectionKey key = iter.next();
iter.remove(); // 必须将事件进行移除
log.debug("key:{}", key);
// 6. 区分事件类型
if (key.isAcceptable()) {
ServerSocketChannel channel = (ServerSocketChannel) key.channel();
SocketChannel sc = channel.accept();
sc.configureBlocking(false);
SelectionKey scKey = sc.register(selector, 0, null);
scKey.interestOps(SelectionKey.OP_READ);
log.debug("{}", sc);
} else if (key.isReadable()) {
try {
SocketChannel channel = (SocketChannel) key.channel(); // 拿到触发事件的 channel
ByteBuffer byteBuffer = ByteBuffer.allocate(16);
int read = channel.read(byteBuffer); // 如果时正常断开 read 方法的返回值是-1
if (read == -1) {
key.cancel();
} else {
byteBuffer.flip();
debugRead(byteBuffer);
}
} catch (IOException e) {
e.printStackTrace();
key.cancel(); // 客户端断开了，因此需要将key取消（从selector 的 keys 集合中真正删除）
}
}
// key.cancel();
}
}
}
5.5.1 为何要 iter.remove()
因为 select 在事件发生后，就会将相关的 key 放入 selectedKeys 集合，但不会在处理完后从 selectedKeys 集合中移除，需要我们
自己编码删除。例如
- 第一次触发了 ssckey 上的 accept 事件，没有移除 ssckey
- 第二次触发了 sckey 上的 read 事件，但这时 selectedKeys 中还有上次的 ssckey ，在处理时因为没有真正的 serverSocket 连上了，就会导致空指针异常
5.5.2 cancel 的作用
cancel 会取消注册在 selector 上的 channel，并从 keys 集合中删除 key 后续不会再监听事件
5.5.3 ⚠️ 不处理边界的问题
以前有同学写过这样的代码，思考注释中两个问题，以 bio 为例，其实 nio 道理是一样的
public class Server {
public static void main(String[] args) throws IOException {
ServerSocket ss=new ServerSocket(9000);
while (true) {
Socket s = ss.accept();
InputStream in = s.getInputStream();
// 这里这么写，有没有问题
byte[] arr = new byte[4];
while(true) {
int read = in.read(arr);
// 这里这么写，有没有问题
if(read == -1) {
break;
}
System.out.println(new String(arr, 0, read));
}
}
}
}
客户端
public class Client {
public static void main(String[] args) throws IOException {
Socket max = new Socket("localhost", 9000);
OutputStream out = max.getOutputStream();
out.write("hello".getBytes());
out.write("world".getBytes());
out.write("你好".getBytes());
max.close();
}
}
输出
hell
owor
ld�
�好
5.5.4 处理消息的边界
会出现半包、粘包问题
- 一种思路是固定消息长度，数据包大小一样，服务器按预定长度读取，缺点是浪费带宽
- 另一种思路是按分隔符拆分，缺点是效率低
- TLV 格式，即 Type 类型、Length 长度、Value 数据，类型和长度已知的情况下，就可以方便获取消息大小，分配合适的 buffer，缺点是 buffer 需要提前分配，如果内容过大，则影响 server 吞吐量
- Http 1.1 是 TLV 格式
- Http 2.0 是 LTV 格式
服务器端
private static void split(ByteBuffer source) {
source.flip();
for (int i = 0; i < source.limit(); i++) {
// 找到一条完整消息
if (source.get(i) == '\n') {
int length = i + 1 - source.position();
// 把这条完整消息存入新的 ByteBuffer
ByteBuffer target = ByteBuffer.allocate(length);
// 从 source 读，向 target 写
for (int j = 0; j < length; j++) {
target.put(source.get());
}
debugAll(target);
}
}
source.compact(); // 0123456789abcdef position 16 limit 16
}
public static void main(String[] args) throws IOException {
// 1. 创建 selector, 管理多个 channel
Selector selector = Selector.open();
ServerSocketChannel ssc = ServerSocketChannel.open();
ssc.configureBlocking(false);
// 2. 建立 selector 和 channel 的联系（注册）
// SelectionKey 就是将来事件发生后，通过它可以知道事件和哪个channel的事件
SelectionKey sscKey = ssc.register(selector, 0, null);
// key 只关注 accept 事件
sscKey.interestOps(SelectionKey.OP_ACCEPT);
log.debug("sscKey:{}", sscKey);
ssc.bind(new InetSocketAddress(8080));
while (true) {
// 3. select 方法, 没有事件发生，线程阻塞，有事件，线程才会恢复运行
// select 在事件未处理时，它不会阻塞, 事件发生后要么处理，要么取消，不能置之不理
selector.select();
// 4. 处理事件, selectedKeys 内部包含了所有发生的事件
Iterator<SelectionKey> iter = selector.selectedKeys().iterator(); // accept, read
while (iter.hasNext()) {
SelectionKey key = iter.next();
// 处理key 时，要从 selectedKeys 集合中删除，否则下次处理就会有问题
iter.remove();
log.debug("key: {}", key);
// 5. 区分事件类型
if (key.isAcceptable()) { // 如果是 accept
ServerSocketChannel channel = (ServerSocketChannel) key.channel();
SocketChannel sc = channel.accept();
sc.configureBlocking(false);
ByteBuffer buffer = ByteBuffer.allocate(16); // attachment
// 将一个 byteBuffer 作为附件关联到 selectionKey 上
SelectionKey scKey = sc.register(selector, 0, buffer);
scKey.interestOps(SelectionKey.OP_READ);
log.debug("{}", sc);
log.debug("scKey:{}", scKey);
} else if (key.isReadable()) { // 如果是 read
try {
SocketChannel channel = (SocketChannel) key.channel(); // 拿到触发事件的channel
// 获取 selectionKey 上关联的附件
ByteBuffer buffer = (ByteBuffer) key.attachment();
int read = channel.read(buffer); // 如果是正常断开，read 的方法的返回值是 -1
if(read == -1) {
key.cancel();
} else {
split(buffer);
// 需要扩容
if (buffer.position() == buffer.limit()) {
ByteBuffer newBuffer = ByteBuffer.allocate(buffer.capacity() * 2);
buffer.flip();
newBuffer.put(buffer); // 0123456789abcdef3333\n
key.attach(newBuffer);
}
}
} catch (IOException e) {
e.printStackTrace();
key.cancel(); // 因为客户端断开了,因此需要将 key 取消（从 selector 的 keys 集合中真正删除 key）
}
}
}
}
}
客户端
SocketChannel sc = SocketChannel.open();
sc.connect(new InetSocketAddress("localhost", 8080));
SocketAddress address = sc.getLocalAddress();
// sc.write(Charset.defaultCharset().encode("hello\nworld\n"));
sc.write(Charset.defaultCharset().encode("0123\n456789abcdef"));
sc.write(Charset.defaultCharset().encode("0123456789abcdef3333\n"));
System.in.read();
5.5.5 ByteBuffer 大小分配
- 每个 channel 都需要记录可能被切分的消息，因为 ByteBuffer 不能被多个 channel 共同使用，因此需要为每个 channel 维护一个独立的 ByteBuffer
- ByteBuffer 不能太大，比如一个 ByteBuffer 1Mb 的话，要支持百万连接就要 1Tb 内存，因此需要设计大小可变的 ByteBuffer
- 一种思路是首先分配一个较小的 buffer，例如 4k，如果发现数据不够，再分配 8k 的 buffer，将 4k buffer 内容拷贝至 8k buffer，优点是消息连续容易处理，缺点是数据拷贝耗费性能，参考实现 http://tutorials.jenkov.com/java-performance/resizable-array.html
- 另一种思路是用多个数组组成 buffer，一个数组不够，把多出来的内容写入新的数组，与前面的区别是消息存储不连续解析复杂，优点是避免了拷贝引起的性能损耗
5.6 处理 write 事件
5.6.1 一次无法写完例子
- 非阻塞模式下，无法保证把 buffer 中所有数据都写入 channel，因此需要追踪 write 方法的返回值（代表实际写入字节数）
- 用 selector 监听所有 channel 的可写事件，每个 channel 都需要一个 key 来跟踪 buffer，但这样又会导致占用内存过多，就有两阶段策略
- 当消息处理器第一次写入消息时，才将 channel 注册到 selector 上
- selector 检查 channel 上的可写事件，如果所有的数据写完了，就取消 channel 的注册
- 如果不取消，会每次可写均会触发 write 事件
public class WriteServer {
public static void main(String[] args) throws IOException {
ServerSocketChannel ssc = ServerSocketChannel.open();
ssc.configureBlocking(false);
ssc.bind(new InetSocketAddress(8080));
Selector selector = Selector.open();
ssc.register(selector, SelectionKey.OP_ACCEPT);
while(true) {
selector.select();
Iterator<SelectionKey> iter = selector.selectedKeys().iterator();
while (iter.hasNext()) {
SelectionKey key = iter.next();
iter.remove();
if (key.isAcceptable()) {
SocketChannel sc = ssc.accept();
sc.configureBlocking(false);
SelectionKey sckey = sc.register(selector, SelectionKey.OP_READ);
// 1. 向客户端发送内容
StringBuilder sb = new StringBuilder();
for (int i = 0; i < 3000000; i++) {
sb.append("a");
}
ByteBuffer buffer = Charset.defaultCharset().encode(sb.toString());
int write = sc.write(buffer);
// 3. write 表示实际写了多少字节
System.out.println("实际写入字节:" + write);
// 4. 如果有剩余未读字节，才需要关注写事件
if (buffer.hasRemaining()) {
// read 1 write 4
// 在原有关注事件的基础上，多关注 写事件
sckey.interestOps(sckey.interestOps() + SelectionKey.OP_WRITE);
// 把 buffer 作为附件加入 sckey
sckey.attach(buffer);
}
} else if (key.isWritable()) {
ByteBuffer buffer = (ByteBuffer) key.attachment();
SocketChannel sc = (SocketChannel) key.channel();
int write = sc.write(buffer);
System.out.println("实际写入字节:" + write);
if (!buffer.hasRemaining()) { // 写完了
key.interestOps(key.interestOps() - SelectionKey.OP_WRITE);
key.attach(null);
}
}
}
}
}
}
客户端
public class WriteClient {
public static void main(String[] args) throws IOException {
Selector selector = Selector.open();
SocketChannel sc = SocketChannel.open();
sc.configureBlocking(false);
sc.register(selector, SelectionKey.OP_CONNECT | SelectionKey.OP_READ);
sc.connect(new InetSocketAddress("localhost", 8080));
int count = 0;
while (true) {
selector.select();
Iterator<SelectionKey> iter = selector.selectedKeys().iterator();
while (iter.hasNext()) {
SelectionKey key = iter.next();
iter.remove();
if (key.isConnectable()) {
System.out.println(sc.finishConnect());
} else if (key.isReadable()) {
ByteBuffer buffer = ByteBuffer.allocate(1024 * 1024);
count += sc.read(buffer);
buffer.clear();
System.out.println(count);
}
}
}
}
}
5.6.2 write 为何要取消
只要向 channel 发送数据时，socket 缓冲可写，这个事件会频繁触发，因此应当只在 socket 缓冲区写不下时再关注可写事件，数据写完
之后再取消关注
5.7 示例代码
5.7.1 服务器端代码
public static void main(String[] args) {
try {
ServerSocketChannel ssc = ServerSocketChannel.open();
ssc.bind(new InetSocketAddress(8000));
ssc.configureBlocking(false);
Selector selector = Selector.open();
ssc.register(selector, SelectionKey.OP_ACCEPT);
ByteBuffer readBuffer = ByteBuffer.allocate(1024);
ByteBuffer writeBuffer = ByteBuffer.allocate(128);
writeBuffer.put("received".getBytes(StandardCharsets.UTF_8));
writeBuffer.flip();
while (selector.select() > 0) {
Iterator<SelectionKey> iter = selector.selectedKeys().iterator();
while (iter.hasNext()) {
// 获取就绪操作
SelectionKey key = iter.next();
iter.remove();
if (key.isAcceptable()) {
// 创建新的连接，并且把连接注册到 selector 上，而且
// 声明这个 channel 只对读操作感兴趣。
SocketChannel sc = ssc.accept();
sc.configureBlocking(false);
sc.register(selector, SelectionKey.OP_READ);
} else if (key.isReadable()) {
SocketChannel sc = (SocketChannel) key.channel();
// 读取数据
int length = 0;
while ((length = sc.read(readBuffer)) > 0) {
readBuffer.flip();
System.out.println("received:" + new String(readBuffer.array(), 0, length));
readBuffer.clear();
}
key.interestOps(SelectionKey.OP_WRITE);
} else if (key.isWritable()) {
writeBuffer.rewind();
SocketChannel sc = (SocketChannel) key.channel();
sc.write(writeBuffer);
key.interestOps(SelectionKey.OP_READ);
}
}
}
} catch (IOException e) {
throw new RuntimeException(e);
}
}
5.7.2 客户端代码
public static void main(String[] args) {
try {
SocketChannel sc = SocketChannel.open();
sc.connect(new InetSocketAddress("localhost", 8000));
ByteBuffer writeBuffer = ByteBuffer.allocate(32);
ByteBuffer readBuffer = ByteBuffer.allocate(32);
writeBuffer.put("hello".getBytes(StandardCharsets.UTF_8));
writeBuffer.flip();
while (true) {
writeBuffer.rewind();
sc.write(writeBuffer);
readBuffer.clear();
sc.read(readBuffer);
}
} catch (IOException e) {
throw new RuntimeException(e);
}
}