Netty三大核心组件之Selector

矫情吗；* 2023-06-18 09:58 14阅读 0赞

之前进行socket编程时，accept方法会一直阻塞，直到有客户端请求的到来，并返回socket进行相应的处理。整个过程是流水线的，处理完一个请求，才能去获取并处理后面的请求，当然也可以把获取socket和处理socket的过程分开，一个线程负责accept，一个线程池负责处理请求。

但NIO提供了更好的解决方案，采用选择器（Selector）返回已经准备好的socket，并按顺序处理，基于通道（Channel）和缓冲区（Buffer）来进行数据的传输。

### Selector ###

这里出来一个新概念，selector，具体是一个什么样的东西？

想想一个场景：在一个养鸡场，有这么一个人，每天的工作就是不停检查几个特殊的鸡笼，如果有鸡进来，有鸡出去，有鸡生蛋，有鸡生病等等，就把相应的情况记录下来，如果鸡场的负责人想知道情况，只需要询问那个人即可。

在这里，这个人就相当Selector，每个鸡笼相当于一个SocketChannel，每个线程通过一个Selector可以管理多个SocketChannel。

![format_png][]

为了实现Selector管理多个SocketChannel，必须将具体的SocketChannel对象注册到Selector，并声明需要监听的事件（这样Selector才知道需要记录什么数据），一共有4种事件：

1、**connect**：客户端连接服务端事件，对应值为SelectionKey.OPCONNECT(8)

2、**accept**：服务端接收客户端连接事件，对应值为SelectionKey.OPACCEPT(16)

3、**read**：读事件，对应值为SelectionKey.OPREAD(1)

4、**write**：写事件，对应值为SelectionKey.OPWRITE(4)

这个很好理解，每次请求到达服务器，都是从connect开始，connect成功后，服务端开始准备accept，准备就绪，开始读数据，并处理，最后写回数据返回。

所以，当SocketChannel有对应的事件发生时，Selector都可以观察到，并进行相应的处理。

ServerSocketChannel注册

serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);

**ServerSocketChannel的Operation Set只能是OP\_ACCEPT**，如果在注册的时候添加了OP\_CONNECT、OP\_WRITE或OP\_READ会报异常。例如按照以下写法

serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT | SelectionKey.OP_CONNECT);

就会抛出下面的异常

Exception in thread "main" java.lang.IllegalArgumentException
    	at java.nio.channels.spi.AbstractSelectableChannel.register(AbstractSelectableChannel.java:199)
    	at java.nio.channels.SelectableChannel.register(SelectableChannel.java:280)
    	at com.nio.sample.selector.SelectorServerSocketChannelSample.main(SelectorServerSocketChannelSample.java:27)

ServerSocketChannel的`validOps`可以看到只有OP\_ACCEPT是合法的

public final int validOps() {
        return SelectionKey.OP_ACCEPT;
    }

SocketChannel注册

socketChannel.register(selector, SelectionKey.OP_CONNECT);

**SocketChannel的Operation Set只能是OP\_CONNECT、OP\_WRITE和OP\_READ**，如果在注册的时候添加了OP\_ACCEPT同样会报异常。

SocketChannel的`validOps`可以看到只有OP\_READ、OP\_WRITE、OP\_CONNECT是合法的

public final int validOps() {
        return (SelectionKey.OP_READ
                | SelectionKey.OP_WRITE
                | SelectionKey.OP_CONNECT);
    }

注册成功之后，我们通过一个demo实现，客户端和服务端交互：

服务端：

public class Server {
        public static void main(String[] args) {
            try (
                    ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
                    Selector selector = Selector.open();
            ) {
                serverSocketChannel.bind(new InetSocketAddress("127.0.0.1", 8080));
                serverSocketChannel.configureBlocking(false);
                System.out.println("server 启动...");
                serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
                while (selector.select() > 0) {
                    Iterator<SelectionKey> keyIterator = selector.selectedKeys().iterator();
                    while(keyIterator.hasNext()) {
                        SelectionKey key = keyIterator.next();
                        if (key.isReadable()) {
                            ByteBuffer buffer = ByteBuffer.allocateDirect(1024);
                            SocketChannel socketChannel = (SocketChannel) key.channel();
                            int readBytes = socketChannel.read(buffer);
                            if (readBytes > 0) {
                                buffer.flip();
                                byte[] bytes = new byte[buffer.remaining()];
                                buffer.get(bytes);
                                String body = new String(bytes, StandardCharsets.UTF_8);
                                System.out.println("server 收到：" + body);
                            }
                        } else if(key.isAcceptable()) {
                            SocketChannel socketChannel = serverSocketChannel.accept();
                            socketChannel.configureBlocking(false);
                            socketChannel.register(selector, SelectionKey.OP_READ);
                        }
                        keyIterator.remove();
                    }
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

客户端:

public class Client {
        public static void main(String[] args) {
            try (
                    SocketChannel socketChannel = SocketChannel.open();
            ) {
                socketChannel.connect(new InetSocketAddress("127.0.0.1", 8080));
                System.out.println("client 启动...");
    
                ByteBuffer buffer = ByteBuffer.allocateDirect(1024);
                buffer.put("hi, 这是client".getBytes(StandardCharsets.UTF_8));
                buffer.flip();
                socketChannel.write(buffer);
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

[format_png]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly91c2VyLWdvbGQtY2RuLnhpdHUuaW8vMjAxNy8xMi8yNi8xNjA5Mjc5ZDhhOTQwZTEw?x-oss-process=image/format,png