NIO又叫New/Non-blocking IO,这个概念基本人人都听过,但是不一定每个人都懂他它的运行的原理。
这里我们来探讨这个问题,先用一个例子解释一下BIO到底阻塞了哪里。
/**
* 这是一个单线程BIOServer
* @author endless
* @create 2020-03-23
*/
public class BioServerDemo {
public static void main(String[] args) throws IOException {
// 创建ServerSocket,并绑定端口
ServerSocket serverSocket = new ServerSocket(9999);
System.out.println("服务启动成功");
while (true) {
Socket socket = serverSocket.accept();
System.out.println("连接成功");
System.out.println("准备接收数据");
byte[] bytes = new byte[1024];
socket.getInputStream().read(bytes);
System.out.println("接收到了数据:" + new String(bytes));
}
}
}
/**
* BIO client
*
* @author endless
* @create 2020-03-23
*/
public class BioClientDemo {
public static void main(String[] args) throws IOException {
// 连接Server
Socket socket = new Socket("127.0.0.1", 9999);
System.out.println("连接成功");
Scanner scanner = new Scanner(System.in);
// 循环等待输入消息
while (true) {
String str = scanner.next();
// 约定退出口令
if ("exit".equalsIgnoreCase(str)) {
socket.close();
System.exit(0);
}
socket.getOutputStream().write(str.getBytes());
socket.getOutputStream().flush();
}
}
}
先运行Server
命令行打印服务启动成功,此时并无客户端连接,所以连接成功并未打印,说明程序被阻塞在了serverSocket.accept()方法
此时运行Client,Server打印日志连接成功和准备接收数据,此时Client尚未发送数据,Server被阻塞在
socket.getInputStream().read(bytes)上,因此其他客户端无法进行连接。
在Client输入Hello回车,此时Server打印接收到了数据:Hello,说明客户端的连接发送过来数据了,此时服务端线程才解阻塞,在这
个情况下,这个Server没有办法处理并发,同时期只能处理一个连接。
那么BIO是如何实现并发呢?答案也很明显,就是使用多线程,我们对Server进行一些小改动。
/** * 这是一个BIOServer * @author endless * @create 2020-03-23 */ public class BioServerDemo { public static void main(String[] args) throws IOException { ServerSocket serverSocket = new ServerSocket(9999); System.out.println("服务启动成功"); while (true) { Socket socket = serverSocket.accept(); new Thread(()->{ System.out.println("连接成功"); System.out.println("准备接收数据"); byte[] bytes = new byte[1024]; try { socket.getInputStream().read(bytes); } catch (IOException e) { e.printStackTrace(); } System.out.println("接收到了数据:" + new String(bytes)); }).start(); } } }
使用子线程来对接收到的Socket进行处理,这样每个连接都被阻塞在单独的线程上,就可以实现并发访问Server。
总结:BIO的阻塞有两个地方:accept()和read(),并且BIO的并发只能通过多线程。
但是这里会有一个问题,就是如果绝大部分的连接都没有进行数据传输,只是建立了连接,这样就会产生很多无效的线程,而线程又
是非常宝贵的稀缺资源,这样就会白白损失很多的性能,这也是BIO最大的性能瓶颈。
那能不能只用一个线程就能实现并发并且处理全部的连接呢?是否能设计一个Api,让accept和read不再阻塞,使用一个线程就能处
理并发连接呢?答案是肯定的,这里就要用到我们的今天的主角NIO了。
NIO在JDK中被封装在了一个新的类中,我们先来写一个例子,这个例子实现了使用单线程来处理多连接。
/** * NIO Server Demo * * @author endless * @create 2020-03-23 */ public class NioServerDemo { // 保存客户端连接 static List<SocketChannel> channelList = new ArrayList<>(); public static void main(String[] args) throws IOException, InterruptedException { // 创建NIO ServerSocketChannel ServerSocketChannel serverSocket = ServerSocketChannel.open(); serverSocket.bind(new InetSocketAddress(9998)); // 设置ServerSocketChannel为非阻塞 serverSocket.configureBlocking(false); System.out.println("服务启动成功"); while (true) { SocketChannel socketChannel = serverSocket.accept(); if (socketChannel != null) { // 如果有客户端进行连接 System.out.println("连接成功"); // 设置SocketChannel为非阻塞 socketChannel.configureBlocking(false); // 保存客户端连接在List中 channelList.add(socketChannel); } // 遍历连接进行数据读取 Iterator<SocketChannel> iterator = channelList.iterator(); while (iterator.hasNext()) { SocketChannel o = iterator.next(); ByteBuffer byteBuffer = ByteBuffer.allocate(128); int read = o.read(byteBuffer); // 如果有数据,把数据打印出来 if (read > 0) { System.out.println("接收到消息:" + new String(byteBuffer.array())); } else if (read == -1) { // 如果客户端断开,把socket从集合中去掉 iterator.remove(); System.out.println("客户端断开连接"); } } } } }
客户端可以复用之前的BIO客户端
运行NIOServer,Server启动完毕后运行两个Client,各发送一条消息进行测试
控制台显示两个连接成功,并且接收到了来自两个客户端的消息,表明Server可以使用单线程处理并发连接,这个Api的原理是什么呢?我们来进一步探究一下。
我们沿着源码一路往下找,会找到一个无法下载源码的文件
这里可以看得出,在Windows系统中,编译后的代码显示直接返回了WindowsSelectorProvider对象,很显然,这个对象是在和windows系统核心中的select方法交互,但是Linux中是不是也是这样呢,我们需要下载一个Linux版本的OpenJDK源码来探究一下。下载
OpenJDK源码下载后解压,Linux源码分布在\openjdk\jdk\src\share和\openjdk\jdk\src\solaris目录中。
在上图中可以看出,JDK在不同的系统中采用了不同的实现方案,这里使用的是EPollSelectorProvider,说明在Linux中,使用的是EPoll来实现的。
我们先来看看serverSocket.configureBlocking(false);到底是如何工作的,沿着源码往下找,发现一个本地方法。
这个本地方法的源码可以在OpenJDK中找到,如下图
上图红框中的函数就是这个本地方法调用的底层C语言的方法,前面的前缀是根据JNI调用规则添加的,我们知道,在C当中是可以直接调用操作系统的Api的,这个方法调用了fcntl这个命令,把传进来的blocking参数设置到了文件描述符上(文件描述符可以看作是一个对象,Linux中一切皆文件,类似于高级语言中的一切皆对象,任何的数据流转都要通过一个文件描述符来操作)。
接着看看serverSocket.accept()是如何实现
上图可以看到,accept方法调用了一个native方法accept0
这个accept0方法的描述简单翻译一下,就是接受一个新的连接,把给定的文件描述符引用设置为新的Socket,并将isaa[0]设置为套接字的远程地址,成功返回1,不成功返回IOStatus.UNAVAILABLE or IOStatus.INTERRUPTED。
继续探究一下本地方法的源码
这里调用了操作系统的accept方法,想知道这个方法文档的同学可以在Linux环境中使用man命令来查看
man命令可以查看Linux详尽的文档,2表示第二章:系统调用指令,最后加上你想查的指令方法即可
这里主要看一下返回值,accept返回一个非负数作为socket的描述符,如果失败返回-1。通过这个文件描述符,Java就可以通过native方法调用操作系统的API来在Socket中进行数据的传输。我们的代码用一个List保存了接收到的Socket,相当于保存了Socket的文件描述符,通过一个线程轮询这些文件描述符即可实现数据通信和处理新连接,这样就节约了大量的线程资源,但是大家想一想这种模型还有什么缺陷。
……
是的,如果连接数太多的话,还是会有大量的无效遍历,例如10000个连接中只有1000个连接时有数据的,但是由于其他9000个连接并没有断开,我们还是要每次轮询都遍历一万次,相当于有十分之九的遍历都是无效的,这显然不是一个让人很满意的状态。
总结:NIO的非阻塞是由操作系统来完成的,SocketChannel与文件描述符一一对应,通过遍历文件描述符来读取数据。
什么是多路复用器
上面的例子还不是Java NIO的完全体,仅仅是将原来的同步阻塞IO优化成了同步非阻塞IO,既然还是同步的,就意味着我们每次遍
历,还是需要对每个Socket进行一次read操作来检查是不是有数据过来,都会调用系统内核的read指令,只不过是把阻塞变成了非阻
塞,如果无用连接很多的话,那么绝大部分的read指令都是无意义的,这就会占用很多的CPU时间。
Linux有select、poll和epoll三个解决方案来实现多路复用,其中的select和poll,有点类似于上面的NIOServerDemo程序,他会把
所有的文件描述符记录在一个数组中,通过在系统内核中遍历来筛选出有数据过来的Socket,只不过是从应用程序中遍历改成了在内核中
遍历,本质还是一样的。
Epoll则使用了事件机制,在复用器中注册了一个回调事件,当Socket中有数据过来的时候调用,通知用户处理信息,这样就不需要对
全部的文件描述符进行轮训了,这就是Epoll对NIO进行的改进。
我们来探究一下在Java中是如何用Epoll实现NIO事件机制的,先对上面的NIOServerDemo再进行改进。
/** * NIO Selector Server Demo * * @author endless * @create 2020-03-23 */ public class NioSelectorServerDemo { public static void main(String[] args) throws IOException, InterruptedException { // 创建NIO ServerSocketChannel ServerSocketChannel serverSocket = ServerSocketChannel.open(); serverSocket.socket().bind(new InetSocketAddress(9998)); // 设置ServerSocketChannel为非阻塞 serverSocket.configureBlocking(false); // 打开Selector处理Channel,即创建epoll Selector selector = Selector.open(); // 将ServerSocket注册到selector用来接收连接 serverSocket.register(selector, SelectionKey.OP_ACCEPT); System.out.println("服务启动成功"); while (true) { // 阻塞等待需要处理的事件发生 selector.select(); // 获取selector中注册的全部事件的 SelectionKey 实例 Set<SelectionKey> selectionKeys = selector.selectedKeys(); Iterator<SelectionKey> iterator = selectionKeys.iterator(); // 遍历SelectionKey对事件进行处理 while (iterator.hasNext()) { SelectionKey key = iterator.next(); iterator.remove(); // 如果是OP_ACCEPT事件,则进行连接获取和事件注册 if (key.isAcceptable()) { ServerSocketChannel server = (ServerSocketChannel) key.channel(); SocketChannel socketChannel = server.accept(); socketChannel.configureBlocking(false); // 这里只注册了读事件,如果需要给客户端发送数据可以注册写事件 socketChannel.register(selector, SelectionKey.OP_READ); System.out.println("客户端连接成功"); } // 如果是OP_READ事件,则进行读取和打印 if (key.isReadable()) { SocketChannel socketChannel = (SocketChannel) key.channel(); ByteBuffer byteBuffer = ByteBuffer.allocate(128); int read = socketChannel.read(byteBuffer); // 如果有数据,把数据打印出来 if (read > 0) { System.out.println("接收到消息:" + new String(byteBuffer.array())); } else if (read == -1) { // 如果客户端断开连接,关闭Socket System.out.println("客户端断开连接"); socketChannel.close(); } } } } } }
这段代码需要关注的点有以下几个方法:
- Selector.open()
- socketChannel.register()
- selector.select()
接下来就来看看这三个方法究竟做了什么。
Select.open()
首先调用了SelectorProvider的openSelector()方法,这个方法返回一个EPollSelectorImpl实例
EPollSelectorProvider.java
public AbstractSelector openSelector() throws IOException {
return new EPollSelectorImpl(this);
}
EPollSelectorImpl的构造方法中new了一个EPollArrayWrapper实例
EPollSelectorImpl.java
EPollSelectorImpl(SelectorProvider sp) throws IOException {
super(sp);
// 创建管道,用long类型返回管道的两个文件描述符。读端以高32位的形式返回,写端以低32位的形式返回。
long pipeFds = IOUtil.makePipe(false);
fd0 = (int) (pipeFds >>> 32);
fd1 = (int) pipeFds;
// 创建Epoll文件描述符,并且创建映射数组记录事件
pollWrapper = new EPollArrayWrapper();
// 初始化中断文件描述符,把新创建的Epoll注册到管道的读文件描述符上
pollWrapper.initInterrupt(fd0, fd1);
fdToKey = new HashMap<>();
}
EPollArrayWrapper的构造方法中创建了Epoll实例
EPollArrayWrapper.java
EPollArrayWrapper() throws IOException {
// 创建了Epoll实例,并将它的事件数组地址记录下来方便操作
epfd = epollCreate();
// the epoll_event array passed to epoll_wait
int allocationSize = NUM_EPOLLEVENTS * SIZE_EPOLLEVENT;
pollArray = new AllocatedNativeObject(allocationSize, true);
pollArrayAddress = pollArray.address();
// eventHigh needed when using file descriptors > 64k
if (OPEN_MAX > MAX_UPDATE_ARRAY_SIZE)
eventsHigh = new HashMap<>();
}
...
private native int epollCreate();
Native方法epollCreate()的源码,调用了内核的epoll_create指令,创建并获取了一个文件描述符。
epoll有三个指令,epoll_create、epoll_ctl、epoll_wait,都可以在Linux环境中使用man命令来查看详细文档。
EPollArrayWrapper.c
JNIEXPORT jint JNICALL
Java_sun_nio_ch_EPollArrayWrapper_epollCreate(JNIEnv *env, jobject this)
{
/*
* epoll_create expects a size as a hint to the kernel about how to
* dimension internal structures. We can't predict the size in advance.
*/
int epfd = epoll_create(256);
if (epfd < 0) {
JNU_ThrowIOExceptionWithLastError(env, "epoll_create failed");
}
return epfd;
}
socketChannel.register()
注册事件时其实并没有对Epoll进行事件添加,而是在只是把它加入了待添加的容器。
public final SelectionKey register(Selector sel, int ops,
Object att)
throws ClosedChannelException
{
synchronized (regLock) {
if (!isOpen())
throw new ClosedChannelException();
if ((ops & ~validOps()) != 0)
throw new IllegalArgumentException();
if (blocking)
throw new IllegalBlockingModeException();
// 判断是否存在事件的key,存在的话就更新,不存在就新建
// 最终都会走到 EPollArrayWrapper#setUpdateEvents方法
SelectionKey k = findKey(sel);
if (k != null) {
k.interestOps(ops);
k.attach(att);
}
if (k == null) {
// New registration
synchronized (keyLock) {
if (!isOpen())
throw new ClosedChannelException();
k = ((AbstractSelector)sel).register(this, ops, att);
addKey(k);
}
}
return k;
}
}
AbstractSelector.register调用实现的子类方法implRegister
EPollSelectorImpl.java
protected void implRegister(SelectionKeyImpl ski) {
if (closed)
throw new ClosedSelectorException();
SelChImpl ch = ski.channel;
int fd = Integer.valueOf(ch.getFDVal());
fdToKey.put(fd, ski);
pollWrapper.add(fd);
keys.add(ski);
}
调用setUpdateEvents写入待更新事件容器
EPollArrayWrapper.java
/**
* Add a file descriptor
*/
void add(int fd) {
// 强制初始update events为0,因为他可能会被之前注册kill掉
synchronized (updateLock) {
assert !registered.get(fd);
setUpdateEvents(fd, (byte)0, true);
}
}
/**
* 设置文件描述符的待更新事件到 eventsHigh 中。
*/
private void setUpdateEvents(int fd, byte events, boolean force) {
if (fd < MAX_UPDATE_ARRAY_SIZE) {
if ((eventsLow[fd] != KILLED) || force) {
eventsLow[fd] = events;
}
} else {
Integer key = Integer.valueOf(fd);
if (!isEventsHighKilled(key) || force) {
eventsHigh.put(key, Byte.valueOf(events));
}
}
}
selector.select()
调用selector.select()时,会将刚才注册的待更新事件绑定到文件描述符上,然后进入阻塞状态等待事件回调。
EPollSelectorImpl.java
protected int doSelect(long timeout) throws IOException {
if (closed)
throw new ClosedSelectorException();
processDeregisterQueue();
try {
begin();
pollWrapper.poll(timeout);
} finally {
end();
}
processDeregisterQueue();
int numKeysUpdated = updateSelectedKeys();
if (pollWrapper.interrupted()) {
// Clear the wakeup pipe
pollWrapper.putEventOps(pollWrapper.interruptedIndex(), 0);
synchronized (interruptLock) {
pollWrapper.clearInterrupted();
IOUtil.drain(fd0);
interruptTriggered = false;
}
}
return numKeysUpdated;
}
EPollArrayWrapper.java
int poll(long timeout) throws IOException {
// 将上一步待更新的时间进行注册
updateRegistrations();
// 进入阻塞状态,等待事件发生
updated = epollWait(pollArrayAddress, NUM_EPOLLEVENTS, timeout, epfd);
for (int i=0; i<updated; i++) {
if (getDescriptor(i) == incomingInterruptFD) {
interruptedIndex = i;
interrupted = true;
break;
}
}
return updated;
}
/**
* Update the pending registrations.
*/
private void updateRegistrations() {
synchronized (updateLock) {
int j = 0;
while (j < updateCount) {
int fd = updateDescriptors[j];
short events = getUpdateEvents(fd);
boolean isRegistered = registered.get(fd);
int opcode = 0;
if (events != KILLED) {
if (isRegistered) {
opcode = (events != 0) ? EPOLL_CTL_MOD : EPOLL_CTL_DEL;
} else {
opcode = (events != 0) ? EPOLL_CTL_ADD : 0;
}
if (opcode != 0) {
// 调用native方法进行事件绑定
epollCtl(epfd, opcode, fd, events);
if (opcode == EPOLL_CTL_ADD) {
registered.set(fd);
} else if (opcode == EPOLL_CTL_DEL) {
registered.clear(fd);
}
}
}
j++;
}
updateCount = 0;
}
}
总结:到此为止,我们已经粗略的把整个NIO调用流程都梳理了一遍,Java调用了操作系统的Api来创建Socket,获取到Socket的文件描述符,再创建一个Selector对象,对应操作系统的EPoll描述符,将获取到的Socket连接的文件描述符的事件绑定到Selector对应的EPoll文件描述符上,进行事件的异步通知,这样就实现了使用一条线程,并且不需要太多的无效的遍历,将事件处理交给了操作系统内核,大大提高了效率。
EPoll指令详解
epoll_create
int epoll_create(int size);
创建一个epoll实例,并返回一个非负数作为文件描述符,用于对epoll接口的所有后续调用。参数size代表可能会容纳size个描述符,但size不是一个最大值,只是提示操作系统它的数量级,现在这个参数基本上已经弃用了。
epoll_ctl
int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event);
使用文件描述符epfd引用的epoll实例,对目标文件描述符fd执行op操作。
参数epfd表示epoll对应的文件描述符,参数fd表示socket对应的文件描述符。
参数op有以下几个值:EPOLL_CTL_ADD:注册新的fd到epfd中,并关联时间eventEPOLL_CTL_MOD:修改已经注册的fd的监听事件;EPOLL_CTL_DEL:从epfd中移除fd,并且忽略掉绑定的event,这时event可以为null;
参数event是一个结构体。
struct epoll_event {
__uint32_t events; /* Epoll events */
epoll_data_t data; /* User data variable */
};
typedef union epoll_data {
void *ptr;
int fd;
__uint32_t u32;
__uint64_t u64;
} epoll_data_t;
events有很多可选值,这里只举例最常见的几个:
EPOLLIN :表示对应的文件描述符是可读的(close也会发送消息);EPOLLOUT:表示对应的文件描述符是可写的;EPOLLERR:表示对应的文件描述符发生了错误;
成功则返回0,失败返回-1
epoll_wait
int epoll_wait(int epfd, struct epoll_event *events, int maxevents, int timeout);
等待文件描述符epfd上的事件。
epfd是Epoll对应的文件描述符,events表示调用者所有可用事件的集合,maxevents表示最多等到多少个事件就返回,timeout是超时时间。
完
操作系统的IO还涉及到零拷贝和直接内存两部分的知识,也是操作系统提高性能的利器,将在以后的文章中进行探讨。