网络与IO知识扫盲（六）：多路复用器

NIO存在的问题

NIO的优势：可以通过一个或几个线程来解决N个IO连接的处理

NIO存在C10K问题：当客户端连接的数量达到10K时，单线程每循环一次所有的fd，成本是O(n)复杂度，每一次循环都会有10K次系统调用，但是有意义的调用可能只有三五个，大多数调用是无意义的浪费资源。

多路复用器

还是沿用上一版当中的 Java 代码

package com.bjmashibing.system.io;import java.io.IOException; import java.net.InetSocketAddress; 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;public class SocketMultiplexingSingleThreadv1 {//坦克一、二期（netty）private ServerSocketChannel server = null;private Selector selector = null; //linux 提供多路复用器（select poll epoll kqueue）等，在java中被封装为Selector。拓展：nginx的event模块int port = 9090;public void initServer() {try {server = ServerSocketChannel.open();server.configureBlocking(false); // 设置成非阻塞server.bind(new InetSocketAddress(port)); // 绑定监听的端口号//如果在epoll模型下，Selector.open()其实完成了epoll_create，可能给你返回了一个 fd3selector = Selector.open(); // 可以选择 select poll *epoll，在linux中会优先选择epoll 但是可以在JVM使用-D参数修正/*server 约等于 listen 状态的 fd4server.register()初始化过程：1、如果在select，poll的模型下，Selector.open()操作实际上是在jvm里开辟一个数组，把fd4放进去2、如果在epoll的模型下，调用了epoll_ctl(fd3,ADD,fd4,EPOLLIN)*/server.register(selector, SelectionKey.OP_ACCEPT);} catch (IOException e) {e.printStackTrace();}}public void start() {initServer();System.out.println("服务器启动了");try {while (true) { //死循环Set<SelectionKey> keys = selector.keys();System.out.println("keys.size() = " + keys.size());//1、selector.select调用多路复用器(分为select,poll or epoll(实质上是调用的epoll_wait))/*java中调用的select()方法是啥意思：1、如果用select，poll模型，其实调的是内核的select方法，并传入参数（fd4），或者poll(fd4)2、如果用epoll模型，其实调用的是内核的epoll_wait()，因为fd4在上面的epoll_ctl已经传进去了注意：参数可以带时间。如果没有时间，或者时间是0，代表阻塞。如果有时间，则设置一个超时时间。方法selector.wakeup()可以外部控制让它不阻塞。这时select的结果返回是0。懒加载：其实再触碰到selector.select()调用的时候，触发了epoll_ctl的调用*/while (selector.select(500) > 0) {Set<SelectionKey> selectionKeys = selector.selectedKeys(); // 拿到返回的有状态的fd结果集Iterator<SelectionKey> iter = selectionKeys.iterator(); // 将有状态的fd结果集转成迭代器//所以，不管你是哪一种多路复用器，你只能告诉我fd的状态，我作为应用程序，还需要一个一个的去处理他们的R/W。同步好辛苦！！！//我们之前用NIO的时候，需要自己对每一个fd都去调用系统调用，浪费资源。那么你看，现在是不是只调用了一次select方法，就能知道具体的那些可以R/W了？是不是很省力？while (iter.hasNext()) {SelectionKey key = iter.next();iter.remove(); //这时一个set，不移除的话会重复循环处理if (key.isAcceptable()) { //我前边强调过，socket分为两种，一种是listen的，一种是用于通信 R/W 的，所以拿到之后需要判断它是可读还是可写。//这里是重点，如果要去接受一个新的连接，语义上，accept接受连接且返回新连接的FD，对吧？//那新的FD放在哪里？//1、如果使用select，poll的时候，因为他们在内核没有开辟空间，那么由jvm去维护一个集合(是个native方法)，和前边的fd4那个listen的放在一起//2、如果使用epoll的话，我们希望通过epoll_ctl把新的客户端fd注册到内核空间acceptHandler(key);} else if (key.isReadable()) {readHandler(key);//在当前线程，readHandler处理了很多东西，那么这个方法可能会阻塞，如果阻塞了十年，其他的IO早就没电了。那其他的IO怎么办？//所以，这就是为什么提出了IO THREADS，我把读到的东西扔出去，而不是现场处理//你想，redis是不是用了epoll？redis是不是有个io threads的概念？redis是不是单线程的？它的worker是单线程的，但是它的io threads是多线程的。//你想，tomcat 8,9版本之后，是不是也提出了一种异步的处理方式？是不是也在 IO 和处理上解耦？//这些都是等效的。}}}}} catch (IOException e) {e.printStackTrace();}}public void acceptHandler(SelectionKey key) {try {ServerSocketChannel ssc = (ServerSocketChannel) key.channel();SocketChannel client = ssc.accept(); //来啦，目的是调用accept接受客户端 fd7client.configureBlocking(false);ByteBuffer buffer = ByteBuffer.allocate(8192); //前边讲过了/*你看，调用了register。register做了什么呢？1、select，poll：在jvm里开辟一个数组，把 fd7 放进去2、epoll：调用epoll_ctl(fd3,ADD,fd7,EPOLLIN*/client.register(selector, SelectionKey.OP_READ, buffer);System.out.println("新客户端：" + client.getRemoteAddress());} catch (IOException e) {e.printStackTrace();}}public void readHandler(SelectionKey key) {SocketChannel client = (SocketChannel) key.channel();ByteBuffer buffer = (ByteBuffer) key.attachment();buffer.clear();int read = 0;try {while (true) {read = client.read(buffer);if (read > 0) {buffer.flip();while (buffer.hasRemaining()) {client.write(buffer);}buffer.clear();} else if (read == 0) {break;} else {client.close();break;}}} catch (IOException e) {e.printStackTrace();}}public static void main(String[] args) {SocketMultiplexingSingleThreadv1 service = new SocketMultiplexingSingleThreadv1();service.start();} }

同一套代码 java NIO 的 Selector 对应到 poll，epoll时的不同底层实现

以poll的形式启动

服务端启动

客户端连接进来之后，观察strace的监控输出

以epoll的形式启动

服务端启动

客户端连接进来之后

客户端发送一些数据，观察strace的监控输出

一端断开连接时，产生的中间状态

FIN_WAIT, CLOSE_WAIT, TIME_WAIT 的关系

如果有客户端断开连接的时候，四次分手的最后一个包没有收到，服务端的资源还要在 TIME_WAIT 停留一会儿。消耗的是socket四元组的规则。

改进1：增加注册写事件、将读写独立抛出线程

在单线程循环中，如果某一个连接的读写操作耗费了大量的时间，会影响其他连接的读写。所以我们将读写独立抛出线程去处理。
但是这样需要在抛出的线程中使用 key. cancel() ，否则，同一个连接在被处理完之前始终保持"有状态"，有可能会被重复调起

package com.bjmashibing.system.io;import java.io.IOException; import java.net.InetSocketAddress; 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;public class SocketMultiplexingSingleThreadv2 {private ServerSocketChannel server = null;private Selector selector = null; //linux 多路复用器（select poll epoll） nginx event{}int port = 9090;public void initServer() {try {server = ServerSocketChannel.open();server.configureBlocking(false);server.bind(new InetSocketAddress(port));selector = Selector.open(); // select poll *epollserver.register(selector, SelectionKey.OP_ACCEPT);} catch (IOException e) {e.printStackTrace();}}public void start() {initServer();System.out.println("服务器启动了。。。。。");try {while (true) { // Set<SelectionKey> keys = selector.keys(); // System.out.println(keys.size()+" size");while (selector.select(50) > 0) {Set<SelectionKey> selectionKeys = selector.selectedKeys();Iterator<SelectionKey> iter = selectionKeys.iterator();while (iter.hasNext()) {SelectionKey key = iter.next();iter.remove();if (key.isAcceptable()) {acceptHandler(key);} else if (key.isReadable()) {key.cancel(); //现在多路复用器里把key cancel了readHandler(key);//还是阻塞的嘛？ 即便以抛出了线程去读取，但是在时差里，这个key的read事件会被重复触发} else if (key.isWritable()) { //我之前没讲过写的事件！！！！！//写事件<-- send-queue 只要是空的，就一定会给你返回可以写的事件，就会回调我们的写方法//你真的要明白：什么时候写？不是依赖send-queue队列是不是有空间，真正的是因为：//1，你已经准备好要写什么了//2，然后你才关心send-queue是否有空间//3，所以，read 操作一开始就要 register，但是write操作依赖以上关系，什么时候用什么时候才注册//4，如果一开始就注册了write的事件，会进入死循环，一直调起！！！key.cancel();writeHandler(key);}}}}} catch (IOException e) {e.printStackTrace();}}private void writeHandler(SelectionKey key) {new Thread(() -> {System.out.println("write handler...");SocketChannel client = (SocketChannel) key.channel();ByteBuffer buffer = (ByteBuffer) key.attachment();buffer.flip();while (buffer.hasRemaining()) {try {client.write(buffer);} catch (IOException e) {e.printStackTrace();}}try {Thread.sleep(2000);} catch (InterruptedException e) {e.printStackTrace();}buffer.clear();key.cancel();try {client.close();System.out.println("here close");} catch (IOException e) {e.printStackTrace();}}).start();}public void acceptHandler(SelectionKey key) {try {ServerSocketChannel ssc = (ServerSocketChannel) key.channel();SocketChannel client = ssc.accept();client.configureBlocking(false);ByteBuffer buffer = ByteBuffer.allocate(8192);client.register(selector, SelectionKey.OP_READ, buffer);System.out.println("-------------------------------------------");System.out.println("新客户端：" + client.getRemoteAddress());System.out.println("-------------------------------------------");} catch (IOException e) {e.printStackTrace();}}public void readHandler(SelectionKey key) {new Thread(() -> {System.out.println("read handler.....");SocketChannel client = (SocketChannel) key.channel();ByteBuffer buffer = (ByteBuffer) key.attachment();buffer.clear();int read = 0;try {while (true) {read = client.read(buffer);System.out.println(Thread.currentThread().getName() + " " + read);if (read > 0) {client.register(key.selector(), SelectionKey.OP_WRITE, buffer);} else if (read == 0) {break;} else {client.close();break;}}} catch (IOException e) {e.printStackTrace();}}).start();}public static void main(String[] args) {SocketMultiplexingSingleThreadv2 service = new SocketMultiplexingSingleThreadv2();service.start();} }

总结

以上是生活随笔为你收集整理的网络与IO知识扫盲（六）：多路复用器的全部内容，希望文章能够帮你解决所遇到的问题。

如果觉得生活随笔网站内容还不错，欢迎将生活随笔推荐给好友。