摘要：允許一個單一的線程來操作多個如果我們的應用程序中使用了多個那么使用很方便的實現這樣的目的但是因為在一個線程中使用了多個因此也會造成了每個傳輸效率的降低使用的圖解如下為了使用我們首先需要將注冊到中隨后調用的方法這個方法會阻塞直到注冊在中的發送

Selector 允許一個單一的線程來操作多個 Channel. 如果我們的應用程序中使用了多個 Channel, 那么使用 Selector 很方便的實現這樣的目的, 但是因為在一個線程中使用了多個 Channel, 因此也會造成了每個 Channel 傳輸效率的降低.
使用 Selector 的圖解如下:

為了使用 Selector, 我們首先需要將 Channel 注冊到 Selector 中, 隨后調用 Selector 的 select()方法, 這個方法會阻塞, 直到注冊在 Selector 中的 Channel 發送可讀寫事件. 當這個方法返回后, 當前的這個線程就可以處理 Channel 的事件了.

通過 Selector.open()方法, 我們可以創建一個選擇器:

Selector selector = Selector.open();

為了使用選擇器管理 Channel, 我們需要將 Channel 注冊到選擇器中:

channel.configureBlocking(false);

SelectionKey key = channel.register(selector, SelectionKey.OP_READ);

注意, 如果一個 Channel 要注冊到 Selector 中, 那么這個 Channel 必須是非阻塞的, 即channel.configureBlocking(false);
因為 Channel 必須要是非阻塞的, 因此 FileChannel 是不能夠使用選擇器的, 因為 FileChannel 都是阻塞的.

注意到, 在使用 Channel.register()方法時, 第二個參數指定了我們對 Channel 的什么類型的事件感興趣, 這些事件有:

Connect, 即連接事件(TCP 連接), 對應于SelectionKey.OP_CONNECT

Accept, 即確認事件, 對應于SelectionKey.OP_ACCEPT

Read, 即讀事件, 對應于SelectionKey.OP_READ, 表示 buffer 可讀.

Write, 即寫事件, 對應于SelectionKey.OP_WRITE, 表示 buffer 可寫.

一個 Channel發出一個事件也可以稱為 對于某個事件, Channel 準備好了. 因此一個 Channel 成功連接到了另一個服務器也可以被稱為 connect ready.
我們可以使用或運算|來組合多個事件, 例如:

int interestSet = SelectionKey.OP_READ | SelectionKey.OP_WRITE;    

注意, 一個 Channel 僅僅可以被注冊到一個 Selector 一次, 如果將 Channel 注冊到 Selector 多次, 那么其實就是相當于更新 SelectionKey 的 interest set. 例如:

channel.register(selector, SelectionKey.OP_READ);
channel.register(selector, SelectionKey.OP_READ | SelectionKey.OP_WRITE);

上面的 channel 注冊到同一個 Selector 兩次了, 那么第二次的注冊其實就是相當于更新這個 Channel 的 interest set 為 SelectionKey.OP_READ | SelectionKey.OP_WRITE.

如上所示, 當我們使用 register 注冊一個 Channel 時, 會返回一個 SelectionKey 對象, 這個對象包含了如下內容:

interest set, 即我們感興趣的事件集, 即在調用 register 注冊 channel 時所設置的 interest set.

ready set

channel

selector

attached object, 可選的附加對象

我們可以通過如下方式獲取 interest set:

int interestSet = selectionKey.interestOps();

boolean isInterestedInAccept  = interestSet & SelectionKey.OP_ACCEPT;
boolean isInterestedInConnect = interestSet & SelectionKey.OP_CONNECT;
boolean isInterestedInRead    = interestSet & SelectionKey.OP_READ;
boolean isInterestedInWrite   = interestSet & SelectionKey.OP_WRITE;    

代表了 Channel 所準備好了的操作.
我們可以像判斷 interest set 一樣操作 Ready set, 但是我們還可以使用如下方法進行判斷:

int readySet = selectionKey.readyOps();

selectionKey.isAcceptable();
selectionKey.isConnectable();
selectionKey.isReadable();
selectionKey.isWritable();

我們可以通過 SelectionKey 獲取相對應的 Channel 和 Selector:

Channel  channel  = selectionKey.channel();
Selector selector = selectionKey.selector();  

我們可以在selectionKey中附加一個對象:

selectionKey.attach(theObject);
Object attachedObj = selectionKey.attachment();

或者在注冊時直接附加:

SelectionKey key = channel.register(selector, SelectionKey.OP_READ, theObject);

我們可以通過 Selector.select()方法獲取對某件事件準備好了的 Channel, 即如果我們在注冊 Channel 時, 對其的可寫事件感興趣, 那么當 select()返回時, 我們就可以獲取 Channel 了.

注意, select()方法返回的值表示有多少個 Channel 可操作.

如果 select()方法返回值表示有多個 Channel 準備好了, 那么我們可以通過 Selected key set 訪問這個 Channel:

Set selectedKeys = selector.selectedKeys();

Iterator keyIterator = selectedKeys.iterator();

while(keyIterator.hasNext()) {
    
    SelectionKey key = keyIterator.next();

    if(key.isAcceptable()) {
        // a connection was accepted by a ServerSocketChannel.

    } else if (key.isConnectable()) {
        // a connection was established with a remote server.

    } else if (key.isReadable()) {
        // a channel is ready for reading

    } else if (key.isWritable()) {
        // a channel is ready for writing
    }

    keyIterator.remove();
}

注意, 在每次迭代時, 我們都調用 "keyIterator.remove()" 將這個 key 從迭代器中刪除, 因為 select() 方法僅僅是簡單地將就緒的 IO 操作放到 selectedKeys 集合中, 因此如果我們從 selectedKeys 獲取到一個 key, 但是沒有將它刪除, 那么下一次 select 時, 這個 key 所對應的 IO 事件還在 selectedKeys 中.
例如此時我們收到 OP_ACCEPT 通知, 然后我們進行相關處理, 但是并沒有將這個 Key 從 SelectedKeys 中刪除, 那么下一次 select() 返回時 我們還可以在 SelectedKeys 中獲取到 OP_ACCEPT 的 key.
注意, 我們可以動態更改 SekectedKeys 中的 key 的 interest set. 例如在 OP_ACCEPT 中, 我們可以將 interest set 更新為 OP_READ, 這樣 Selector 就會將這個 Channel 的 讀 IO 就緒事件包含進來了.

通過 Selector.open() 打開一個 Selector.

將 Channel 注冊到 Selector 中, 并設置需要監聽的事件(interest set)

不斷重復:

調用 select() 方法

調用 selector.selectedKeys() 獲取 selected keys

迭代每個 selected key:

*從 selected key 中獲取 對應的 Channel 和附加信息(如果有的話)

*判斷是哪些 IO 事件已經就緒了, 然后處理它們. 如果是 OP_ACCEPT 事件, 則調用 "SocketChannel clientChannel = ((ServerSocketChannel) key.channel()).accept()" 獲取 SocketChannel, 并將它設置為 非阻塞的, 然后將這個 Channel 注冊到 Selector 中.

*根據需要更改 selected key 的監聽事件.

*將已經處理過的 key 從 selected keys 集合中刪除.

當調用了 Selector.close()方法時, 我們其實是關閉了 Selector 本身并且將所有的 SelectionKey 失效, 但是并不會關閉 Channel.

public class NioEchoServer {
    private static final int BUF_SIZE = 256;
    private static final int TIMEOUT = 3000;

    public static void main(String args[]) throws Exception {
        // 打開服務端 Socket
        ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();

        // 打開 Selector
        Selector selector = Selector.open();

        // 服務端 Socket 監聽8080端口, 并配置為非阻塞模式
        serverSocketChannel.socket().bind(new InetSocketAddress(8080));
        serverSocketChannel.configureBlocking(false);

        // 將 channel 注冊到 selector 中.
        // 通常我們都是先注冊一個 OP_ACCEPT 事件, 然后在 OP_ACCEPT 到來時, 再將這個 Channel 的 OP_READ
        // 注冊到 Selector 中.
        serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);

        while (true) {
            // 通過調用 select 方法, 阻塞地等待 channel I/O 可操作
            if (selector.select(TIMEOUT) == 0) {
                System.out.print(".");
                continue;
            }

            // 獲取 I/O 操作就緒的 SelectionKey, 通過 SelectionKey 可以知道哪些 Channel 的哪類 I/O 操作已經就緒.
            Iterator keyIterator = selector.selectedKeys().iterator();

            while (keyIterator.hasNext()) {

                SelectionKey key = keyIterator.next();

                // 當獲取一個 SelectionKey 后, 就要將它刪除, 表示我們已經對這個 IO 事件進行了處理.
                keyIterator.remove();

                if (key.isAcceptable()) {
                    // 當 OP_ACCEPT 事件到來時, 我們就有從 ServerSocketChannel 中獲取一個 SocketChannel,
                    // 代表客戶端的連接
                    // 注意, 在 OP_ACCEPT 事件中, 從 key.channel() 返回的 Channel 是 ServerSocketChannel.
                    // 而在 OP_WRITE 和 OP_READ 中, 從 key.channel() 返回的是 SocketChannel.
                    SocketChannel clientChannel = ((ServerSocketChannel) key.channel()).accept();
                    clientChannel.configureBlocking(false);
                    //在 OP_ACCEPT 到來時, 再將這個 Channel 的 OP_READ 注冊到 Selector 中.
                    // 注意, 這里我們如果沒有設置 OP_READ 的話, 即 interest set 仍然是 OP_CONNECT 的話, 那么 select 方法會一直直接返回.
                    clientChannel.register(key.selector(), OP_READ, ByteBuffer.allocate(BUF_SIZE));
                }

                if (key.isReadable()) {
                    SocketChannel clientChannel = (SocketChannel) key.channel();
                    ByteBuffer buf = (ByteBuffer) key.attachment();
                    long bytesRead = clientChannel.read(buf);
                    if (bytesRead == -1) {
                        clientChannel.close();
                    } else if (bytesRead > 0) {
                        key.interestOps(OP_READ | SelectionKey.OP_WRITE);
                        System.out.println("Get data length: " + bytesRead);
                    }
                }

                if (key.isValid() && key.isWritable()) {
                    ByteBuffer buf = (ByteBuffer) key.attachment();
                    buf.flip();
                    SocketChannel clientChannel = (SocketChannel) key.channel();

                    clientChannel.write(buf);

                    if (!buf.hasRemaining()) {
                        key.interestOps(OP_READ);
                    }
                    buf.compact();
                }
            }
        }
    }
}

本文由 yongshun 發表于個人博客, 采用署名-非商業性使用-相同方式共享 3.0 中國大陸許可協議.
非商業轉載請注明作者及出處. 商業轉載請聯系作者本人
Email: yongshun1228@gmail .com
本文標題為: Java NIO 的前生今世 之四 NIO Selector 詳解
本文鏈接為: segmentfault.com/a/1190000006824196