Netty深入浅出笔记
- 1. NIO基础
- 1.1 三大组件
- 1.1.1 Channel & Buffer
- 1.1.2 Selector
- 1.2 ByteBuffer
- 1.2.1 ByteBuffer 正确使用姿势
- 1.2.2 ByteBuffer结构
- 1.2.3 ByteBuffer核心属性
- 1.2.4 ByteBuffer常见方法
- 1.2.5 字符串与 ByteBuffer 互转
- 1.2.6 Scattering Reads(分散读取)
- 1.2.7 Gathering Writes(集中写)
- 1.2.8 粘包与半包
- 1.3 文件编程
- 1.3.1 FileChannel
- 1.3.2 两个 Channel 传输数据
- 1.3.3 Path
- 1.3.4 Files
- 1.4 网络编程
- 1.4.1 阻塞
- 1.4.2 非阻塞
- 1.4.3 多路复用
- 1.4.4 Selector
- 1.4.5 处理 accept事件
- 1.4.6 处理Read事件
- 1.4.7 处理 write 事件
- 1.5 多线程优化
- 1.5.1 利用多线程优化
- 1.6 UDP
- 1.7 NIO vs BIO
- 1.7.1 stream vs channel
- 1.7.2 IO模型
- 1.7.3 零拷贝
- 1.7.4 AIO
1. NIO基础
1.1 三大组件
1.1.1 Channel & Buffer
Java NIO系统的
核心在于：通道(Channel)和缓冲区(Buffer)。通道表示打开到 IO 设备(例如：文件、套接字)的连接。若需要使用 NIO 系统，需要获取用于连接 IO 设备的通道以及用于容纳数据的缓冲区。然后操作缓冲区，对数据进行处理
channel 有一点类似于 stream，它就是读写数据的
双向通道，可以从 channel 将数据读入 buffer，也可以将 buffer 的数据写入 channel，而之前的 stream 要么是输入，要么是输出，channel 比 stream 更为底层
简而言之，通道负责传输，缓冲区负责存储
常见的Channel有以下四种，其中FileChannel主要用于文件传输，其余三种用于网络通信
- FileChannel
- DatagramChannel
- SocketChannel
- ServerSocketChannel
Buffer有以下几种，其中使用较多的是ByteBuffer
ByteBuffer
- MappedByteBuffer
- DirectByteBuffer
- HeapByteBuffer
- ShortBuffer
- IntBuffer
- LongBuffer
- FloatBuffer
- DoubleBuffer
- CharBuffer
1.1.2 Selector
在使用Selector之前，处理socket连接还有以下两种方法
多线程版设计
这种方法存在以下几个问题
- 内存占用高
- 每个线程都需要占用一定的内存，当连接较多时，会开辟大量线程，导致占用大量内存
- 线程上下文切换成本高
- 只适合连接数少的场景
- 连接数过多，会导致创建很多线程，从而出现问题
线程池版设计
这种方法存在以下几个问题
- 阻塞模式下，线程仅能处理一个连接
- 线程池中的线程获取任务（task）后，
只有当其执行完任务之后（断开连接后），才会去获取并执行下一个任务- 若socke连接一直未断开，则其对应的线程无法处理其他socke连接
- 仅适合
短连接场景
- 短连接即建立连接发送请求并响应后就立即断开，使得线程池中的线程可以快速处理其他连接
selector 版设计
selector 的作用就是配合一个线程来管理多个 channel（fileChannel因为是阻塞式的，所以无法使用selector），获取这些 channel 上发生的事件，这些 channel 工作在非阻塞模式下，当一个channel中没有执行任务时，可以去执行其他channel中的任务。适合连接数多，但流量较少的场景
若事件未就绪，调用 selector 的 select() 方法会阻塞线程，直到 channel 发生了就绪事件。这些事件就绪后，select 方法就会返回这些事件交给 thread 来处理
1.2 ByteBuffer
1.2.1 ByteBuffer 正确使用姿势
使用方式
- 向 buffer 写入数据，例如调用 channel.read(buffer)
- 调用 flip() 切换至
读模式
flip会使得buffer中的limit变为position，position变为0- 从 buffer 读取数据，例如调用 buffer.get()
- 调用 clear() 或者compact()切换至
写模式
- 调用clear()方法时
position=0，limit变为capacity- 调用compact()方法时，
会将缓冲区中的未读数据压缩到缓冲区前面- 重复以上步骤
使用案例
有一普通文本文件 data.txt，内容为
1234567890abcd
使用 FileChannel 来读取文件内容
@Slf4j
public class TestByteBuffer {public static void main(String[] args) {//FileChannel//1.输入输出流 2.RandomAccessFiletry (FileChannel channel = new FileInputStream("data.txt").getChannel()){//准备缓存区ByteBuffer byteBuffer = ByteBuffer.allocate(10);while(true){//从channel中读取数据,想buffer写入int len = channel.read(byteBuffer);log.debug("读取到的字节数 {}", len);if(len == -1){ //读到没有数据了break;}//打印buffer的内容byteBuffer.flip(); //切换至读模式while (byteBuffer.hasRemaining()){ //判断是否还有剩余未读的数据byte b = byteBuffer.get();log.debug("实际字节 {}", (char) b);}byteBuffer.clear(); //切换至写模式}}catch (IOException e) {e.printStackTrace();}}
}
输出结果:
17:48:36 [DEBUG] [main] c.c.n.c.TestByteBuffer - 读取到的字节数 10
17:48:36 [DEBUG] [main] c.c.n.c.TestByteBuffer - 实际字节 1
17:48:36 [DEBUG] [main] c.c.n.c.TestByteBuffer - 实际字节 2
17:48:36 [DEBUG] [main] c.c.n.c.TestByteBuffer - 实际字节 3
17:48:36 [DEBUG] [main] c.c.n.c.TestByteBuffer - 实际字节 4
17:48:36 [DEBUG] [main] c.c.n.c.TestByteBuffer - 实际字节 5
17:48:36 [DEBUG] [main] c.c.n.c.TestByteBuffer - 实际字节 6
17:48:36 [DEBUG] [main] c.c.n.c.TestByteBuffer - 实际字节 7
17:48:36 [DEBUG] [main] c.c.n.c.TestByteBuffer - 实际字节 8
17:48:36 [DEBUG] [main] c.c.n.c.TestByteBuffer - 实际字节 9
17:48:36 [DEBUG] [main] c.c.n.c.TestByteBuffer - 实际字节 0
17:48:36 [DEBUG] [main] c.c.n.c.TestByteBuffer - 读取到的字节数 4
17:48:36 [DEBUG] [main] c.c.n.c.TestByteBuffer - 实际字节 a
17:48:36 [DEBUG] [main] c.c.n.c.TestByteBuffer - 实际字节 b
17:48:36 [DEBUG] [main] c.c.n.c.TestByteBuffer - 实际字节 c
17:48:36 [DEBUG] [main] c.c.n.c.TestByteBuffer - 实际字节 d
17:48:36 [DEBUG] [main] c.c.n.c.TestByteBuffer - 读取到的字节数 -1
1.2.2 ByteBuffer结构
ByteBuffer 有以下重要属性
- capacity
- position
- limit
一开始
写模式下，position 是写入位置，limit 等于容量，下图表示写入了 4 个字节后的状态
flip 动作发生后，position 切换为读取位置，limit 切换为读取限制
读取 4 个字节后，状态
clear 动作发生后，状态
compact 方法，是把未读完的部分向前压缩，然后切换至写模式
1.2.3 ByteBuffer核心属性
字节缓冲区的父类Buffer中有几个核心属性，如下
// Invariants: mark <= position <= limit <= capacity
private int mark = -1;
private int position = 0;
private int limit;
private int capacity;
-
capacity：缓冲区的容量。通过构造函数赋予，一旦设置，无法更改
- limit：缓冲区的界限。位于limit 后的数据不可读写。缓冲区的限制不能为负，
并且不能大于其容量- position：
下一个读写位置的索引（类似PC）。缓冲区的位置不能为负，并且不能大于limit- mark：记录当前position的值。
position被改变后，可以通过调用reset() 方法恢复到mark的位置。
以上四个属性必须满足以下要求
mark <= position <= limit <= capacity
1.2.4 ByteBuffer常见方法
allocate()方法
可以使用 allocate 方法为 ByteBuffer 分配空间，其它 buffer 类也有该方法
public class TestByteBufferAllocate {public static void main(String[] args) {System.out.println(ByteBuffer.allocate(16).getClass());System.out.println(ByteBuffer.allocateDirect(16).getClass());/*class java.nio.HeapByteBuffer - java 堆内存，读写效率较低，受到 GC 的影响class java.nio.DirectByteBuffer - 直接内存，读写效率高（少一次拷贝），不会受 GC 影响，分配的效率低*/}
}
put()方法
- put()方法可以将一个数据放入到缓冲区中。
- 进行该操作后，postition的值会+1，指向下一个可以放入的位置。capacity = limit ，为缓冲区容量的值。
flip()方法
- flip()方法会
切换对缓冲区的操作模式，由写->读 / 读->写- 进行该操作后
- 如果是写模式->读模式，position = 0 ， limit 指向最后一个元素的下一个位置，capacity不变
- 如果是读->写，则恢复为put()方法中的值
get()方法
- get()方法会读取缓冲区中的一个值
- 进行该操作后，position会+1，如果超过了limit则会抛出异常
注意：get(i)方法不会改变position的值
rewind()方法
- 该方法只能在
读模式下使用- rewind()方法后，会恢复position、limit和capacity的值，变为进行get()前的值
clean()方法
- clean()方法会将缓冲区中的各个属性恢复为最初的状态，position = 0, capacity = limit
- 此时缓冲区的数据依然存在，处于“被遗忘”状态，下次进行写操作时会覆盖这些数据
mark()和reset()方法
- mark()方法会将postion的值保存到mark属性中
- reset()方法会将position的值改为mark中保存的值
compact()方法
此方法为ByteBuffer的方法，而不是Buffer的方法
- compact会把未读完的数据向前压缩，然后切换到写模式
- 数据前移后，原位置的值并未清零，写时会
覆盖之前的值
clear() VS compact()
- clear只是对position、limit、mark进行重置，而compact在对position进行设置，以及limit、mark进行重置的同时，还涉及到数据在内存中拷贝（会调用arraycopy）。
所以compact比clear更耗性能。但compact能保存你未读取的数据，将新数据追加到为读取的数据之后；而clear则不行，若你调用了clear，则未读取的数据就无法再读取到了
所以需要根据情况来判断使用哪种方法进行模式切换
ByteBuffer常用方法示例
public class TestByteBuffer {public static void main(String[] args) {ByteBuffer buffer = ByteBuffer.allocate(10);// 向buffer中写入1个字节的数据buffer.put((byte)97);// 使用工具类，查看buffer状态ByteBufferUtil.debugAll(buffer);// 向buffer中写入4个字节的数据buffer.put(new byte[]{98, 99, 100, 101});ByteBufferUtil.debugAll(buffer);// 获取数据buffer.flip();ByteBufferUtil.debugAll(buffer);System.out.println(buffer.get());System.out.println(buffer.get());ByteBufferUtil.debugAll(buffer);// 使用compact切换模式buffer.compact();ByteBufferUtil.debugAll(buffer);// 再次写入buffer.put((byte)102);buffer.put((byte)103);ByteBufferUtil.debugAll(buffer);}
}
输出结果:
// 向缓冲区写入了一个字节的数据，此时postition为1
+--------+-------------------- all ------------------------+----------------+
position: [1], limit: [10]+-------------------------------------------------+| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 61 00 00 00 00 00 00 00 00 00 |a......... |
+--------+-------------------------------------------------+----------------+// 向缓冲区写入四个字节的数据，此时position为5
+--------+-------------------- all ------------------------+----------------+
position: [5], limit: [10]+-------------------------------------------------+| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 61 62 63 64 65 00 00 00 00 00 |abcde..... |
+--------+-------------------------------------------------+----------------+// 调用flip切换模式，此时position为0，表示从第0个数据开始读取
+--------+-------------------- all ------------------------+----------------+
position: [0], limit: [5]+-------------------------------------------------+| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 61 62 63 64 65 00 00 00 00 00 |abcde..... |
+--------+-------------------------------------------------+----------------+
// 读取两个字节的数据
97
98// position变为2
+--------+-------------------- all ------------------------+----------------+
position: [2], limit: [5]+-------------------------------------------------+| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 61 62 63 64 65 00 00 00 00 00 |abcde..... |
+--------+-------------------------------------------------+----------------+// 调用compact切换模式，此时position及其后面的数据被压缩到ByteBuffer前面去了
// 此时position为3，会覆盖之前的数据
+--------+-------------------- all ------------------------+----------------+
position: [3], limit: [10]+-------------------------------------------------+| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 63 64 65 64 65 00 00 00 00 00 |cdede..... |
+--------+-------------------------------------------------+----------------+// 再次写入两个字节的数据，之前的 0x64 0x65 被覆盖
+--------+-------------------- all ------------------------+----------------+
position: [5], limit: [10]+-------------------------------------------------+| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 63 64 65 66 67 00 00 00 00 00 |cdefg..... |
+--------+-------------------------------------------------+----------------+
public class TestByteBufferRead {public static void main(String[] args) {ByteBuffer buffer = ByteBuffer.allocate(10);buffer.put(new byte[]{'a', 'b', 'c', 'd'});buffer.flip();// rewind 从头开始读/*buffer.get(new byte[4]);debugAll(buffer);buffer.rewind();System.out.println((char)buffer.get());*/// mark & reset// mark 做一个标记，记录 position 位置， reset 是将 position 重置到 mark 的位置/*System.out.println((char) buffer.get());System.out.println((char) buffer.get());buffer.mark(); // 加标记，索引2 的位置System.out.println((char) buffer.get());System.out.println((char) buffer.get());buffer.reset(); // 将 position 重置到索引 2System.out.println((char) buffer.get());System.out.println((char) buffer.get());*/// get(i) 不会改变读索引的位置System.out.println((char) buffer.get(3));ByteBufferUtil.debugAll(buffer);}
}
public class TestByteBufferReadWrite {public static void main(String[] args) {ByteBuffer buffer = ByteBuffer.allocate(10);buffer.put((byte) 0x61); // 'a'ByteBufferUtil.debugAll(buffer);buffer.put(new byte[]{0x62, 0x63, 0x64}); // b c dByteBufferUtil.debugAll(buffer);
// System.out.println(buffer.get());buffer.flip();System.out.println(buffer.get());ByteBufferUtil.debugAll(buffer);buffer.compact();ByteBufferUtil.debugAll(buffer);buffer.put(new byte[]{0x65, 0x6f});ByteBufferUtil.debugAll(buffer);}
}1.2.5 字符串与 ByteBuffer 互转
方法一
- 编码：字符串调用getByte方法获得byte数组，将byte数组放入ByteBuffer中
- 解码：先调用ByteBuffer的flip方法，然后通过StandardCharsets的decoder方法解码
public class Translate {public static void main(String[] args) {// 准备两个字符串String str1 = "hello";String str2 = "";ByteBuffer buffer1 = ByteBuffer.allocate(16);// 通过字符串的getByte方法获得字节数组，放入缓冲区中buffer1.put(str1.getBytes());ByteBufferUtil.debugAll(buffer1);// 将缓冲区中的数据转化为字符串// 切换模式buffer1.flip();// 通过StandardCharsets解码，获得CharBuffer，再通过toString获得字符串str2 = StandardCharsets.UTF_8.decode(buffer1).toString();System.out.println(str2);ByteBufferUtil.debugAll(buffer1);}
}
方法二
- 编码：通过StandardCharsets的encode方法获得ByteBuffer，此时获得的ByteBuffer为读模式，无需通过flip切换模式
- 解码：通过StandardCharsets的decoder方法解码
public class Translate {public static void main(String[] args) {// 准备两个字符串String str1 = "hello";String str2 = "";// 通过StandardCharsets的encode方法获得ByteBuffer// 此时获得的ByteBuffer为读模式，无需通过flip切换模式ByteBuffer buffer1 = StandardCharsets.UTF_8.encode(str1);ByteBufferUtil.debugAll(buffer1);// 将缓冲区中的数据转化为字符串// 通过StandardCharsets解码，获得CharBuffer，再通过toString获得字符串str2 = StandardCharsets.UTF_8.decode(buffer1).toString();System.out.println(str2);ByteBufferUtil.debugAll(buffer1);}
}
方法三
- 编码：字符串调用getByte()方法获得字节数组，将字节数组传给ByteBuffer的wrap()方法，通过该方法获得ByteBuffer。同样无需调用flip方法切换为读模式
- 解码：通过StandardCharsets的decoder方法解码
public class Translate {public static void main(String[] args) {// 准备两个字符串String str1 = "hello";String str2 = "";// 通过StandardCharsets的encode方法获得ByteBuffer// 此时获得的ByteBuffer为读模式，无需通过flip切换模式ByteBuffer buffer1 = ByteBuffer.wrap(str1.getBytes());ByteBufferUtil.debugAll(buffer1);// 将缓冲区中的数据转化为字符串// 通过StandardCharsets解码，获得CharBuffer，再通过toString获得字符串str2 = StandardCharsets.UTF_8.decode(buffer1).toString();System.out.println(str2);ByteBufferUtil.debugAll(buffer1);}
}
1.2.6 Scattering Reads(分散读取)
分散读取，有一个文本文件 words.txt
onetwothree
使用如下方式读取，可以将数据填充至多个 buffer
public class TestScatteringReads {public static void main(String[] args) {try (FileChannel channel = new RandomAccessFile("words.txt", "r").getChannel()) {ByteBuffer b1 = ByteBuffer.allocate(3);ByteBuffer b2 = ByteBuffer.allocate(3);ByteBuffer b3 = ByteBuffer.allocate(5);channel.read(new ByteBuffer[]{b1, b2, b3});b1.flip();b2.flip();b3.flip();ByteBufferUtil.debugAll(b1);ByteBufferUtil.debugAll(b2);ByteBufferUtil.debugAll(b3);} catch (IOException e) {}}
}
输出结果:
+-------------------------------------------------+| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 6f 6e 65 |one |
+--------+-------------------------------------------------+----------------++-------------------------------------------------+| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 74 77 6f |two |
+--------+-------------------------------------------------+----------------++-------------------------------------------------+| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 74 68 72 65 65 |three |
+--------+-------------------------------------------------+----------------+
1.2.7 Gathering Writes(集中写)
使用如下方式写入，可以将多个 buffer 的数据填充至 channel
public class TestGatheringWrites {public static void main(String[] args) {ByteBuffer b1 = StandardCharsets.UTF_8.encode("hello");ByteBuffer b2 = StandardCharsets.UTF_8.encode("world");ByteBuffer b3 = StandardCharsets.UTF_8.encode("你好");try (FileChannel channel = new RandomAccessFile("words2.txt", "rw").getChannel()) {channel.write(new ByteBuffer[]{b1, b2, b3});} catch (IOException e) {}}
}
1.2.8 粘包与半包
- 网络上有多条数据发送给服务端，数据之间使用 \n 进行分隔
- 但由于某种原因这些数据在接收时，被进行了重新组合，例如原始数据有3条为
- Hello,world\n
- I’m Nyima\n
- How are you?\n
变成了下面的两个 byteBuffer (粘包，半包)
- Hello,world\nI’m Nyima\nHo
- w are you?\n
现在要求你编写程序，将错乱的数据恢复成原始的按 \n 分隔的数据
出现原因
粘包:发送方在发送数据时，并不是一条一条地发送数据，而是将数据整合在一起，当数据达到一定的数量后再一起发送。这就会导致多条信息被放在一个缓冲区中被一起发送出去半包:接收方的缓冲区的大小是有限的，当接收方的缓冲区满了以后，就需要将信息截断，等缓冲区空了以后再继续放入数据。这就会发生一段完整的数据最后被截断的现象
解决办法
- 通过get(index)方法遍历ByteBuffer，遇到分隔符时进行处理。
注意：get(index)不会改变position的值
- 记录该段数据长度，以便于申请对应大小的缓冲区
- 将缓冲区的数据通过get()方法写入到target中
- 调用
compact方法切换模式，因为缓冲区中可能还有未读的数据
网络上有多条数据发送给服务端，数据之间使用 \n 进行分隔
但由于某种原因这些数据在接收时，被进行了重新组合，例如原始数据有3条为
public class TestByteBufferExam {public static void main(String[] args) {/*网络上有多条数据发送给服务端，数据之间使用 \n 进行分隔但由于某种原因这些数据在接收时，被进行了重新组合，例如原始数据有3条为Hello,world\nI'm zhangsan\nHow are you?\n变成了下面的两个 byteBuffer (黏包，半包)Hello,world\nI'm zhangsan\nHow are you?\n现在要求你编写程序，将错乱的数据恢复成原始的按 \n 分隔的数据*/ByteBuffer source = ByteBuffer.allocate(32);source.put("Hello,world\nI'm zhangsan\nHo".getBytes());split(source);source.put("w are you?\n".getBytes());split(source);}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();// 把这条完整消息存入新的 ByteBufferByteBuffer target = ByteBuffer.allocate(length);// 从 source 读，向 target 写for (int j = 0; j < length; j++) {target.put(source.get());}ByteBufferUtil.debugAll(target);}}source.compact();}
}
1.3 文件编程
1.3.1 FileChannel
工作模式
FileChannel只能在阻塞模式下工作，所以无法搭配Selector
获取
不能直接打开 FileChannel，必须通过 FileInputStream、FileOutputStream 或者 RandomAccessFile 来获取
FileChannel，它们都有 getChannel 方法
- 通过 FileInputStream 获取的 channel
只能读- 通过 FileOutputStream 获取的 channel
只能写- 通过 RandomAccessFile 是否能读写
根据构造 RandomAccessFile 时的读写模式决定
读取
通过 FileInputStream 获取channel，通过read方法将数据写入到ByteBuffer中
read方法的返回值表示读到了多少字节，若读到了文件末尾则返回-1
int readBytes = channel.read(buffer);
可根据返回值判断是否读取完毕
while(channel.read(buffer) > 0) {// 进行对应操作...
}
写入
因为channel也是有大小的，所以 write 方法并不能保证一次将 buffer 中的内容全部写入 channel。必须需要按照以下规则进行写入
ByteBuffer buffer = ...;
buffer.put(...); // 存入数据
buffer.flip(); // 切换读模式
// 通过hasRemaining()方法查看缓冲区中是否还有数据未写入到通道中
while(buffer.hasRemaining()) {channel.write(buffer);
}
关闭
通道需要close，一般情况通过try-with-resource进行关闭，最好使用以下方法获取strea以及channel，避免某些原因使得资源未被关闭，channel 必须关闭，不过调用了 FileInputStream、FileOutputStream 或者 RandomAccessFile 的 close 方法会间接地调用 channel 的 close 方法
位置
获取当前位置
channel也拥有一个保存读取数据位置的属性，即position
//获取位置
long pos = channel.position();
设置当前位置
long newPos = ...;
channel.position(newPos);
设置当前位置时，如果设置为文件的末尾
- 这时读取会返回 -1
- 这时写入，会追加内容，但要注意如果 position 超过了文件末尾，再写入时在新内容和原末尾之间会有空洞（00）
大小
使用 size 方法获取文件的大小
强制写入
操作系统出于性能的考虑，会将数据缓存，不是立刻写入磁盘。可以调用 force(true) 方法将文件内容和元数据（文件的权限等信息）立刻写入磁盘
1.3.2 两个 Channel 传输数据
transferTo方法
使用transferTo方法可以快速、高效地将一个channel中的数据传输到另一个channel中，
但一次只能传输2G的内容
public class TestFileChannelTransferTo01 {public static void main(String[] args) {try (FileChannel from = new FileInputStream("from.txt").getChannel();FileChannel to = new FileOutputStream("to.txt").getChannel();){// 效率高，底层会利用操作系统的零拷贝进行优化, 2g 数据long size = from.size();// left 变量代表还剩余多少字节for(long left = size;left > 0;){System.out.println("position"+(size - left)+", left:"+left);left -= from.transferTo((size - left), left, to);}}catch (IOException e) {e.printStackTrace();}}
}
当传输的文件大于2G时，需要使用以下方法进行多次传输
public class TestFileChannelTransferTo02{public static void main(String[] args){try (FileInputStream fis = new FileInputStream("from.txt");FileOutputStream fos = new FileOutputStream("to.txt");FileChannel inputChannel = fis.getChannel();FileChannel outputChannel = fos.getChannel()) {long size = inputChannel.size();long capacity = inputChannel.size();// 分多次传输while (capacity > 0) {// transferTo返回值为传输了的字节数capacity -= inputChannel.transferTo(size-capacity, capacity, outputChannel);}} catch (IOException e) {e.printStackTrace();}}
}
1.3.3 Path
jdk7 引入了 Path 和 Paths 类
- Path 用来表示文件路径
- Paths 是工具类，用来获取 Path 实例
Path source = Paths.get("1.txt"); // 相对路径 使用 user.dir 环境变量来定位 1.txtPath source = Paths.get("d:\\1.txt"); // 绝对路径 代表了 d:\1.txtPath source = Paths.get("d:/1.txt"); // 绝对路径 同样代表了 d:\1.txtPath projects = Paths.get("d:\\data", "projects"); // 代表了 d:\data\projects
.代表了当前路径..代表了上一级路径
例如目录结构如下
d:|- data|- projects|- a|- b
代码
Path path = Paths.get("d:\\data\\projects\\a\\..\\b");
System.out.println(path);
System.out.println(path.normalize()); // 正常化路径
会输出
d:\data\projects\a\..\b
d:\data\projects\b
1.3.4 Files
检查文件是否存在
Path path = Paths.get("helloword/data.txt");
System.out.println(Files.exists(path));
创建一级目录
Path path = Paths.get("helloword/d1");
Files.createDirectory(path);
- 如果目录已存在，会抛异常 FileAlreadyExistsException
- 不能一次创建多级目录，否则会抛异常 NoSuchFileException
创建多级目录
Path path = Paths.get("helloword/d1/d2");
Files.createDirectories(path);
拷贝文件
Path source = Paths.get("helloword/data.txt");
Path target = Paths.get("helloword/target.txt");
Files.copy(source, target);
如果文件已存在，会抛异常 FileAlreadyExistsException
如果希望用 source覆盖掉 target，需要用 StandardCopyOption 来控制
Files.copy(source, target, StandardCopyOption.REPLACE_EXISTING);
public class TestFilesCopy {public static void main(String[] args) throws IOException {long start = System.currentTimeMillis();String source = "D:\\tmp";String target = "D:\\tmptmp";Files.walk(Paths.get(source)).forEach(path ->{try {//将当前的路径替换成目标路径String targetName = path.toString().replace(source,target);//是目录if(Files.isDirectory(path)){Files.createDirectories(Paths.get(targetName));}else if(Files.isRegularFile(path)){ //如果是普通文件Files.copy(path,Paths.get(targetName));}}catch (IOException e) {e.printStackTrace();}});long end = System.currentTimeMillis();System.out.println(end - start);}
}移动文件
Path source = Paths.get("helloword/data.txt");
Path target = Paths.get("helloword/data.txt");Files.move(source, target, StandardCopyOption.ATOMIC_MOVE);
StandardCopyOption.ATOMIC_MOVE 保证文件移动的原子性
删除文件
Path target = Paths.get("helloword/target.txt");Files.delete(target);
删除目录
Path target = Paths.get("helloword/d1");Files.delete(target);
如果目录还有内容，会抛异常 DirectoryNotEmptyException
遍历目录文件
可以使用Files工具类中的walkFileTree(Path, FileVisitor)方法，其中需要传入两个参数
- Path：文件起始路径
- FileVisitor：文件访问器，使用访问者模式
- 接口的实现类SimpleFileVisitor有四个方法
- preVisitDirectory：访问目录前的操作
- visitFile：访问文件的操作
- visitFileFailed：访问文件失败时的操作
- postVisitDirectory：访问目录后的操作
//遍历文件目录
public static void method01() throws IOException {AtomicInteger dirCount = new AtomicInteger(); //目录数AtomicInteger fileCount = new AtomicInteger();//文件数Files.walkFileTree(Paths.get("D:\\apache-tomcat-7.0.0"),new SimpleFileVisitor<Path>(){//在目录中的条目之后调用目录，并且已经访问了所有的后代。@Overridepublic FileVisitResult preVisitDirectory(Path dir, BasicFileAttributes attrs) throws IOException {System.out.println("===>"+dir);dirCount.incrementAndGet();return super.preVisitDirectory(dir, attrs);}//为目录中的文件调用。@Overridepublic FileVisitResult visitFile(Path file, BasicFileAttributes attrs) throws IOException {System.out.println(file);fileCount.incrementAndGet();return super.visitFile(file, attrs);}});System.out.println("dir count:"+dirCount);System.out.println("file count"+fileCount);}
}
统计jar数量
//统计jar包数量
public static void method02() throws IOException {AtomicInteger jarCount = new AtomicInteger();Files.walkFileTree(Paths.get("D:\\apache-tomcat-7.0.0"),new SimpleFileVisitor<Path>(){@Overridepublic FileVisitResult visitFile(Path file, BasicFileAttributes attrs) throws IOException {if(file.toString().endsWith(".jar")){System.out.println(file);jarCount.incrementAndGet();}return super.visitFile(file, attrs);}});System.out.println("jarCount:"+jarCount);
}
删除多级目录
//删除多级目录
public static void method03() throws IOException {Files.walkFileTree(Paths.get("D:\\tmp"),new SimpleFileVisitor<Path>(){@Overridepublic FileVisitResult visitFile(Path file, BasicFileAttributes attrs) throws IOException {Files.delete(file);return super.visitFile(file, attrs);}@Overridepublic FileVisitResult postVisitDirectory(Path dir, IOException exc) throws IOException {Files.delete(dir);return super.postVisitDirectory(dir, exc);}});
}
拷贝多级目录
//拷贝多级目录
public static void method04() throws IOException {long start = System.currentTimeMillis();String source = "D:\\apache-tomcat-7.0.0";String target = "D:\\apache-tomcat-7.0.0copy";Files.walk(Paths.get(source)).forEach(path ->{try {String targetName = path.toString().replace(source,target);//是目录if(Files.isDirectory(path)){Files.createDirectories(Paths.get(targetName));}else if(Files.isRegularFile(path)){Files.copy(path,Paths.get(targetName));}} catch (IOException e) {e.printStackTrace();}});long end = System.currentTimeMillis();System.out.println(end - start);
}
1.4 网络编程
1.4.1 阻塞
- 阻塞模式下，相关方法都会导致线程暂停
- ServerSocketChannel.accept 会在
没有连接建立时让线程暂停- SocketChannel.read 会在
通道中没有数据可读时让线程暂停- 阻塞的表现其实就是线程暂停了，暂停期间不会占用 cpu，但线程相当于闲置
- 单线程下，阻塞方法之间相互影响，几乎不能正常工作，需要多线程支持
- 但多线程下，有新的问题，体现在以下方面
- 32 位 jvm 一个线程 320k，64 位 jvm 一个线程 1024k，如果连接数过多，必然导致 OOM，并且线程太多，反而会因为频繁上下文切换导致性能降低
- 可以采用线程池技术来减少线程数和线程上下文切换，但治标不治本，如果有很多连接建立，但长时间 inactive，会阻塞线程池中所有线程，因此不适合长连接，只适合短连接
服务端代码
public class Server {public static void main(String[] args) {//创建缓冲区ByteBuffer byteBuffer = ByteBuffer.allocate(16);//获取服务器通道try (ServerSocketChannel ssc = ServerSocketChannel.open()){//为通道绑定端口ssc.bind(new InetSocketAddress(8085));//用户存放连接的集合ArrayList<SocketChannel> channels = new ArrayList<>();//循环接收连接while (true){System.out.println("before connecting...");//没有连接时,会阻塞线程,accept 建立与客户端连接， SocketChannel 用来与客户端之间通信SocketChannel socketChannel = ssc.accept(); //阻塞方法,线程停止运行System.out.println("after connecting...");channels.add(socketChannel); //将连接存放到集合中//循环遍历集合中的连接for(SocketChannel channel : channels){System.out.println("before reading...");//处理通道中的数据,如果数据可读时,会阻塞线程channel.read(byteBuffer);byteBuffer.flip(); //切换读写模式ByteBufferUtil.debugRead(byteBuffer); //打印数据byteBuffer.clear();System.out.println("after reading...");}}} catch (IOException e) {e.printStackTrace();}}
}
客户端代码
public class Client {public static void main(String[] args) {try (SocketChannel socketChannel = SocketChannel.open()){//建立连接socketChannel.connect(new InetSocketAddress("localhost",8085));//客户端写入数据//socketChannel.write(Charset.defaultCharset().encode("hello nio..."));System.out.println("waiting...");} catch (IOException e) {e.printStackTrace();}}
}
1.4.2 非阻塞
- 非阻塞模式下，相关方法都会不会让线程暂停
- 在 ServerSocketChannel.accept 在没有连接建立时，会返回 null，继续运行
- SocketChannel.read 在没有数据可读时，会返回 0，但线程不必阻塞，可以去执行其它 SocketChannel 的 read 或是去执行 ServerSocketChannel.accept
- 写数据时，线程只是等待数据写入 Channel 即可，无需等 Channel 通过网络把数据发送出去
- 但非阻塞模式下，即使没有连接建立，和可读数据，线程仍然在不断运行，白白浪费了 cpu
- 数据复制过程中，线程实际还是阻塞的（AIO 改进的地方）
- 可以通过
ServerSocketChannel的configureBlocking(false)方法将获得连接设置为非阻塞的。此时若没有连接，accept会返回null- 可以通过
SocketChannel的configureBlocking(false)方法将从通道中读取数据设置为非阻塞的。若此时通道中没有数据可读，read会返回-1
服务端代码如下,客户端保持上面不变
public class Server02 {public static void main(String[] args) {//创建缓冲区ByteBuffer byteBuffer = ByteBuffer.allocate(16);//获得服务器通道try (ServerSocketChannel ssc = ServerSocketChannel.open()){//为服务器通道绑定端口ssc.bind(new InetSocketAddress(8085));//用户存放连接的集合List<SocketChannel> channels = new ArrayList<>();//循环接收连接while (true){//设置为非阻塞模式,没有连接时返回null,不会阻塞线程ssc.configureBlocking(false);//非阻塞，线程还会继续运行，如果没有连接建立，但sc是nullSocketChannel socketChannel = ssc.accept();//通道不为空时才会将连接放入到集合中if(socketChannel != null){System.out.println("after connecting...");channels.add(socketChannel);}//循环遍历结婚中的连接for (SocketChannel channel : channels) {//处理通道的数据,设置为非阻塞模式,若通道没有数据,会返回0,不会阻塞线程socketChannel.configureBlocking(false); //非阻塞模式// 非阻塞，线程仍然会继续运行，如果没有读到数据，read 返回 0int read = channel.read(byteBuffer);if(read > 0){byteBuffer.flip(); //切换读写模式ByteBufferUtil.debugRead(byteBuffer); //打印数据byteBuffer.clear();System.out.println("after reading...");}}}} catch (IOException e) {e.printStackTrace();}}
}
这样写存在一个问题，因为设置为了非阻塞，会一直执行while(true)中的代码，CPU一直处于忙碌状态，会使得性能变低，所以实际情况中不使用这种方法处理请求
1.4.3 多路复用
单线程可以配合 Selector 完成对多个 Channel 可读写事件的监控，这称之为多路复用
- 多路复用
仅针对网络 IO、普通文件 IO没法利用多路复用- 如果不用 Selector 的非阻塞模式，线程大部分时间都在做无用功，而 Selector 能够保证
- 有可连接事件时才去连接
- 有可读事件才去读取
- 有可写事件才去写入
- 限于网络传输能力，Channel 未必时时可写，一旦 Channel 可写，会触发 Selector 的可写事件
1.4.4 Selector
好处
- 一个线程配合 selector 就可以监控多个 channel 的事件，事件发生线程才去处理。避免非阻塞模式下所做无用功
- 让这个线程能够被充分利用
- 节约了线程的数量
- 减少了线程上下文切换
创建
Selector selector = Selector.open();
绑定 Channel 事件
也称之为注册事件，绑定的事件 selector 才会关心
channel.configureBlocking(false);
SelectionKey key = channel.register(selector, 绑定事件);
- channel 必须工作在非阻塞模式
- FileChannel 没有非阻塞模式，因此不能配合 selector 一起使用
- 绑定的事件类型可以有
- connect - 客户端连接成功时触发
- accept - 服务器端成功接受连接时触发
- read - 数据可读入时触发，有因为接收能力弱，数据暂不能读入的情况
- write - 数据可写出时触发，有因为发送能力弱，数据暂不能写出的情况
监听 Channel 事件
可以通过下面三种方法来监听是否有事件发生，方法的返回值代表有多少 channel 发生了事件
- 方法1，阻塞直到绑定事件发生
int count = selector.select();
- 方法2，阻塞直到绑定事件发生，或是超时（时间单位为 ms）
int count = selector.select(long timeout);
- 方法3，不会阻塞，也就是不管有没有事件，立刻返回，自己根据返回值检查是否有事件
int count = selector.selectNow();
select 何时不阻塞
- 事件发生时
- 客户端发起连接请求，会触发 accept 事件
- 客户端发送数据过来，客户端正常、异常关闭时，都会触发 read 事件，另外如果发送的数据大于 buffer 缓冲区，会触发多次读取事件
- channel 可写，会触发 write 事件
- 在 linux 下 nio bug 发生时
- 调用 selector.wakeup()
- 调用 selector.close()
- selector 所在线程 interrupt
1.4.5 处理 accept事件
要使用Selector实现多路复用，服务端代码如下改进
public class SelectServer {public static void main(String[] args) {ByteBuffer byteBuffer = ByteBuffer.allocate(16);//获得服务器通道try (ServerSocketChannel ssc = ServerSocketChannel.open()){ssc.bind(new InetSocketAddress(8080));//创建选择器Selector selector = Selector.open();//通道必须设置成非阻塞模式ssc.configureBlocking(false);//将通道注册到选择器中,并设置感兴趣的事件ssc.register(selector, SelectionKey.OP_ACCEPT);while (true){//若没有事件就绪,线程会被阻塞,反之不会阻塞,从而避免了CPU空转//返回值为就绪的事件个数int ready = selector.select();//int ready2 = selector.selectNow();System.out.println("selector ready counts:"+ready);//获取所有事件Set<SelectionKey> selectionKeys = selector.selectedKeys();//使用迭代器遍历事件Iterator<SelectionKey> iterator = selectionKeys.iterator();while (iterator.hasNext()){SelectionKey key = iterator.next();//判断key的类型if(key.isAcceptable()){//获得key对应的channelServerSocketChannel channel =(ServerSocketChannel) key.channel();System.out.println("before accepting...");//获取连接并处理,而且是必须处理,否则需要取消SocketChannel socketChannel = channel.accept();System.out.println("after accepting...");//处理完成后移除iterator.remove();}}}} catch (IOException e) {e.printStackTrace();}}
}
客户代码如下:
public class SelectClient {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();}}
}
代码步骤解析
- 获得选择器Selector
Selector selector = Selector.open();
- 将
通道设置为非阻塞模式，并注册到选择器中，并设置感兴趣的事件
- channel 必须工作在非阻塞模式
- FileChannel 没有非阻塞模式，因此不能配合 selector 一起使用
- 绑定的
事件类型可以有
- connect - 客户端连接成功时触发
- accept - 服务器端成功接受连接时触发
- read - 数据可读入时触发，有因为接收能力弱，数据暂不能读入的情况
- write - 数据可写出时触发，有因为发送能力弱，数据暂不能写出的情况
// 通道必须设置为非阻塞模式
server.configureBlocking(false);
// 将通道注册到选择器中，并设置感兴趣的实践
server.register(selector, SelectionKey.OP_ACCEPT);
- 通过Selector监听事件，并获得就绪的通道个数，若没有通道就绪，线程会被阻塞
阻塞直到绑定事件发生
int count = selector.select();
阻塞直到绑定事件发生，或是超时（时间单位为 ms）
int count = selector.select(long timeout);
不会阻塞，也就是不管有没有事件，立刻返回，自己根据返回值检查是否有事件
int count = selector.selectNow();
获取就绪事件并得到对应的通道，然后进行处理
// 获取所有事件
Set<SelectionKey> selectionKeys = selector.selectedKeys();// 使用迭代器遍历事件
Iterator<SelectionKey> iterator = selectionKeys.iterator();while (iterator.hasNext()) {SelectionKey key = iterator.next();// 判断key的类型，此处为Accept类型if(key.isAcceptable()) {// 获得key对应的channelServerSocketChannel channel = (ServerSocketChannel) key.channel();// 获取连接并处理，而且是必须处理，否则需要取消SocketChannel socketChannel = channel.accept();// 处理完毕后移除iterator.remove();}
}
事件发生后能否不处理
事件发生后，要么处理，要么取消（cancel），不能什么都不做，否则下次该事件仍会触发，这是因为 nio 底层使用的是水平触发
1.4.6 处理Read事件
public class SelectServer03 {public static void main(String[] args){ByteBuffer byteBuffer = ByteBuffer.allocate(16);//获取服务器通道try (ServerSocketChannel ssc = ServerSocketChannel.open()){ssc.bind(new InetSocketAddress(8080));//创建选择器Selector selector = Selector.open();//通道必须设置为非阻塞ssc.configureBlocking(false);//将通道注册到选择器上,并设置感兴趣的附件ssc.register(selector, SelectionKey.OP_ACCEPT);//为serverKey设置感兴趣的实践while(true){//若没有事件就绪,线程会被阻塞,反之不会阻塞,从而避免CPU空转//返回值为就绪的事件个数int ready = selector.select();System.out.println("selector ready counts:"+ready);//获取所有事件Set<SelectionKey> selectionKeys = selector.selectedKeys();//使用迭代器遍历事件Iterator<SelectionKey> iterator = selectionKeys.iterator();while (iterator.hasNext()){SelectionKey key = iterator.next();//判断key的类型if(key.isAcceptable()){//获取key对应的类型ServerSocketChannel channel = (ServerSocketChannel )key.channel();System.out.println("before accepting...");//获取连接SocketChannel socketChannel = channel.accept();System.out.println("after accepting...");//设置为非阻塞模式,同时将连接的通道也注册到选择当中socketChannel.configureBlocking(false);socketChannel.register(selector,SelectionKey.OP_READ);//处理完毕后移除iterator.remove();}else if(key.isReadable()){SocketChannel channel = (SocketChannel) key.channel();System.out.println("before reading...");channel.read(byteBuffer);System.out.println("after reading...");byteBuffer.flip();ByteBufferUtil.debugRead(byteBuffer);byteBuffer.clear();//处理完后移除iterator.remove();}}}}catch (IOException e){e.printStackTrace();}}
}
为何要 iter.remove()
因为 select 在事件发生后，就会将相关的 key 放入 selectedKeys 集合，但不会在处理完后从 selectedKeys 集合中移除，需要我们自己编码删除。例如
- 第一次触发了 ssckey 上的 accept 事件，没有移除 ssckey
- 第二次触发了 sckey 上的 read 事件，但这时 selectedKeys 中还有上次的 ssckey ，在处理时因为没有真正的 serverSocket 连上了，就会导致空指针异常
cancel 的作用
cancel 会取消注册在 selector 上的 channel，并从 keys 集合中删除 key 后续不会再监听事件
断开处理
当客户端与服务器之间的连接断开时，会给服务器端发送一个读事件，对异常断开和正常断开需要加以不同的方式进行处理
- 正常断开
正常断开时，服务器端的channel.read(buffer)方法的返回值为-1，所以当结束到返回值为-1时，需要调用key的cancel方法取消此事件，并在取消后移除该事件
int read = channel.read(buffer);
// 断开连接时，客户端会向服务器发送一个写事件，此时read的返回值为-1
if(read == -1) {// 取消该事件的处理key.cancel();channel.close();
} else {...
}
// 取消或者处理，都需要移除key
iterator.remove();
- 异常断开
异常断开时，会抛出IOException异常， 在try-catch的catch块中捕获异常并调用key的cancel方法即可
不处理边界的问题
以前有同学写过这样的代码，思考注释中两个问题，以 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�
�好
这是因为UTF-8字符集下，1个汉字占用3个字节，此时缓冲区大小为4个字节，一次读时间无法处理完通道中的所有数据，所以一共会触发两次读事件。这就导致 你好 的 好 字被拆分为了前半部分和后半部分发送，解码时就会出现问题
处理消息的边界
传输的文本可能有以下三种情况
- 文本大于缓冲区大小
- 此时需要将缓冲区进行扩容
- 发生半包现象
- 发生粘包现象
解决思路大致有以下三种
- 固定消息长度，数据包大小一样，服务器按预定长度读取，当发送的数据较少时，需要将数据进行填充，直到长度与消息规定长度一致。缺点是浪费带宽
- 另一种思路是按分隔符拆分，缺点是效率低，需要一个一个字符地去匹配分隔符
- TLV 格式，即 Type 类型、Length 长度、Value 数据（也就是在消息开头用一些空间存放后面数据的长度），如HTTP请求头中的Content-Type与Content-Length。类型和长度已知的情况下，就可以方便获取消息大小，分配合适的 buffer，缺点是 buffer 需要提前分配，如果内容过大，则影响 server 吞吐量
- Http 1.1 是 TLV 格式
- Http 2.0 是 LTV 格式
下文的消息边界处理方式为第二种：按分隔符拆分
附件与扩容
Channel的register方法还有
第三个参数：附件，可以向其中放入一个Object类型的对象，该对象会与登记的Channel以及其对应的SelectionKey绑定，可以从SelectionKey获取到对应通道的附件
public final SelectionKey register(Selector sel, int ops, Object att)
可通过SelectionKey的attachment()方法获得附件
ByteBuffer buffer = (ByteBuffer) key.attachment();
我们需要在Accept事件发生后，将通道注册到Selector中时，对
每个通道添加一个ByteBuffer附件，让每个通道发生读事件时都使用自己的通道，避免与其他通道发生冲突而导致问题
// 设置为非阻塞模式，同时将连接的通道也注册到选择其中，同时设置附件
socketChannel.configureBlocking(false);
ByteBuffer buffer = ByteBuffer.allocate(16);
// 添加通道对应的Buffer附件
socketChannel.register(selector, SelectionKey.OP_READ, buffer);
当Channel中的数据大于缓冲区时，需要对缓冲区进行
扩容操作。此代码中的扩容的判定方法：Channel调用compact方法后，position与limit相等，说明缓冲区中的数据并未被读取（容量太小），此时创建新的缓冲区，其大小扩大为两倍。同时还要将旧缓冲区中的数据拷贝到新的缓冲区中，同时调用SelectionKey的attach方法将新的缓冲区作为新的附件放入SelectionKey中
// 如果缓冲区太小，就进行扩容
if (buffer.position() == buffer.limit()) {ByteBuffer newBuffer = ByteBuffer.allocate(buffer.capacity()*2);// 将旧buffer中的内容放入新的buffer中ewBuffer.put(buffer);// 将新buffer作为附件放到key中key.attach(newBuffer);
}
改造后的服务器代码如下
@Slf4j
public class SelectServer04 {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();// 把这条完整消息存入新的 ByteBufferByteBuffer target = ByteBuffer.allocate(length);// 从 source 读，向 target 写for (int j = 0; j < length; j++) {target.put(source.get());}ByteBufferUtil.debugAll(target);}}source.compact(); // 0123456789abcdef position 16 limit 16}public static void main(String[] args) throws IOException {// 1. 创建 selector, 管理多个 channelSelector 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, readwhile (iter.hasNext()) {SelectionKey key = iter.next();// 处理key 时，要从 selectedKeys 集合中删除，否则下次处理就会有问题iter.remove();log.debug("key: {}", key);// 5. 区分事件类型if (key.isAcceptable()) { // 如果是 acceptServerSocketChannel 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()) { // 如果是 readtry {SocketChannel channel = (SocketChannel) key.channel(); // 拿到触发事件的channel// 获取 selectionKey 上关联的附件ByteBuffer buffer = (ByteBuffer) key.attachment();int read = channel.read(buffer); // 如果是正常断开，read 的方法的返回值是 -1if(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\nkey.attach(newBuffer);}}} catch (IOException e) {e.printStackTrace();key.cancel(); // 因为客户端断开了,因此需要将 key 取消（从 selector 的 keys 集合中真正删除 key）}}}}}
}客户端:
public class SelectClient04 {public static void main(String[] args) {try (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();} catch (IOException e) {e.printStackTrace();}}
}
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，一个数组不够，把多出来的内容写入新的数组，与前面的区别是消息存储不连续解析复杂，优点是避免了拷贝引起的性能损耗
1.4.7 处理 write 事件
一次无法写完例子
- 非阻塞模式下，无法保证把 buffer 中所有数据都写入 channel，因此需要追踪 write 方法的返回值（代表实际写入字节数）
- 用 selector 监听所有 channel 的可写事件，每个 channel 都需要一个 key 来跟踪 buffer，但这样又会导致占用内存过多，就有两阶段策略
- 当消息处理器第一次写入消息时，才将 channel 注册到 selector 上
- selector 检查 channel 上的可写事件，如果所有的数据写完了，就取消 channel 的注册
- 如果不取消，会每次可写均会触发 write 事件
服务器通过Buffer向通道中写入数据时，可能因为通道容量小于Buffer中的数据大小，导致无法一次性将Buffer中的数据全部写入到Channel中，这时便需要分多次写入，具体步骤如下
- 执行一次写操作，向将buffer中的内容写入到SocketChannel中，然后判断Buffer中是否还有数据
- 若Buffer中还有数据，则需要将SockerChannel注册到Seletor中，并关注写事件，同时将未写完的Buffer作为附件一起放入到SelectionKey中
int write = socket.write(buffer);
// 通道中可能无法放入缓冲区中的所有数据
if (buffer.hasRemaining()) {// 注册到Selector中，关注可写事件，并将buffer添加到key的附件中socket.configureBlocking(false);socket.register(selector, SelectionKey.OP_WRITE, buffer);
}
- 添加写事件的相关操作key.isWritable()，对Buffer再次进行写操作
- 每次写后需要判断Buffer中是否还有数据（是否写完）。若写完，需要移除SelecionKey中的Buffer附件，避免其占用过多内存，同时还需移除对写事件的关注
SocketChannel socket = (SocketChannel) key.channel();
// 获得buffer
ByteBuffer buffer = (ByteBuffer) key.attachment();
// 执行写操作
int write = socket.write(buffer);
System.out.println(write);
// 如果已经完成了写操作，需要移除key中的附件，同时不再对写事件感兴趣
if (!buffer.hasRemaining()) {key.attach(null);key.interestOps(0);
}
write 为何要取消
- 只要向 channel 发送数据时，socket 缓冲可写，这个事件会频繁触发，因此应当只在 socket 缓冲区写不下时再关注可写事件，数据写完之后再取消关注
1.5 多线程优化
1.5.1 利用多线程优化
现在都是多核 cpu，设计时要充分考虑别让 cpu 的力量被白白浪费
前面的代码只有一个选择器，没有充分利用多核 cpu，如何改进呢？
- 分两组选择器
- 单线程配一个选择器，专门处理 accept 事件
- 创建 cpu 核心数的线程，每个线程配一个选择器，轮流处理 read 事件
实现思路
创建一个负责处理Accept事件的Boss线程，与多个负责处理Read事件的Worker线程
Boss线程执行的操作
- 接受并处理Accepet事件，当Accept事件发生后，调用Worker的register(SocketChannel socket)方法，让Worker去处理Read事件，其中
需要根据标识robin去判断将任务分配给哪个Worker
// 创建固定数量的Worker
Worker[] workers = new Worker[4];
// 用于负载均衡的原子整数
AtomicInteger robin = new AtomicInteger(0);
// 负载均衡，轮询分配Worker
workers[robin.getAndIncrement()% workers.length].register(socket);
- register(SocketChannel socket)方法会
通过同步队列完成Boss线程与Worker线程之间的通信，让SocketChannel的注册任务被Worker线程执行。添加任务后需要调用selector.wakeup()来唤醒被阻塞的Selector
public void register(final SocketChannel socket) throws IOException {// 只启动一次if (!started) {// 初始化操作}// 向同步队列中添加SocketChannel的注册事件// 在Worker线程中执行注册事件queue.add(new Runnable() {@Overridepublic void run() {try {socket.register(selector, SelectionKey.OP_READ);} catch (IOException e) {e.printStackTrace();}}});// 唤醒被阻塞的Selector// select类似LockSupport中的park，wakeup的原理类似LockSupport中的unparkselector.wakeup();
}
Worker线程执行的操作
- 从同步队列中获取注册任务，并处理Read事件
实现代码
@Slf4j
public class ThreadServer {public static void main(String[] args) throws IOException {Thread.currentThread().setName("boss");ServerSocketChannel ssc = ServerSocketChannel.open();ssc.configureBlocking(false);Selector boss = Selector.open();SelectionKey bossKey = ssc.register(boss, 0, null);bossKey.interestOps(SelectionKey.OP_ACCEPT);ssc.bind(new InetSocketAddress(8080));// 1. 创建固定数量的 worker 并初始化Worker[] workers = new Worker[Runtime.getRuntime().availableProcessors()];for (int i = 0; i < workers.length; i++) {workers[i] = new Worker("worker-" + i);}AtomicInteger index = new AtomicInteger();while(true) {boss.select();Iterator<SelectionKey> iter = boss.selectedKeys().iterator();while (iter.hasNext()) {SelectionKey key = iter.next();iter.remove();if (key.isAcceptable()) {SocketChannel sc = ssc.accept();sc.configureBlocking(false);log.debug("connected...{}", sc.getRemoteAddress());// 2. 关联 selectorlog.debug("before register...{}", sc.getRemoteAddress());// round robin 轮询workers[index.getAndIncrement() % workers.length].register(sc); // boss 调用 初始化 selector , 启动 worker-0log.debug("after register...{}", sc.getRemoteAddress());}}}}static class Worker implements Runnable{private Thread thread;private Selector selector;private String name;private volatile boolean start = false; // 还未初始化private ConcurrentLinkedQueue<Runnable> queue = new ConcurrentLinkedQueue<>();public Worker(String name) {this.name = name;}// 初始化线程，和 selectorpublic void register(SocketChannel sc) throws IOException {if(!start) {selector = Selector.open();thread = new Thread(this, name);thread.start();start = true;}selector.wakeup(); // 唤醒 select 方法 bosssc.register(selector, SelectionKey.OP_READ, null); // boss}@Overridepublic void run() {while(true) {try {selector.select(); // worker-0 阻塞Iterator<SelectionKey> iter = selector.selectedKeys().iterator();while (iter.hasNext()) {SelectionKey key = iter.next();iter.remove();if (key.isReadable()) {ByteBuffer buffer = ByteBuffer.allocate(16);SocketChannel channel = (SocketChannel) key.channel();log.debug("read...{}", channel.getRemoteAddress());channel.read(buffer);buffer.flip();}}} catch (IOException e) {e.printStackTrace();}}}}
}
如何拿到 cpu 个数
- Runtime.getRuntime().availableProcessors() 如果工作在 docker 容器下，因为容器不是物理隔离的，会拿到物理 cpu 个数，而不是容器申请时的个数
- 这个问题直到 jdk 10 才修复，使用 jvm 参数 UseContainerSupport 配置， 默认开启
1.6 UDP
- UDP 是无连接的，client 发送数据不会管 server 是否开启
- server 这边的 receive 方法会将接收到的数据存入 byte buffer，但如果数据报文超过 buffer 大小，多出来的数据会被默默抛弃
首先启动服务器端
public class UdpServer {public static void main(String[] args) {try (DatagramChannel channel = DatagramChannel.open()) {channel.socket().bind(new InetSocketAddress(9999));System.out.println("waiting...");ByteBuffer buffer = ByteBuffer.allocate(32);channel.receive(buffer);buffer.flip();debug(buffer);} catch (IOException e) {e.printStackTrace();}}
}
输出结果:
waiting...
运行客户端
public class UdpClient {public static void main(String[] args) {try (DatagramChannel channel = DatagramChannel.open()) {ByteBuffer buffer = StandardCharsets.UTF_8.encode("hello");InetSocketAddress address = new InetSocketAddress("localhost", 9999);channel.send(buffer, address);} catch (Exception e) {e.printStackTrace();}}
}
接下来服务器端输出
+-------------------------------------------------+| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 68 65 6c 6c 6f |hello |
+--------+-------------------------------------------------+----------------+
1.7 NIO vs BIO
1.7.1 stream vs channel
- stream 不会自动缓冲数据，channel 会利用系统提供的发送缓冲区、接收缓冲区（更为底层）
- stream 仅支持阻塞 API，channel 同时支持阻塞、非阻塞 API，网络 channel 可配合 selector 实现多路复用
- 二者均为全双工，即读写可以同时进行
- 虽然Stream是单向流动的，但是它也是全双工的
1.7.2 IO模型
同步阻塞、同步非阻塞、同步多路复用、异步阻塞（没有此情况）、异步非阻塞
同步：线程自己去获取结果（一个线程）
- 例如：线程调用一个方法后，需要等待方法返回结果
异步：线程自己不去获取结果，而是由其它线程返回结果（至少两个线程）
- 例如：线程A调用一个方法后，继续向下运行，运行结果由线程B返回
当调用一次
channel.read或stream.read后，会切换至操作系统内核态来完成真正数据读取，而读取又分为两个阶段，分别为：
- 等待数据阶段
- 复制数据阶段
阻塞IO
用户线程进行read操作时，
需要等待操作系统执行实际的read操作，此期间用户线程是被阻塞的，无法执行其他操作
非阻塞IO
- 用户线程
在一个循环中一直调用read方法，若内核空间中还没有数据可读，立即返回
只是在等待阶段非阻塞- 用户线程发现内核空间中有数据后，等待内核空间执行复制数据，待复制结束后返回结果
多路复用
Java中通过Selector实现多路复用
- 当没有事件是，调用select方法会被阻塞住
- 一旦有一个或多个事件发生后，就会处理对应的事件，从而实现多路复用
多路复用与阻塞IO的区别
- 阻塞IO模式下，
若线程因accept事件被阻塞，发生read事件后，仍需等待accept事件执行完成后，才能去处理read事件- 多路复用模式下，一个事件发生后，若另一个事件处于阻塞状态，不会影响该事件的执行
异步 IO
- 线程1调用方法后立即返回，
不会被阻塞也不需要立即获取结果- 当方法的运行结果出来以后，由线程2将结果返回给线程1
**阻塞IO VS 多路复用
1.7.3 零拷贝
零拷贝指的是数据无需拷贝到 JVM 内存中，同时具有以下三个优点
- 更少的用户态与内核态的切换
- 不利用 cpu 计算，减少 cpu 缓存伪共享
- 零拷贝适合小文件传输
传统 IO 问题
传统的 IO 将一个文件通过 socket 写出
File f = new File("helloword/data.txt");
RandomAccessFile file = new RandomAccessFile(file, "r");byte[] buf = new byte[(int)f.length()];
file.read(buf);Socket socket = ...;
socket.getOutputStream().write(buf);
内部工作流程是这样的：
- Java 本身并不具备 IO 读写能力，因此 read 方法调用后，要从 Java 程序的
用户态切换至内核态，去调用操作系统（Kernel）的读能力，将数据读入内核缓冲区。这期间用户线程阻塞，操作系统使用 DMA（Direct Memory Access）来实现文件读，其间也不会使用 CPU
DMA 也可以理解为硬件单元，用来解放 cpu 完成文件 IO- 从
内核态切换回用户态，将数据从内核缓冲区读入用户缓冲区（即 byte[] buf），这期间CPU 会参与拷贝，无法利用 DMA- 调用 write 方法，这时将数据从
用户缓冲区（byte[] buf）写入socket 缓冲区，CPU 会参与拷贝- 接下来要向网卡写数据，这项能力 Java 又不具备，因此又得从
用户态切换至内核态，调用操作系统的写能力，使用 DMA 将socket 缓冲区的数据写入网卡，不会使用 CPU
可以看到中间环节较多，java 的 IO 实际不是物理设备级别的读写，而是缓存的复制，底层的真正读写是操作系统来完成的
- 用户态与内核态的切换发生了 3 次，这个操作比较重量级
- 数据拷贝了共 4 次
NIO 优化
通过 ''DirectByteBuf`
- ByteBuffer.allocate(10)
- 底层对应 HeapByteBuffer，使用的还是 Java 内存
- ByteBuffer.allocateDirect(10)
- 底层对应DirectByteBuffer，使用的是
操作系统内存
大部分步骤与优化前相同，唯有一点：
Java 可以使用 DirectByteBuffer 将堆外内存映射到 JVM 内存中来直接访问使用
- 这块内存不受 JVM 垃圾回收的影响，因此内存地址固定，有助于 IO 读写
- Java 中的 DirectByteBuf 对象仅维护了此内存的虚引用，内存回收分成两步
- DirectByteBuffer 对象被垃圾回收，将虚引用加入引用队列
- 当引用的对象ByteBuffer被垃圾回收以后，虚引用对象Cleaner就会被放入引用队列中，然后调用Cleaner的clean方法来释放直接内存
- DirectByteBuffer 的释放底层调用的是 Unsafe 的 freeMemory 方法
- 通过专门线程访问引用队列，根据虚引用释放堆外内存
减少了一次数据拷贝，用户态与内核态的切换次数没有减少
进一步优化
以下两种方式都是
零拷贝，即无需将数据拷贝到用户缓冲区中（JVM内存中）
底层采用了 linux 2.1 后提供的 sendFile 方法，Java 中对应着两个 channel 调用 transferTo/transferFrom 方法拷贝数据
- Java 调用 transferTo 方法后，要从 Java 程序的用户态切换至内核态，使用 DMA将数据读入内核缓冲区，不会使用 CPU
- 数据从内核缓冲区传输到 socket 缓冲区，CPU 会参与拷贝
- 最后使用 DMA 将 socket 缓冲区的数据写入网卡，不会使用 CPU
- 这种方法下
- 只发生了1次用户态与内核态的切换
- 数据拷贝了 3 次
进一步优化2
linux 2.4 对上述方法再次进行了优化
- java 调用 transferTo 方法后，要从 java 程序的用户态切换至内核态，使用 DMA将数据读入内核缓冲区，不会使用 cpu
- 只会将一些 offset 和 length 信息拷入 socket 缓冲区，几乎无消耗
- 使用 DMA 将 内核缓冲区的数据写入网卡，不会使用 cpu
- 整个过程仅只发生了一次用户态与内核态的切换，数据拷贝了 2 次。所谓的【零拷贝】，并不是真正无拷贝，而是在不会拷贝重复数据到 jvm 内存中，零拷贝的优点有
- 更少的用户态与内核态的切换
- 不利用 cpu 计算，减少 cpu 缓存伪共享
- 零拷贝适合小文件传输
1.7.4 AIO
AIO 用来解决数据复制阶段的阻塞问题
- 同步意味着，在进行读写操作时，线程需要等待结果，还是相当于闲置
- 异步意味着，在进行读写操作时，线程不必等待结果，而是将来由操作系统来通过回调方式由另外的线程来获得结果
异步模型需要底层操作系统（Kernel）提供支持
- Windows 系统通过 IOCP
实现了真正的异步 IO- Linux 系统异步 IO 在 2.6 版本引入，但其
底层实现还是用多路复用模拟了异步 IO，性能没有优势
文件IO
先来看看 AsynchronousFileChannel
@Slf4j
public class AioDemo1 {public static void main(String[] args) throws IOException {try{AsynchronousFileChannel s = AsynchronousFileChannel.open(Paths.get("1.txt"), StandardOpenOption.READ);ByteBuffer buffer = ByteBuffer.allocate(2);log.debug("begin...");s.read(buffer, 0, null, new CompletionHandler<Integer, ByteBuffer>() {@Overridepublic void completed(Integer result, ByteBuffer attachment) {log.debug("read completed...{}", result);buffer.flip();debug(buffer);}@Overridepublic void failed(Throwable exc, ByteBuffer attachment) {log.debug("read failed...");}});} catch (IOException e) {e.printStackTrace();}log.debug("do other things...");System.in.read();}
}
输出结果:
13:44:56 [DEBUG] [main] c.i.aio.AioDemo1 - begin...
13:44:56 [DEBUG] [main] c.i.aio.AioDemo1 - do other things...
13:44:56 [DEBUG] [Thread-5] c.i.aio.AioDemo1 - read completed...2+-------------------------------------------------+| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 61 0d |a. |
+--------+-------------------------------------------------+----------------+
可以看到
- 响应文件读取成功的是另一个线程 Thread-5
- 主线程并没有 IO 操作阻塞
守护线程
默认文件 AIO 使用的线程都是守护线程，所以最后要执行
System.in.read()以避免守护线程意外结束
网络 AIO
public class AioServer {public static void main(String[] args) throws IOException {AsynchronousServerSocketChannel ssc = AsynchronousServerSocketChannel.open();ssc.bind(new InetSocketAddress(8080));ssc.accept(null, new AcceptHandler(ssc));System.in.read();}private static void closeChannel(AsynchronousSocketChannel sc) {try {System.out.printf("[%s] %s close\n", Thread.currentThread().getName(), sc.getRemoteAddress());sc.close();} catch (IOException e) {e.printStackTrace();}}private static class ReadHandler implements CompletionHandler<Integer, ByteBuffer> {private final AsynchronousSocketChannel sc;public ReadHandler(AsynchronousSocketChannel sc) {this.sc = sc;}@Overridepublic void completed(Integer result, ByteBuffer attachment) {try {if (result == -1) {closeChannel(sc);return;}System.out.printf("[%s] %s read\n", Thread.currentThread().getName(), sc.getRemoteAddress());attachment.flip();System.out.println(Charset.defaultCharset().decode(attachment));attachment.clear();// 处理完第一个 read 时，需要再次调用 read 方法来处理下一个 read 事件sc.read(attachment, attachment, this);} catch (IOException e) {e.printStackTrace();}}@Overridepublic void failed(Throwable exc, ByteBuffer attachment) {closeChannel(sc);exc.printStackTrace();}}private static class WriteHandler implements CompletionHandler<Integer, ByteBuffer> {private final AsynchronousSocketChannel sc;private WriteHandler(AsynchronousSocketChannel sc) {this.sc = sc;}@Overridepublic void completed(Integer result, ByteBuffer attachment) {// 如果作为附件的 buffer 还有内容，需要再次 write 写出剩余内容if (attachment.hasRemaining()) {sc.write(attachment);}}@Overridepublic void failed(Throwable exc, ByteBuffer attachment) {exc.printStackTrace();closeChannel(sc);}}private static class AcceptHandler implements CompletionHandler<AsynchronousSocketChannel, Object> {private final AsynchronousServerSocketChannel ssc;public AcceptHandler(AsynchronousServerSocketChannel ssc) {this.ssc = ssc;}@Overridepublic void completed(AsynchronousSocketChannel sc, Object attachment) {try {System.out.printf("[%s] %s connected\n", Thread.currentThread().getName(), sc.getRemoteAddress());} catch (IOException e) {e.printStackTrace();}ByteBuffer buffer = ByteBuffer.allocate(16);// 读事件由 ReadHandler 处理sc.read(buffer, buffer, new ReadHandler(sc));// 写事件由 WriteHandler 处理sc.write(Charset.defaultCharset().encode("server hello!"), ByteBuffer.allocate(16), new WriteHandler(sc));// 处理完第一个 accpet 时，需要再次调用 accept 方法来处理下一个 accept 事件ssc.accept(null, this);}@Overridepublic void failed(Throwable exc, Object attachment) {exc.printStackTrace();}}
}
