ZENDER1、NIO基础
1.1、Channel(通道) & Buffer
Channel主要的实现类
Buffer主要的实现类
1.2、Selector(选择器/多路复用器)
阻塞-BIO
-
线程多内存占用高。
-
线程直接上下文切换成本高。
-
只适合连接数少的场景。
非阻塞-NIO
通过Selector去管理多个channel。获取这些channel上发生的事件，这些channel工作在非阻塞模式下，不会让线程绑定在一个channel上。
二、buffer
使用Buffer读写数据一般遵循以下四个步骤：
-
写入数据到Buffer。
-
调用flip()方法。
-
从Buffer中读取数据。
-
调用clear()方法或者compact()方法。
当向buffer写入数据时，buffer会记录下写了多少数据。一旦要读取数据，需要通过flip()方法将Buffer从写模式切换到读模式。在读模式下，可以读取之前写入到buffer的所有数据。
一旦读取完数据，需要清空缓冲区，让它可再次被写入。有两种方式能清空缓冲区：调用clear()或compact()方法。clear()方法会清空整个缓冲区。compact()方法只会清除已经读过的数据。
public static void main(String[] args) {
try (RandomAccessFile file = new RandomAccessFile("G:\\data.txt", "rw")) {
FileChannel channel = file.getChannel();
ByteBuffer buffer = ByteBuffer.allocate(10);
do {
// 读取文件数据，向buffer写入
int len = channel.read(buffer);
System.out.println("读到字节数：" + len);
if (len == -1) {
break;
}
// 切换为读模式，准备去buffer中读取数据
buffer.flip();
while(buffer.hasRemaining()) {
System.out.println((char)buffer.get());
}
// 切换为写模式，同时清理缓冲区，准备下一次写入数据
buffer.clear();
} while (true);
} catch (IOException e) {
e.printStackTrace();
}
}
Buffer有以下重要属性
-
mark=标记位置
-
capacity=容量
-
position=写入位置
-
limit=限制位置
写入4个字节后：position写入位置向前移动了4个位置，limit=capacity保持不变。
flip()方法切换为读取模式后，position回到初始读取位置，limit切换为读取限制。
读取4个字节后：position和limit处于同一个位置。
调用clear()方法后：
compact方法，是把未读完的部分向前压缩，然后切换至写模式。
ByteBuffer常见方法
分配空间
// 使用allocate方法为ByteBuffer分配内存空间，其它buffer类也有该方法 Bytebuffer buf = ByteBuffer.allocate(16);
有两种办法
-
调用 channel 的 read 方法
-
调用 buffer 自己的 put 方法
int readBytes = channel.read(buf); // 或者 buf.put((byte)127);
有两种办法
-
调用 channel 的 write 方法
-
调用 buffer 自己的 get 方法
int writeBytes = channel.write(buf); // 或者 byte b = buf.get();
get 方法会让position读指针向后走，如果想重复读取数据
-
可以调用 rewind 方法将position重新置为0。
-
或者调用get(int i)方法获取索引 i 的内容，它不会移动读指针。
ByteBuffer buffer1 = StandardCharsets.UTF_8.encode("你好");
ByteBuffer buffer2 = Charset.forName("utf-8").encode("你好");
CharBuffer buffer3 = StandardCharsets.UTF_8.decode(buffer1);
System.out.println(buffer3.getClass());
System.out.println(buffer3.toString());
分散读取
try (RandomAccessFile file = new RandomAccessFile("G:\\data.txt", "rw")) {
FileChannel channel = file.getChannel();
ByteBuffer a = ByteBuffer.allocate(3);
ByteBuffer b = ByteBuffer.allocate(3);
ByteBuffer c = ByteBuffer.allocate(5);
channel.read(new ByteBuffer[]{a, b, c});
a.flip();
while(a.hasRemaining()) {
System.out.print((char)a.get());
}
System.out.println();
b.flip();
while(b.hasRemaining()) {
System.out.print((char)b.get());
}
System.out.println();
c.flip();
while(c.hasRemaining()) {
System.out.print((char)c.get());
}
} catch (IOException e) {
e.printStackTrace();
}
分散写入
try (RandomAccessFile file = new RandomAccessFile("G:\\data.txt", "rw")) {
FileChannel channel = file.getChannel();
ByteBuffer a = ByteBuffer.allocate(4);
ByteBuffer b = ByteBuffer.allocate(4);
channel.position(11);
a.put(new byte[]{'1', '2', '3', '4'});
b.put(new byte[]{'5', '6', '7', '8'});
a.flip();
b.flip();
//写入文件
channel.write(new ByteBuffer[]{a, b});
} catch (IOException e) {
e.printStackTrace();
}
黏包半包
NIO是面向缓冲区进行通信的，不是面向流的，既然是缓冲区，那它一定存在一个固定大小。
这样一来通常会遇到两个问题：
-
消息粘包(消息过多)：当缓冲区足够大，由于网络不稳定种种原因，可能会有多条消息从通道读入缓冲区，此时如果无法分清数据包之间的界限，就会导致粘包问题。
-
半包(消息不完整)：若消息没有接收完，缓冲区就被填满了，会导致从缓冲区取出的消息不完整，即半包的现象。
固定头部方案
不如头部设置四个字节，存储一个int值，记录后面数据的长度。以此来标记一个消息体。
读取数据时，根据头部的长度信息，按序读取缓冲区中的数据，若没有达到长度要求，继续读下一个缓冲区。这样自然不会出现粘包、半包问题。
输出数据时，也采用同样的机制封装数据。
三、网络编程
3.1、阻塞/非阻塞
-
阻塞模式下，相关方法都会导致线程暂停
-
ServerSocketChannel.accept()会在没有连接建立时让线程暂停。
-
SocketChannel.read()会在没有数据可读时让线程暂停。
-
阻塞的表现其实就是线程暂停了，暂停期间不会占用 cpu，但线程相当于闲置。
-
单线程下，阻塞方法之间相互影响，几乎不能正常工作，需要多线程支持。
-
但多线程下，如果连接数过多，必然导致OOM，并且线程太多，反而会因为频繁上下文切换导致性能降低。虽然可以采用线程池技术来减少线程数和线程上下文切换，但治标不治本，如果有很多连接建立，但长时间闲置，会阻塞线程池中所有线程，因此不适合长连接，只适合短连接。
Server服务端
public static void main(String[] args) throws IOException {
// 使用nio来理解阻塞模式, 单线程
ByteBuffer buffer = ByteBuffer.allocate(16);
// 1. 创建了服务器
ServerSocketChannel ssc = ServerSocketChannel.open();
// 2. 绑定监听端口
ssc.bind(new InetSocketAddress(8080));
// 3. 连接集合
List<SocketChannel> channels = new ArrayList<>();
while (true) {
// 4. accept 建立与客户端连接， SocketChannel 用来与客户端之间通信
System.out.println("connecting...");
SocketChannel sc = ssc.accept(); // 阻塞方法，线程停止运行
System.out.println("connected..." + sc);
channels.add(sc);
for (SocketChannel channel : channels) {
// 5. 接收客户端发送的数据
System.out.println("before read... {}" + channel);
channel.read(buffer); // 阻塞方法，线程停止运行
buffer.flip();
while(buffer.hasRemaining()) {
System.out.print((char)buffer.get());
}
System.out.println();
buffer.clear();
System.out.println("after read..." + channel);
}
}
}
Channel客户端
public static void main(String[] args) throws IOException {
SocketChannel sc = SocketChannel.open();
sc.connect(new InetSocketAddress("localhost", 8080));
System.out.println("waiting...");
//向服务端发送一条数据
sc.write(Charset.defaultCharset().encode("hello!"));
System.in.read();
}
-
-
在ServerSocketChannel.accept()在没有连接建立时，会返回 null，继续运行。
-
SocketChannel.read()在没有数据可读时，会返回 0，但线程不必阻塞，可以去执行其它SocketChannel.read()或是去执行ServerSocketChannel.accept()。
-
写数据时，线程只是等待数据写入Channel即可，无需等Channel通过网络把数据发送出去。
-
但非阻塞模式下，即使没有连接建立和可读数据，线程仍然在不断运行，消耗cpu资源。
-
Server服务端
public static void main(String[] args) throws IOException {
// 使用 nio 来理解非阻塞模式, 单线程
ByteBuffer buffer = ByteBuffer.allocate(16);
// 1. 创建了服务器
ServerSocketChannel ssc = ServerSocketChannel.open();
ssc.configureBlocking(false); // 非阻塞模式
// 2. 绑定监听端口
ssc.bind(new InetSocketAddress(8080));
// 3. 连接集合
List<SocketChannel> channels = new ArrayList<>();
while (true) {
// 4. accept建立与客户端连接， SocketChannel用来与客户端之间通信
SocketChannel sc = ssc.accept(); // 非阻塞，线程还会继续运行，如果没有连接建立，但sc是null
if (sc != null) {
System.out.println("connected..." + sc);
sc.configureBlocking(false); // 非阻塞模式
channels.add(sc);
}
for (SocketChannel channel : channels) {
// 5. 接收客户端发送的数据
int read = channel.read(buffer);// 非阻塞，线程仍然会继续运行，如果没有读到数据，read返回0
if (read > 0) {
buffer.flip();
while (buffer.hasRemaining()) {
System.out.print((char) buffer.get());
}
buffer.clear();
System.out.println("after read..." + channel);
}
}
}
}
Channel客户端不变
public static void main(String[] args) throws IOException {
SocketChannel sc = SocketChannel.open();
sc.connect(new InetSocketAddress("localhost", 8080));
System.out.println("waiting...");
//向服务端发送一条数据
sc.write(Charset.defaultCharset().encode("hello!"));
System.in.read();
}
多路复用
-
多路复用仅针对网络 IO、普通文件 IO 没法利用多路复用。
-
如果不用Selector的非阻塞模式，线程大部分时间都在做无用功，而Selector能够保证。
-
限于网络传输能力，Channel未必时时可写，一旦Channel可写，会触发Selector的可写事件。
-
有可连接事件时才去连接。
-
有可读事件才去读取。
-
有可写事件才去写入。
3.2、selector监听Channel事件
// 阻塞直到绑定事件发生 int count = selector.select(); // 阻塞直到绑定事件发生，或是超时（时间单位为 ms） int count = selector.select(long timeout); // 不会阻塞，也就是不管有没有事件，立刻返回，自己根据返回值检查是否有事件 int count = selector.selectNow();
-
-
SelectionKey.OP_ACCEPT - 服务器端成功接受连接时触发。
-
SelectionKey.OP_READ - 数据可读入时触发。
-
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);
// sscKey只关注accept事件
sscKey.interestOps(SelectionKey.OP_ACCEPT);
System.out.println("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();
System.out.println("key: " + key);
// 5. 区分事件类型
if (key.isAcceptable()) { // 如果是accept
ServerSocketChannel channel = (ServerSocketChannel) key.channel();
SocketChannel sc = channel.accept();
// 设置为非阻塞模式
sc.configureBlocking(false);
SelectionKey scKey = sc.register(selector, 0, null);
//注册，关联READ事件
scKey.interestOps(SelectionKey.OP_READ);
System.out.println(sc);
System.out.println("scKey:" + scKey);
// 处理key完成后，要从selectedKeys集合中删除，否则下次处理就会有问题。
iter.remove();
} else if (key.isReadable()) { // 如果是read
try {
SocketChannel channel = (SocketChannel) key.channel(); // 拿到触发事件的channel
ByteBuffer buffer = ByteBuffer.allocate(4);
int read = channel.read(buffer); // 如果是正常断开，read的方法的返回值是-1
if(read == -1) {
key.cancel();
} else {
buffer.flip();
System.out.println(Charset.defaultCharset().decode(buffer));
}
} catch (IOException e) {
e.printStackTrace();
key.cancel(); // 因为客户端断开了,因此需要将key取消（从selector的keys集合中真正删除key）
}
}
}
}
}
客户端代码不变，运行结果
3.3、消息边界问题
public static void main(String[] args) throws IOException {
SocketChannel sc = SocketChannel.open();
sc.connect(new InetSocketAddress("localhost", 8080));
System.out.println("waiting...");
sc.write(Charset.defaultCharset().encode("中文"));
}
上面代码客户端发送中文时，服务端接收会出现乱码，由于定义的Buffer的长度不够导致无法解析完所有中文，导致乱码！
-
情况1：Message1消息过大，导致半包。
-
情况2：Message2 Message3消息小，导致半包。
-
情况2：Message4 Message4 Message6消息小，导致粘包。
解决方案
-
固定消息长度，数据包大小一样，服务器按预定长度读取，缺点是浪费带宽。
-
按分隔符拆分，缺点是效率低。
-
TLV格式(即Type类型、Length长度、Value数据)，类型和长度已知的情况下，就可以方便获取消息大小，分配合适的buffer，缺点是buffer需要提前分配，如果内容过大，则影响服务器吞吐量。
服务端代码更改
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);
// sscKey只关注accept事件
sscKey.interestOps(SelectionKey.OP_ACCEPT);
// 绑定端口
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();
// 5. 区分事件类型
if (key.isAcceptable()) { // 如果是accept
ServerSocketChannel channel = (ServerSocketChannel) key.channel();
SocketChannel sc = channel.accept();
// 设置为非阻塞模式
sc.configureBlocking(false);
SelectionKey scKey = sc.register(selector, 0, ByteBuffer.allocate(6));
//注册，关联READ事件
scKey.interestOps(SelectionKey.OP_READ);
// 处理key完成后，要从selectedKeys集合中删除，否则下次处理就会有问题。
iter.remove();
} else if (key.isReadable()) { // 如果是read
try {
SocketChannel channel = (SocketChannel) key.channel(); // 拿到触发事件的channel
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);
// 替换掉原有的Buffer
key.attach(newBuffer);
System.out.println("扩容成功！");
}
}
} catch (IOException e) {
e.printStackTrace();
key.cancel(); // 因为客户端断开了,因此需要将key取消（从selector的keys集合中真正删除key）
}
}
}
}
}
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());
}
// 切换为读模式
target.flip();
System.out.println("完整消息：" + Charset.defaultCharset().decode(target));
}
}
source.compact();
}
客户端代码：
public static void main(String[] args) throws IOException {
SocketChannel sc = SocketChannel.open();
sc.connect(new InetSocketAddress("localhost", 8080));
System.out.println("waiting...");
sc.write(Charset.defaultCharset().encode("中文1\n中文2\n"));
System.in.read();
}
3.4、处理write事件写入内容过多问题
public static void main(String[] args) throws IOException {
ServerSocketChannel ssc = ServerSocketChannel.open();
// 非阻塞
ssc.configureBlocking(false);
Selector selector = Selector.open();
// 设置事件为：服务器端成功接受客户端连接
ssc.register(selector, SelectionKey.OP_ACCEPT);
ssc.bind(new InetSocketAddress(8080));
while (true) {
selector.select();
Iterator<SelectionKey> iter = selector.selectedKeys().iterator();
while (iter.hasNext()) {
SelectionKey key = iter.next();
if (key.isAcceptable()) {
//获取客户端连接的SocketChannel
SocketChannel sc = ssc.accept();
sc.configureBlocking(false);
// 1. 向客户端发送大量数据
StringBuilder sb = new StringBuilder();
for (int i = 0; i < 5000000; i++) {
sb.append("a");
}
ByteBuffer buffer = Charset.defaultCharset().encode(sb.toString());
// 判断是否有剩余字节未写入
while (buffer.hasRemaining()){
// 2. 返回值代表实际写入的字节数
int write = sc.write(buffer);
System.out.println("实际写入的字节数：" + write);
}
}
//处理完，移除事件
iter.remove();
}
}
}
客户端代码
public static void main(String[] args) throws IOException {
SocketChannel sc = SocketChannel.open();
sc.connect(new InetSocketAddress("localhost", 8080));
// 接收数据
int count = 0;
while (true) {
ByteBuffer buffer = ByteBuffer.allocate(1024 * 1024);
count += sc.read(buffer);
System.out.println(count);
buffer.clear();
}
}
服务器这边一次无法写完，解决方案
-
非阻塞模式下，无法保证把buffer中所有数据都写入channel，因此需要追踪write方法的返回值（代表实际写入字节数）。
-
用selector监听所有channel的可写事件，每个channel都需要一个key来跟踪 buffer，但这样又会导致占用内存过多，就有两阶段策略：
-
当消息处理器第一次写入消息时，才将channel注册到selector上。
-
selector检查channel上的可写事件，如果所有的数据写完了，就取消channel的注册。
-
如果不取消，会每次可写均会触发write事件。
优化后的服务端代码
public static void main(String[] args) throws IOException {
ServerSocketChannel ssc = ServerSocketChannel.open();
// 非阻塞
ssc.configureBlocking(false);
Selector selector = Selector.open();
// 设置事件为：服务器端成功接受客户端连接
ssc.register(selector, SelectionKey.OP_ACCEPT);
ssc.bind(new InetSocketAddress(8080));
while (true) {
selector.select();
Iterator<SelectionKey> iter = selector.selectedKeys().iterator();
while (iter.hasNext()) {
SelectionKey key = iter.next();
if (key.isAcceptable()) {
//获取客户端连接的SocketChannel
SocketChannel sc = ssc.accept();
sc.configureBlocking(false);
// 1. 向客户端发送大量数据
StringBuilder sb = new StringBuilder();
for (int i = 0; i < 5000000; i++) {
sb.append("a");
}
ByteBuffer buffer = Charset.defaultCharset().encode(sb.toString());
// 3. 判断是否有剩余内容
if (buffer.hasRemaining()) {
SelectionKey sckey = sc.register(selector, 0, null);
sckey.interestOps(SelectionKey.OP_READ);
// 4. 关注可写事件
sckey.interestOps(sckey.interestOps() + SelectionKey.OP_WRITE);
// 5. 把未写完的数据挂到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);
// 6. 清理操作
if (!buffer.hasRemaining()) {
key.attach(null); // 需要清除buffer
key.interestOps(key.interestOps() - SelectionKey.OP_WRITE);//不需关注可写事件
}
}
//处理完，移除事件
iter.remove();
}
}
}
3.5、利用多线程
版权声明
非特殊说明，本文由Zender原创或收集发布，欢迎转载。
作者文章
- MyBatisCodeHelper-Pro3.3.6+2321破解 1年前 (2024-10-17)
- 网站迁移公告 1年前 (2024-10-10)
- Java项目防止SQL注入4总方式 2年前 (2023-09-06)
- JAVA开发小技巧--读取文件魔数来识别文件类型 2年前 (2023-08-24)
- 分类树菜单优化 2年前 (2023-08-22)
发表评论:
◎欢迎参与讨论，请在这里发表您的看法、交流您的观点。