【Socket与IO框架学习】3. NIO（非阻塞IO）基础知识
快有半个月没继续看这东西了，今天也没啥事情干，就继续记录下学习的NIO这个N代表的是"new"的意思，据我百度所致，NIO是在jdk 1.4的版本中引入的，目的是为了弥补原来的I/O的不足之处，提供了一个更高速的、面向块的I/O。
1. 流与块
standard IO是对流的读写，以流的形式处理数据，每次进行IO操作的时候都要创建一个流对象（如InputStream、OutputStream），流对象进行IO操作都是按字节进行操作，一个一个字节的进行读写操作，因此操作效率比较慢。
而NIO则是将IO抽象成块，以块的形式处理数据，类似硬盘的读写，每次IO操作的单位都是一个块，块存入内存后便是一个byte[]数组，一次可以读写多个字节，因此按块处理数据比按流处理数据要快得多。
jdk1.4的版本中在java.io.*包中已经以NIO的基础重新实现了一遍IO类，所以它可以利用NIO的一些特性。
通道与缓冲区
1. 通道
通道 Channel 是对原 I/O 包中的流的模拟，可以通过它读取和写入数据。
通道与流的不同之处在于，流只能在一个方向上移动(一个流必须是 InputStream 或者 OutputStream 的子类)，而通道是双向的，可以用于读、写或者同时用于读写。
通道包括以下类型:
- FileChannel: 从文件中读写数据；
- DatagramChannel: 通过 UDP 读写网络中数据；
- SocketChannel: 通过 TCP 读写网络中数据；
- ServerSocketChannel: 可以监听新进来的 TCP 连接，对每一个新进来的连接都会创建一个 SocketChannel。
2. 缓冲区
发送给一个通道的所有数据都必须首先放到缓冲区中，同样地，从通道中读取的任何数据都要先读到缓冲区中。也就是说，不会直接对通道进行读写数据，而是要先经过缓冲区。
缓冲区实质上是一个数组，但它不仅仅是一个数组。缓冲区提供了对数据的结构化访问，而且还可以跟踪系统的读/写进程。
缓冲区包括以下类型:
- ByteBuffer
- CharBuffer
- ShortBuffer
- IntBuffer
- LongBuffer
- FloatBuffer
- DoubleBuffer
缓冲区状态变量
- capacity: 最大容量；
- position: 当前已经读写的字节数；
- limit: 还可以读写的字节数。
状态变量的改变过程举例:
① 新建一个大小为 8 个字节的缓冲区，此时 position 为 0，而 limit = capacity = 8。capacity 变量不会改变，下面的讨论会忽略它。
② 从输入通道中读取 5 个字节数据写入缓冲区中，此时 position 移动设置为 5，limit 保持不变。
③ 在将缓冲区的数据写到输出通道之前，需要先调用 flip() 方法，这个方法将 limit 设置为当前 position，并将 position 设置为 0。
④ 从缓冲区中取 4 个字节到输出缓冲中，此时 position 设为 4。
⑤ 最后需要调用 clear() 方法来清空缓冲区，此时 position 和 limit 都被设置为最初位置。
选择器(Selector)
NIO常常被叫做非阻塞IO，主要是因为NIO在网络通信中的非阻塞特性被广泛使用。
NIO实现了IO多路复用中的Reactor模型（这个还不太了解），一个线程Thread使用一个选择器Selector通过轮询的方式去监听多个Channel上的事件，从而让一个线程就可以处理多个事件
通过配置监听的通道Channel为非阻塞，那么当Channel上的IP事件还未到达时，就不会进入阻塞状态一直等待，二十继续轮询其他Channel，找到IP事件已经到达的Channel执行。
因为创建和切换线程的开销很大，因此使用一个线程来处理多个事件具有更好的性能
- 通道事件
将通道注册到选择器上，选择如下具体事件，可以通过"|"将事件组成事件集
如：int interestSet = Selection.OP_READ | Selection.OP_WRITE
public static final int OP_READ = 1 << 0;
public static final int OP_WRITE = 1 << 2;
public static final int OP_CONNECT = 1 << 3;
public static final int OP_ACCEPT = 1 << 4;
demo
NioServer.java
package NioDemo.SocketNioDemo;
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;
/**
* @author Mo
* @createTime 2022/2/1 23:06
* @description
*/
public class NioServer {
public static void main(String[] args) throws IOException {
Selector selector = Selector.open();
ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
// 将通道设置为非阻塞模式
serverSocketChannel.configureBlocking(false);
/**
* 将通道注册到选择器上，选择如下具体事件，可以通过"|"将事件组成事件集
* 如：int interestSet = Selection.OP_READ | Selection.OP_WRITE
* public static final int OP_READ = 1 << 0;
* public static final int OP_WRITE = 1 << 2;
* public static final int OP_CONNECT = 1 << 3;
* public static final int OP_ACCEPT = 1 << 4;
*/
serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
ServerSocket serverSocket = serverSocketChannel.socket();
InetSocketAddress address = new InetSocketAddress("localhost", 80);
serverSocket.bind(address);
while (true) {
// 监听事件,会一直阻塞到监听事件的到达
selector.select();
Set<SelectionKey> keys = selector.selectedKeys();
Iterator<SelectionKey> keyIterator = keys.iterator();
while (keyIterator.hasNext()) {
SelectionKey key = keyIterator.next();
if (key.isAcceptable()) {
ServerSocketChannel serverSocketChannel1 = (ServerSocketChannel) key.channel();
// 服务器会为每个新连接新建一个 SocketChannel, 并注册到选择器上, 事件为读取, 大家可以自己debug看看这个流程
SocketChannel socketChannel = serverSocketChannel1.accept();
socketChannel.configureBlocking(false);
socketChannel.register(selector, SelectionKey.OP_READ);
} else if (key.isReadable()) {
// 上面那个if语块给通道注册的OP_READ事件就会循环到这个代码块中, 进行读操作
SocketChannel socketChannel = (SocketChannel) key.channel();
System.out.println("[address: " + socketChannel.getRemoteAddress() + ", data: " + readDataFromSocketChannel(socketChannel) + "]");
socketChannel.close();
}
keyIterator.remove();
}
}
}
public static String readDataFromSocketChannel (SocketChannel socketChannel) throws IOException {
ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
StringBuilder dataString = new StringBuilder();
while (true) {
if (socketChannel.read(byteBuffer) == -1)
break;
byteBuffer.flip();
int limit = byteBuffer.limit();
char[] dst = new char[limit];
for (int i = 0; i < limit; i ++) {
dst[i] = (char) byteBuffer.get(i);
}
dataString.append(dst);
byteBuffer.clear();
}
return dataString.toString();
}
}
NioClient.java
package NioDemo.SocketNioDemo;
import java.io.IOException;
import java.io.OutputStream;
import java.net.Socket;
/**
* @author Mo
* @createTime 2022/2/3 12:30
* @description
*/
public class NioClient {
public static void main(String[] args) throws IOException {
for (int i = 0; i < 10; i++) {
// 建立socket连接就是一个accept事件的到达
Socket socket = new Socket("localhost", 80);
OutputStream outputStream = socket.getOutputStream();
String data = "hello, world" + i;
// OutputStream进行一个写的操作，对于服务器来说就是一个Read事件的到达
outputStream.write(data.getBytes());
outputStream.close();
}
}
}
建议大家用代码去debug看看背后的运行流程
模拟一个并发，服务端代码一样唯一变化的就是这个客户端的请求代码不一样，大家可以自己写着玩玩
守护线程
package NioDemo.SocketNioDemo;
import java.util.concurrent.CountDownLatch;
/**
* @author Mo
* @createTime 2022/2/3 13:43
* @description
*/
public class NioClientDaemon {
public static void main(String[] args) throws InterruptedException {
Integer clientNumber = 3;
CountDownLatch countDownLatch = new CountDownLatch(clientNumber);
for (int i = 0; i < clientNumber; i++, countDownLatch.countDown()) {
NioRequestThread nioRequestThread = new NioRequestThread(countDownLatch, i);
new Thread(nioRequestThread).start();
}
synchronized (NioClientDaemon.class) {
NioClientDaemon.class.wait();
}
}
}
请求线程
package NioDemo.SocketNioDemo;
import java.io.IOException;
import java.io.OutputStream;
import java.net.Socket;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.CountDownLatch;
/**
* @author Mo
* @createTime 2022/2/3 13:19
* @description
*/
public class NioRequestThread implements Runnable{
private CountDownLatch countDownLatch;
private Integer clientIndex;
public NioRequestThread() {
}
public NioRequestThread(CountDownLatch countDownLatch, Integer clientIndex) {
this.countDownLatch = countDownLatch;
this.clientIndex = clientIndex;
}
@Override
public void run() {
Socket socket = null;
OutputStream outputStream = null;
try {
socket = new Socket("localhost", 80);
outputStream = socket.getOutputStream();
this.countDownLatch.await();
outputStream.write(("这是 " + clientIndex + "号线程发送的消息").getBytes(StandardCharsets.UTF_8));
outputStream.flush();
} catch (IOException | InterruptedException e) {
e.printStackTrace();
} finally {
try {
if (outputStream != null) {
outputStream.close();
}
} catch (IOException e) {
e.printStackTrace();
}
}
}
}