网络编程IO多路复用

╰+哭是因爲堅強的太久メ 2023-01-19 08:10 23阅读 0赞

> **关键词：**IO多路复用    Selector    select

在《[网络编程从BIO到NIO][BIO_NIO]》中，最后提到。在使用**NIO**进行网络编程时，服务器端可以在非阻塞的情况下接收客户端**SocketChannel**连接（accept）、接收客户端发送过来的内容（read）。在单线程处理接收SocketChannel连接时，为了记录每一个连接的SocketChannel，需要将其存放在一个集合中，并不断的循环遍历集合。因为我们并不能知道哪一个SocketChannel发送了数据，哪一个没有发送数据，所以只能每一个SocketChannel都调用read方法，若有数据则接收数据，没有数据此方法也不会阻塞。

假如此时有10000个SocketChannel连接进来，此时集合中存放了SocketChannel个元素，在while循环中不断地遍历集合，调用SocketChannel的read方法接收数据。若只有1个SocketChannel发送了数据，也就是在一次遍历中只有一个SocketChannel是有效的，其余的9999个SocketChannel只是陪跑，做了无用功。当调用SocketChannel的read方法时，会调用一次操作系统对应的read方法进行系统调用。而系统调用就会发送从用户态到内核态的上下文切换。9999个无效的系统调用极大的造成了资源上的开销和浪费。

好在操作系统提供了一个功能，不用程序员编写代码将每一个连接的SocketChannel记录在集合中了。操作系统负责监听SocketChannel以及**ServerSocketChannel**的状态，当SocketChannel发送了数据进来，或者ServerSocketChannel写了数据出去（即状态变化），会被操作系统监听到，并将监听到的SocketChannel或ServerSocketChannel加入到集合中，返回给系统调用者。系统调用者只需要每次调用一次操作系统提供的方法去监听是否有状态变化的Channel（可读、可写等）集合，减少了很多无效的系统调用。

上述功能即多路复用。

下面先来看看Linux操作系统提供的多路复用系统调用API，以**select**为例，**poll**、**epoll**大致相似。

--------------------

# **█** Linux下的select #

（1）*yum install man-pages -y*

以CentOS7为例，下载man手册

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl81MDUxODI3MQ_size_16_color_FFFFFF_t_70][]

（2）*man 2 select*，查看select帮助手册

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl81MDUxODI3MQ_size_16_color_FFFFFF_t_70 1][]

**①select**

多路复用select对外提供的api。

int select(int nfds, fd_set *readfds, fd_set *writefds,
              fd_set *exceptfds, struct timeval *timeout);

\*readfds：监听可读状态的文件描述符集合。操作系统会监听此集合中的文件描述符是否有数据可读。在Java网络编程中，可将文件描述符看做SocketChannel、ServerSocketChannel对象等。

\*writefds：监听可写状态的文件描述符集合。操作系统会监听此集合中的文件描述符是否有数据可写。

\*exceptfds：监听有异常的文件描述符集合。操作系统会监听此集合中的文件描述符是否有异常发生。

对于上述三个集合，在一次调用select返回时，三个集合的内容会变更为满足相应状态的文件描述符集合。

nfds：上述三个集合中，任意一个集合元素数量的最大个数，即集合中最多存放多少个文件描述符。

\*timeout：阻塞等待时长。当达到等待时长后，不管有没有符合条件的文件描述符，方法会立即返回0。设置的值为0时，不阻塞等待，方法立即返回。设置的值为NULL，方法会一直阻塞，直到有元素被添加到集合中。

返回值：返回三个集合中文件描述符的个数。0，表示没有符合状态的文件描述符。-1，表示发生错误。

**②FD\_CLR**

将指定的文件描述符从指定的集合中移除。

void FD_CLR(int fd, fd_set *set);

fd：文件描述符。

\*set：固定大小的集合，可以为可读集合、可写集合、异常集合。

**③FD\_ISSET**

查看指定的文件描述符是否在指定的集合中。

int  FD_ISSET(int fd, fd_set *set);

**④FD\_SET**

将指定的文件描述符添加到指定的集合中。

void FD_SET(int fd, fd_set *set);

**⑤FD\_ZERO**

清空指定的集合。

void FD_ZERO(fd_set *set);

--------------------

# **█** Java中的Selector #

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl81MDUxODI3MQ_size_16_color_FFFFFF_t_70 2][]

*  **Selector**

Java提供抽象类Selector。Selector对外提供API给程序员使用，内部调用操作系统提供的select函数。

// 抽象类Selector
    public abstract class Selector implements Closeable {
        // protected访问权限，只能同一个包或不同包的子类调用，
        // 所以无法通过new 构造器来创建Selector对象
        protected Selector() { }
        
        // 获取Selector对象，开启一个Selector。与close方法对应
        public static Selector open() throws IOException {
            return SelectorProvider.provider().openSelector();
        }
    
        public abstract boolean isOpen();
        
        // 获取SelectorProvider
        public abstract SelectorProvider provider();
         
        // 获取所有已注册到当前Selector中的Channel。 
        public abstract Set<SelectionKey> keys();
        
        // 获取可读、可写、异常状态的SelectionKey集合
        public abstract Set<SelectionKey> selectedKeys();
        
        // 即调用Linux操作系统提供的select方法，此方法不会阻塞。
        public abstract int selectNow() throws IOException;
        
        // 阻塞超时的select，在timeout时间内，及时没有可用的SelectionKey也立即返回。
        public abstract int select(long timeout)
            throws IOException;
        
        // 一直阻塞，直到有可用的SelectionKey。
        public abstract int select() throws IOException;
        
        // 唤醒select()方法的阻塞操作，使其立即返回结果。
        public abstract Selector wakeup();
        
        // 关闭Selector，与open方法对应
        public abstract void close() throws IOException;
    
    }

①open、close

open创建和获取Selector对象。close关闭Selector。

②select

调用操作系统提供的select方法去获取可读、可写等状态的文件描述符。（注1：文件描述符在java中即SocketChannel、ServerSocketChannel。注2：可读、可写、异常状态后文统称为可用状态）

public abstract int select() throws IOException;

和Linux中select函数对应。此方法是阻塞方法，即当没有可用状态的Channel时，此方法会一直阻塞不返回，直到任一集合中有元素，则返回元素的个数。对应Linux中select函数的\*timeout参数设置为NULL。

在windows版本的jdk中，最终调用的是Selector实现类**WindowsSelectorImpl**的poll方法：

private int poll() throws IOException {
        return this.poll0(WindowsSelectorImpl.this.pollWrapper.pollArrayAddress, Math.min(WindowsSelectorImpl.this.totalChannels, 1024), this.readFds, this.writeFds, this.exceptFds, WindowsSelectorImpl.this.timeout);
    }

private native int poll0(long var1, int var3, int[] var4, int[] var5, int[] var6, long var7);

看看上面的参数，是不是和Linux操作系统提供的select方法参数大致相同。（不管是Windows还是Linux操作系统，万变不离其宗）

③select(long timeout)

带有阻塞超时等待的select，当阻塞等待时间达到timeout时长，及时没有可用的Channel，方法也会立即返回。对应Linux中select函数的\*timeout参数。

public abstract int select(long timeout)
        throws IOException;

④selectNow()

不管有没有可用状态的Channel，调用方法立即返回。没有的话返回0。对应Linux中select函数的\*timeout参数设置为0。

public abstract int selectNow() throws IOException;

⑤selectedKeys

存放可用状态的Channel集合。当集合有元素时，集合中元素的个数与select方法的返回值相等。

⑥keys

与selectedKeys不同的是，keys返回所有已注册到当前Selector中的Channel。即使Channel还未达到可用状态。

⑦wakeup

因为select()方法是会阻塞等待的，可使用wakeup唤醒阻塞，方法立即返回。没有可用状态的channel则返回0。

在Selector中并没有提供方法用于与SocketChannel、ServerSocketChannel建立联系。因为Java提供的API并不是Selector去绑定SocketChannel、ServerSocketChannel，而是SocketChannel、ServerSocketChannel将自己注册到指定的Selector上，在SocketChannel、ServerSocketChannel中提供对应注册方法。

Selector关系图：

WindowsSelectorImpl为windows平台的jdk版本中的实现类。Linux版本的jdk中实现类为**PollSelectorImpl**。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl81MDUxODI3MQ_size_16_color_FFFFFF_t_70 3][]

*  **SelectableChannel**

SocketChannel、ServerSocketChannel共同的父类。提供register方法用于将Channel注册到指定的Selector上。

public abstract class SelectableChannel
        extends AbstractInterruptibleChannel
        implements Channel
    {
        public abstract SelectionKey register(Selector sel, int ops, Object att)
            throws ClosedChannelException;
        
        public final SelectionKey register(Selector sel, int ops)
            throws ClosedChannelException
       {
            return register(sel, ops, null);
       }
        
    }

register参数：

①sel

Selector类型的参数，即指定一个Selector，将当前SelectableChannel注册到Selector上。

②ops

设置当当前的SelectableChannel达到此可用状态时，调用select方法会将加入到对应的集合中，并返回。

当前SelectableChannel为可读状态时：int **OP\_READ** = 1 << 0 = 1

当前SelectableChannel为可写状态时：int **OP\_WRITE** = 1 << 2 = 4

当前SelectableChannel为可连接状态时：int **OP\_CONNECT** = 1 << 3 = 8

当前SelectableChannel为可接收状态时：int **OP\_ACCEPT** = 1 << 4 = 16

可读状态，表示此Channel可调用read方法获取到数据。

可写状态，表示此Channel可调用write方法写数据。

可连接状态，表示此Channel可调用connect方法连接。

可接收状态，表示此Channel可调用accept方法接收，即有客户端连接进来了。

SocketChannel具有read、write、connect方法，所以SocketChannel可设置的有效值为OP\_READ、OP\_WRITE、OP\_CONNECT。

ServerSocketChannel具有accept方法，所以ServerSocketChannel可设置的有效值为OP\_ACCEPT。

当向ServerSocketChannel设置OP\_READ、OP\_WRITE、OP\_CONNECT值时，会抛出java.lang.IllegalArgumentException。同理，当向SocketChannel设置OP\_ACCEPT，也会抛出异常。（此处关于ops的合法值，就看该Channel有哪些方法，有read方法就可以设置OP\_READ，有accept方法就可以设置OP\_ACCEPT）

serverSocketChannel.register(selector, SelectionKey.OP_READ);
    
    // 抛出异常 java.lang.IllegalArgumentException

③att

附件。比如在网络中传输一个文件，可通过此参数传输。

**注意点：**

①当调用ServerSocketChannel或SocketChannel的register方法时，当前的ServerSocketChannel或SocketChannel必须设置为非阻塞，即调用**configureBlocking(false)**，否则会抛出java.nio.channels.IllegalBlockingModeException。

// 创建Selector
    Selector selector = Selector.open();
    
    // 创建ServerSocketChannel
    ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
    SocketAddress socketAddress = new InetSocketAddress(8899);
    serverSocketChannel.bind(socketAddress);
    
    // 此处代码为必须
    serverSocketChannel.configureBlocking(false);
    
    // 将当前serverSocketChannel注册到selector上
    serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);

②一个Channel可以注册到多个Selector上，一个Selector也可以被多个Channel注册。

Selector selector1 = Selector.open();
    Selector selector2 = Selector.open();
    Selector selector3 = Selector.open();
    
    ServerSocketChannel serverSocketChannel1 = ServerSocketChannel.open();
    ServerSocketChannel serverSocketChannel2 = ServerSocketChannel.open();
    SocketChannel socketChannel = SocketChannel.open();
    
    // 此处省略部分代码
    
    serverSocketChannel1.register(selector1, SelectionKey.OP_ACCEPT);
    serverSocketChannel1.register(selector2, SelectionKey.OP_ACCEPT);
    serverSocketChannel1.register(selector3, SelectionKey.OP_ACCEPT);
    
    serverSocketChannel2.register(selector1, SelectionKey.OP_ACCEPT);
    socketChannel.register(selector1, SelectionKey.OP_READ);

*  **SelectionKey**

可以将其看做是Channel与Selector的组合。一个Channel对应一个SelectionKey，可通过SelectionKey获取到其对应的Channel和Selector。

public abstract class SelectionKey {
    
        protected SelectionKey() { }
        
        // 获取对应的SocketChannel或ServerSocketChannel
        public abstract SelectableChannel channel();
    
        // 获取对应的selector
        public abstract Selector selector();
    
        public abstract boolean isValid();
        
        public abstract void cancel();
    
        public abstract int interestOps();
    
        public abstract SelectionKey interestOps(int ops);
    
        public abstract int readyOps();
    
        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;
    
        // 是否可读状态
        public final boolean isReadable() {
            return (readyOps() & OP_READ) != 0;
        }
    
        // 是否可写状态
        public final boolean isWritable() {
            return (readyOps() & OP_WRITE) != 0;
        }
    
        // 是否是可连接状态
        public final boolean isConnectable() {
            return (readyOps() & OP_CONNECT) != 0;
        }
    
        // 是否是可accept状态
        public final boolean isAcceptable() {
            return (readyOps() & OP_ACCEPT) != 0;
        }
    }

调用SocketChannel或ServerSocketChannel的register方法，其返回值类型就为SelectionKey。

SelectionKey selectionKey = serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);

--------------------

简单的服务端代码示例：

public static void main(String[] args) {
    
        try {
            // 创建一个ServerSocketChannel
            ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
            // 将ServerSocketChannel绑定到端口号：8899
            serverSocketChannel.bind(new InetSocketAddress(8899));
            // 设置serverSocketChannel为非阻塞
            serverSocketChannel.configureBlocking(false);
    
            // 获取Selector
            Selector selector = Selector.open();
    
            SelectorProvider.provider().openSelector();
    
            // 将serverSocketChannel注册到selector上，并设置关注的事件为ACCEPT事件
            serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
    
            while (true) {
                // 去获取Channel关注的事件，返回值为关注的事件发生的个数
                if (selector.select()>0) {
                    // 发生的事件集合
                    Set<SelectionKey> selectionKeys = selector.selectedKeys();
                    Iterator<SelectionKey> iterator = selectionKeys.iterator();
                    while (iterator.hasNext()) {
                        SelectionKey selectionKey = iterator.next();
                        // 如果是ACCEPT事件
                        if (selectionKey.isAcceptable()) {
                            ServerSocketChannel channel = (ServerSocketChannel) selectionKey.channel();
                            SocketChannel socketChannel = channel.accept();
                            if (socketChannel!=null) {
                                socketChannel.configureBlocking(false);
                                socketChannel.register(selector, SelectionKey.OP_READ);
                            }
                        } else if (selectionKey.isReadable()) { // 如果是READABLE事件
                            SocketChannel channel = (SocketChannel)selectionKey.channel();
                            channel.configureBlocking(false);
                            ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
                            // 接收客户端发来的信息
                            int read = channel.read(byteBuffer);
                            if (read>0) {
                                byteBuffer.flip();
                                System.out.println("客户端【"+channel.getRemoteAddress()+"】："+new String(byteBuffer.array()));
                            }
                            channel.register(selector, SelectionKey.OP_WRITE);
                        } else if (selectionKey.isWritable()) { // 如果是写事件
                            SocketChannel socketChannel = (SocketChannel) selectionKey.channel();
                            ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
                            byteBuffer.put("服务端已接收到你发送的消息,over！".getBytes());
                            byteBuffer.flip();
                            socketChannel.write(byteBuffer);
                            socketChannel.register(selector, SelectionKey.OP_READ);
                        }
                        // 一定要将SeletionKey从集合中移除。
                        iterator.remove();
                    }
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

在第55行代码中，在处理完当前的SelectionKey后，将其从selectedKeys集合中移除，为什么要移除呢？selectedKeys集合中存放在调用一次select方法后可用的channel集合，当处理完成从集合中移除则告诉程序，你已经处理了当前的Channel任务，当下一次该Channel有新的状态可用时才能继续加到集合中。如果一直存在集合中不移除，则表明此时的Channel任务你还没有处理完成，则关于这个Channel的后续任务就不能加入集合中，为了避免覆盖前面的任务。因为selectedKeys是Set类型的集合，不允许重复。对于一个Channel对象来说，其对应的SelectionKey是不变的。所以为了避免一个Channel后续的内容被覆盖掉，当集合中还存在该Channel对应的selectionKey时，不能向集合中继续添加相同的SelectionKey对象。

比如，一个客户端分两次给服务端发送了数据。当第一次发送数据时，服务端感知到此客户端可读，但是还没有开始读取数据时，客户端又发送了第二次数据。此时在第一次数据没有读取完成后，自然第二次数据也不能被读取。当读取完第一次数据之后将其从集合中移除，则表明第一次数据读取完成了，可以进行第二次数据的读取。

[BIO_NIO]: https://blog.csdn.net/weixin_50518271/article/details/108984343?spm=1001.2014.3001.5501
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl81MDUxODI3MQ_size_16_color_FFFFFF_t_70]: /images/20221021/6d7d9c0bf4a34d26ba4071689dcc8880.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl81MDUxODI3MQ_size_16_color_FFFFFF_t_70 1]: /images/20221021/6b5f7622541f4a28a4a472569dfbefd4.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl81MDUxODI3MQ_size_16_color_FFFFFF_t_70 2]: /images/20221021/de7c450e7c644945b2910021abc2d19d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl81MDUxODI3MQ_size_16_color_FFFFFF_t_70 3]: /images/20221021/3784e0e05c914b539303221f46b44f7b.png