摘要：允許一個(gè)單一的線程來(lái)操作多個(gè)如果我們的應(yīng)用程序中使用了多個(gè)那么使用很方便的實(shí)現(xiàn)這樣的目的但是因?yàn)樵谝粋€(gè)線程中使用了多個(gè)因此也會(huì)造成了每個(gè)傳輸效率的降低使用的圖解如下為了使用我們首先需要將注冊(cè)到中隨后調(diào)用的方法這個(gè)方法會(huì)阻塞直到注冊(cè)在中的發(fā)送

Selector 允許一個(gè)單一的線程來(lái)操作多個(gè) Channel. 如果我們的應(yīng)用程序中使用了多個(gè) Channel, 那么使用 Selector 很方便的實(shí)現(xiàn)這樣的目的, 但是因?yàn)樵谝粋€(gè)線程中使用了多個(gè) Channel, 因此也會(huì)造成了每個(gè) Channel 傳輸效率的降低.
使用 Selector 的圖解如下:

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

通過(guò) Selector.open()方法, 我們可以創(chuàng)建一個(gè)選擇器:

Selector selector = Selector.open();

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

channel.configureBlocking(false);

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

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

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

Connect, 即連接事件(TCP 連接), 對(duì)應(yīng)于SelectionKey.OP_CONNECT

Accept, 即確認(rèn)事件, 對(duì)應(yīng)于SelectionKey.OP_ACCEPT

Read, 即讀事件, 對(duì)應(yīng)于SelectionKey.OP_READ, 表示 buffer 可讀.

Write, 即寫事件, 對(duì)應(yīng)于SelectionKey.OP_WRITE, 表示 buffer 可寫.

一個(gè) Channel發(fā)出一個(gè)事件也可以稱為 對(duì)于某個(gè)事件, Channel 準(zhǔn)備好了. 因此一個(gè) Channel 成功連接到了另一個(gè)服務(wù)器也可以被稱為 connect ready.
我們可以使用或運(yùn)算|來(lái)組合多個(gè)事件, 例如:

int interestSet = SelectionKey.OP_READ | SelectionKey.OP_WRITE;    

注意, 一個(gè) Channel 僅僅可以被注冊(cè)到一個(gè) Selector 一次, 如果將 Channel 注冊(cè)到 Selector 多次, 那么其實(shí)就是相當(dāng)于更新 SelectionKey 的 interest set. 例如:

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

上面的 channel 注冊(cè)到同一個(gè) Selector 兩次了, 那么第二次的注冊(cè)其實(shí)就是相當(dāng)于更新這個(gè) Channel 的 interest set 為 SelectionKey.OP_READ | SelectionKey.OP_WRITE.

如上所示, 當(dāng)我們使用 register 注冊(cè)一個(gè) Channel 時(shí), 會(huì)返回一個(gè) SelectionKey 對(duì)象, 這個(gè)對(duì)象包含了如下內(nèi)容:

interest set, 即我們感興趣的事件集, 即在調(diào)用 register 注冊(cè) channel 時(shí)所設(shè)置的 interest set.

ready set

channel

selector

attached object, 可選的附加對(duì)象

我們可以通過(guò)如下方式獲取 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 所準(zhǔn)備好了的操作.
我們可以像判斷 interest set 一樣操作 Ready set, 但是我們還可以使用如下方法進(jìn)行判斷:

int readySet = selectionKey.readyOps();

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

我們可以通過(guò) SelectionKey 獲取相對(duì)應(yīng)的 Channel 和 Selector:

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

我們可以在selectionKey中附加一個(gè)對(duì)象:

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

或者在注冊(cè)時(shí)直接附加:

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

我們可以通過(guò) Selector.select()方法獲取對(duì)某件事件準(zhǔn)備好了的 Channel, 即如果我們?cè)谧?cè) Channel 時(shí), 對(duì)其的可寫事件感興趣, 那么當(dāng) select()返回時(shí), 我們就可以獲取 Channel 了.

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

如果 select()方法返回值表示有多個(gè) Channel 準(zhǔn)備好了, 那么我們可以通過(guò) Selected key set 訪問(wèn)這個(gè) 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();
}

注意, 在每次迭代時(shí), 我們都調(diào)用 "keyIterator.remove()" 將這個(gè) key 從迭代器中刪除, 因?yàn)?select() 方法僅僅是簡(jiǎn)單地將就緒的 IO 操作放到 selectedKeys 集合中, 因此如果我們從 selectedKeys 獲取到一個(gè) key, 但是沒(méi)有將它刪除, 那么下一次 select 時(shí), 這個(gè) key 所對(duì)應(yīng)的 IO 事件還在 selectedKeys 中.
例如此時(shí)我們收到 OP_ACCEPT 通知, 然后我們進(jìn)行相關(guān)處理, 但是并沒(méi)有將這個(gè) Key 從 SelectedKeys 中刪除, 那么下一次 select() 返回時(shí) 我們還可以在 SelectedKeys 中獲取到 OP_ACCEPT 的 key.
注意, 我們可以動(dòng)態(tài)更改 SekectedKeys 中的 key 的 interest set. 例如在 OP_ACCEPT 中, 我們可以將 interest set 更新為 OP_READ, 這樣 Selector 就會(huì)將這個(gè) Channel 的 讀 IO 就緒事件包含進(jìn)來(lái)了.

通過(guò) Selector.open() 打開一個(gè) Selector.

將 Channel 注冊(cè)到 Selector 中, 并設(shè)置需要監(jiān)聽的事件(interest set)

不斷重復(fù):

調(diào)用 select() 方法

調(diào)用 selector.selectedKeys() 獲取 selected keys

迭代每個(gè) selected key:

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

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

*根據(jù)需要更改 selected key 的監(jiān)聽事件.

*將已經(jīng)處理過(guò)的 key 從 selected keys 集合中刪除.

當(dāng)調(diào)用了 Selector.close()方法時(shí), 我們其實(shí)是關(guān)閉了 Selector 本身并且將所有的 SelectionKey 失效, 但是并不會(huì)關(guān)閉 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 {
        // 打開服務(wù)端 Socket
        ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();

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

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

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

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

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

            while (keyIterator.hasNext()) {

                SelectionKey key = keyIterator.next();

                // 當(dāng)獲取一個(gè) SelectionKey 后, 就要將它刪除, 表示我們已經(jīng)對(duì)這個(gè) IO 事件進(jìn)行了處理.
                keyIterator.remove();

                if (key.isAcceptable()) {
                    // 當(dāng) OP_ACCEPT 事件到來(lái)時(shí), 我們就有從 ServerSocketChannel 中獲取一個(gè) 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 到來(lái)時(shí), 再將這個(gè) Channel 的 OP_READ 注冊(cè)到 Selector 中.
                    // 注意, 這里我們?nèi)绻麤](méi)有設(shè)置 OP_READ 的話, 即 interest set 仍然是 OP_CONNECT 的話, 那么 select 方法會(huì)一直直接返回.
                    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 發(fā)表于個(gè)人博客, 采用署名-非商業(yè)性使用-相同方式共享 3.0 中國(guó)大陸許可協(xié)議.
非商業(yè)轉(zhuǎn)載請(qǐng)注明作者及出處. 商業(yè)轉(zhuǎn)載請(qǐng)聯(lián)系作者本人
Email: yongshun1228@gmail .com
本文標(biāo)題為: Java NIO 的前生今世 之四 NIO Selector 詳解
本文鏈接為: segmentfault.com/a/1190000006824196