深入底层探析网络编程之多路复用器（select,poll,epoll）

行百里er

发布于 2020-12-02 14:30:49

9710

发布于 2020-12-02 14:30:49

文章被收录于专栏：JavaJourney

IO模型

只关注IO，不关注IO读写完成后的事情。同步：程序（APP）自己进行读/写操作异步：由Kernel完成读/写，程序跑起来感觉像没有访问IO，访问的是buffer 阻塞：BLOCKING，一直等待着方法有效的返回结果非阻塞：NONBLOCKING，调用方法的时候就返回是否读取到，（java中要么返回null，要么返回具体的对象）所以IO模型有：同步阻塞：程序（APP）自己读取，调用了方法后一直等待着有效的返回结果同步非阻塞：程序（APP）自己读取，调用方法的瞬间就给出是否读取到的返回结果，这个时候程序要考虑下一次再去读取的问题（比如用while循环）那么异步呢？异步的只有非阻塞的，（异步阻塞无意义）。其实异步的问题暂时不需要讨论，因为IO模型下，目前Linux没有通用内核的异步处理方案。

NIO和多路复用器

nio 需要全部遍历内核fd（比如处于listen状态的文件描述符），用户态内核态需要切换（一次切换就是一次系统调用）才能实现多路复用器：多条路（指IO）只通过一个系统调用，获得所有IO（fd）的状态，然后由程序自己对有状态的IO进行R/W操作。只要是程序自己读写，就说明IO模型是同步的多路复用一般有SELECT,POSIX,POLL,EPOLL,KQUEUE

NIO

多路复用

linux内核多路复用器select,poll,epoll

来看一下底层关于select的描述及api。这里借助于man select指令。

man select

看到描述 select, pselect, FD_CLR, FD_ISSET, FD_SET, FD_ZERO - synchronous I/O multiplexing---这里很明确的给出了select的功能含义：同步I/O多路复用 再来几个描述： An fd_set is a fixed size buffer. Executing FD_CLR() or FD_SET() with a value of fd that is negative or is equal to or larger than FD_SETSIZE will result in undefined behavior. Moreover, POSIX requires fd to be a valid file descriptor 大致意思是：fd_set是固定大小的缓冲区。如果fd值为负或大于等于FD_SETSIZE，则执行FD_CLR（）或FD_SET（），这样会导致不确定的行为。而且，POSIX要求fd是有效的文件描述符。这里提到了FD_SETSIZE，这是一个固定的数值（1024,2048之类的），这是select要求的，而poll则没有。这也是select和poll的区别。

其实到这里，我们可以得出结论：无论nio、select、poll都是要遍历所有的IO，询问状态，但是， NIO的遍历是需要很多次系统调用的，成本在用户态与内核态的切换上；而多路复用器select/poll，这个过程触发了一次系统调用，在这次用户态内核态切换的过程中，把fds传递到内核，内核根据这次用户传递过来的fds进行遍历，修改状态。

多路复用器select/poll的弊端：

每次都要重新重复传递fds（内核开辟空间）
每次内核被调用了之后，针对这次调用，触发了一个遍历fds全量的复杂度

由此，引入epoll这个牛逼的东西。

还是直接看linux对epoll的描述：

man epoll:

epoll - I/O event notification facility 一看到这个大致知道是和事件通知相关的 epoll API执行与poll类似的任务：监视多个文件描述符以查看其中的任何文件是否可以进行I / O。 epoll_create 创建一个epoll实例并返回引用该实例的文件描述符。 epoll_ctl 通过epoll_ctl注册指定文件描述符集合。 epoll_wait 等待I / O事件，如果当前没有可用的事件，则阻塞调用线程。

画个图和select/poll类比一下

实际结合理论

Java中是如何使用多路复用的？我们用一段程序解释一下。

初始化Server

private ServerSocketChannel server = null;
private Selector selector = null;
int port = 9090;

public void initServer() {
    try {
        server = ServerSocketChannel.open();
        server.configureBlocking(false);
        server.bind(new InetSocketAddress(port));

        selector = Selector.open();

        server.register(selector, SelectionKey.OP_ACCEPT);
    } catch (IOException e) {
        e.printStackTrace();
    }
}

Selector 类似于linux下的多路复用器（select poll epoll kqueue），nginx，event{} server.configureBlocking(false) 设置非阻塞 selector = Selector.open() 如果在epoll模型下，open()相当于前面提到的epoll_create方法，返回一个fd实例---假设是fd3 server.register(selector, SelectionKey.OP_ACCEPT) server约等于处于listen状态的fd---设fd4，对于register：若是select/poll模型，则jvm里开辟一个数组把这个fd放进去；若是epoll，则epoll_ctl(fd3,ADD,fd4,EPOLLIN

Start启动Server，并处理事件

public void start() {
    initServer();
    System.out.println("服务器启动了。。。。。");
    try {
        while (true) {
            while (selector.select() > 0) {
                Set<SelectionKey> selectionKeys = selector.selectedKeys(); 
                Iterator<SelectionKey> iter = selectionKeys.iterator();
                
                while (iter.hasNext()) {
                    SelectionKey key = iter.next();
                    iter.remove(); //set  不移除会重复循环处理
                    if (key.isAcceptable()) {
                        acceptHandler(key);
                    } else if (key.isReadable()) {
                        readHandler(key);
                    }
                }
            }
        }
    } catch (IOException e) {
        e.printStackTrace();
    }
}

这段代码除了启动server，重要的逻辑是调用多路复用器。 selector.select()：

select/poll模型下，select()等于内核的select（fd4） poll(fd4)
epoll等于内核的 epoll_wait()，这个可以带时间参数。若没有时间或者参数为0表示阻塞，若有时间则设置一个超时。selector.wakeup() 结果返回0

while (iter.hasNext())：这段代码表示，管你是什么多路复用器，你只能给我状态，我的程序还得一个一个的去处理他们的R/W。这就是同步---真的好辛苦！！！！！！！！ if (key.isAcceptable()) ：这里是重点，如果要去接受一个新的连接，语义上，accept接受连接且返回新连接的FD，那么这个新的FD如何处理？ select/poll，因为他们内核没有空间，那么在jvm中保存和前边的fd4那个listen的一起 epoll：通过epoll_ctl把新的客户端fd注册到内核空间这个具体看下面acceptHandler这个方法

public void acceptHandler(SelectionKey key) {
    try {
        ServerSocketChannel ssc = (ServerSocketChannel) key.channel();
        SocketChannel client = ssc.accept(); //来啦，目的是调用accept接受客户端  fd7
        client.configureBlocking(false);

        ByteBuffer buffer = ByteBuffer.allocate(8192); 

        // 0.0  我类个去
        //你看，调用了register
        /*
        select，poll：jvm里开辟一个数组 fd7 放进去
        epoll：  epoll_ctl(fd3,ADD,fd7,EPOLLIN
         */
        client.register(selector, SelectionKey.OP_READ, buffer);
        System.out.println("-------------------------------------------");
        System.out.println("新客户端：" + client.getRemoteAddress());
        System.out.println("-------------------------------------------");

    } catch (IOException e) {
        e.printStackTrace();
    }
}

这一段和前面的理论知识就对上了，^_^