小師妹學JavaIO之:用Selector來發好人卡

時間 2020-06-15

標籤師妹 javaio selector 好人欄目 Java 简体版

原文原文鏈接

簡介

NIO有三寶:Buffer,Channel，Selector少不了。本文將會介紹NIO三件套中的最後一套Selector，並在理解Selector的基礎上，協助小師妹發一張好人卡。咱們開始吧。java

Selector介紹

小師妹：F師兄，最近個人桃花有點旺，好幾個師兄莫名其妙的跟我打招呼，但是我一心向着工做，不想談論這些事情。畢竟先有事業纔有家嘛。我又很差直接拒絕，有沒有什麼比較隱晦的方法來讓他們放棄這個想法？程序員

更多內容請訪問 www.flydean.com

這個問題，我沉思了大約0.001秒，因而給出了答案：給他們發張好人卡吧，應該就不會再來糾纏你了。spring

小師妹：F師兄，若是給他們發無缺人卡尚未用呢？服務器

那就只能切斷跟他們的聯繫了，來個一刀兩斷。哈哈。socket

這樣吧，小師妹你最近不是在學NIO嗎？恰好咱們能夠用Selector來模擬一下發好人卡的過程。ide

假如你的志偉師兄和子丹師兄想跟你創建聯繫，每一個人都想跟你創建一個溝統統道，那麼你就須要建立兩個channel。spring-boot

兩個channel其實還好，若是有多我的都想同時跟你創建聯繫通道，那麼要維持這些通道就須要保持鏈接，從而浪費了資源。區塊鏈

可是創建的這些鏈接並非時時刻刻都有消息在傳輸，因此其實大多數時間這些創建聯繫的通道實際上是浪費的。spa

若是使用Selector就能夠只啓用一個線程來監聽通道的消息變更，這就是Selector。.net

從上面的圖能夠看出，Selector監聽三個不一樣的channel，而後交給一個processor來處理，從而節約了資源。

建立Selector

先看下selector的定義：

public abstract class Selector implements Closeable

Selector是一個abstract類，而且實現了Closeable，表示Selector是能夠被關閉的。

雖然Selector是一個abstract類，可是能夠經過open來簡單的建立：

Selector selector = Selector.open();

若是細看open的實現能夠發現一個頗有趣的現象：

public static Selector open() throws IOException {
        return SelectorProvider.provider().openSelector();
    }

open方法調用的是SelectorProvider中的openSelector方法。

再看下provider的實現：

public SelectorProvider run() {
   if (loadProviderFromProperty())
        return provider;
    if (loadProviderAsService())
        return provider;
      provider = sun.nio.ch.DefaultSelectorProvider.create();
      return provider;
    }
 });

有三種狀況能夠加載一個SelectorProvider，若是系統屬性指定了java.nio.channels.spi.SelectorProvider，那麼從指定的屬性加載。

若是沒有直接指定屬性，則從ServiceLoader來加載。

最後若是都找不到的狀況下，使用默認的DefaultSelectorProvider。

關於ServiceLoader的用法，咱們後面會有專門的文章來說述。這裏先不作多的解釋。

註冊Selector到Channel中

ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
        serverSocketChannel.bind(new InetSocketAddress("localhost", 9527));
        serverSocketChannel.configureBlocking(false);
        serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);

若是是在服務器端，咱們須要先建立一個ServerSocketChannel，綁定Server的地址和端口，而後將Blocking設置爲false。由於咱們使用了Selector，它其實是一個非阻塞的IO。

注意FileChannels是不能使用Selector的，由於它是一個阻塞型IO。

小師妹：F師兄，爲啥FileChannel是阻塞型的呀？作成非阻塞型的不是更快？

小師妹，咱們使用FileChannel的目的是什麼？就是爲了讀文件呀，讀取文件確定是一直讀一直讀，沒有可能讀一會這個channel再讀另一個channel吧，由於對於每一個channel本身來說，在文件沒讀取完以前，都是繁忙狀態，沒有必要在channel中切換。

最後咱們將建立好的Selector註冊到channel中去。

SelectionKey

SelectionKey表示的是咱們但願監聽到的事件。

總的來講，有4種Event：

SelectionKey.OP_READ 表示服務器準備好，能夠從channel中讀取數據。
SelectionKey.OP_WRITE 表示服務器準備好，能夠向channel中寫入數據。
SelectionKey.OP_CONNECT 表示客戶端嘗試去鏈接服務端
SelectionKey.OP_ACCEPT 表示服務器accept一個客戶端的請求

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;

咱們能夠看到上面的4個Event是用位運算來定義的，若是將這個四個event使用或運算合併起來，就獲得了SelectionKey中的interestOps。

和interestOps相似，SelectionKey還有一個readyOps。

一個表示感興趣的操做，一個表示ready的操做。

最後，SelectionKey在註冊的時候，還能夠attach一個Object，好比咱們能夠在這個對象中保存這個channel的id：

SelectionKey key = channel.register(
  selector, SelectionKey.OP_ACCEPT, object);
key.attach(Object);
Object object = key.attachment();

object能夠在register的時候傳入，也能夠調用attach方法。

最後，咱們能夠經過key的attachment方法，得到該對象。

selector 和 SelectionKey

咱們經過selector.select()這個一個blocking操做，來獲取一個ready的channel。

而後咱們經過調用selector.selectedKeys()來獲取到SelectionKey對象。

在SelectionKey對象中，咱們經過判斷ready的event來處理相應的消息。

總的例子

接下來，咱們把以前將的串聯起來，先創建一個小師妹的ChatServer：

public class ChatServer {

    private static String BYE_BYE="再見";

    public static void main(String[] args) throws IOException, InterruptedException {
        Selector selector = Selector.open();
        ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
        serverSocketChannel.bind(new InetSocketAddress("localhost", 9527));
        serverSocketChannel.configureBlocking(false);
        serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
        ByteBuffer byteBuffer = ByteBuffer.allocate(512);

        while (true) {
            selector.select();
            Set<SelectionKey> selectedKeys = selector.selectedKeys();
            Iterator<SelectionKey> iter = selectedKeys.iterator();
            while (iter.hasNext()) {
                SelectionKey selectionKey = iter.next();
                if (selectionKey.isAcceptable()) {
                    register(selector, serverSocketChannel);
                }
                if (selectionKey.isReadable()) {
                    serverResonse(byteBuffer, selectionKey);
                }
                iter.remove();
            }
            Thread.sleep(1000);
        }
    }

    private static void serverResonse(ByteBuffer byteBuffer, SelectionKey selectionKey)
            throws IOException {
        SocketChannel socketChannel = (SocketChannel) selectionKey.channel();
        socketChannel.read(byteBuffer);
        byteBuffer.flip();
        byte[] bytes= new byte[byteBuffer.limit()];
        byteBuffer.get(bytes);
        log.info(new String(bytes).trim());
        if(new String(bytes).trim().equals(BYE_BYE)){
            log.info("說再見不如不見！");
            socketChannel.write(ByteBuffer.wrap("再見".getBytes()));
            socketChannel.close();
        }else {
            socketChannel.write(ByteBuffer.wrap("你是個好人".getBytes()));
        }
        byteBuffer.clear();
    }

    private static void register(Selector selector, ServerSocketChannel serverSocketChannel)
            throws IOException {
        SocketChannel socketChannel = serverSocketChannel.accept();
        socketChannel.configureBlocking(false);
        socketChannel.register(selector, SelectionKey.OP_READ);
    }
}

上面例子有兩點須要注意，咱們在循環遍歷中，當selectionKey.isAcceptable時，表示服務器收到了一個新的客戶端鏈接，這個時候咱們須要調用register方法，再註冊一個OP_READ事件到這個新的SocketChannel中，而後繼續遍歷。

第二，咱們定義了一個stop word，當收到這個stop word的時候，會直接關閉這個client channel。

再看看客戶端的代碼：

public class ChatClient {

    private static SocketChannel socketChannel;
    private static ByteBuffer byteBuffer;

    public static void main(String[] args) throws IOException {

        ChatClient chatClient = new ChatClient();
        String response = chatClient.sendMessage("hello 小師妹！");
        log.info("response is {}", response);
        response = chatClient.sendMessage("能不能？");
        log.info("response is {}", response);
        chatClient.stop();

    }

    public void stop() throws IOException {
        socketChannel.close();
        byteBuffer = null;
    }

    public ChatClient() throws IOException {
        socketChannel = SocketChannel.open(new InetSocketAddress("localhost", 9527));
        byteBuffer = ByteBuffer.allocate(512);
    }

    public String sendMessage(String msg) throws IOException {
        byteBuffer = ByteBuffer.wrap(msg.getBytes());
        String response = null;
        socketChannel.write(byteBuffer);
        byteBuffer.clear();
        socketChannel.read(byteBuffer);
        byteBuffer.flip();
        byte[] bytes= new byte[byteBuffer.limit()];
        byteBuffer.get(bytes);
        response =new String(bytes).trim();
        byteBuffer.clear();
        return response;

    }
}

客戶端代碼沒什麼特別的，須要注意的是Buffer的讀取。

最後輸出結果：

server收到： INFO com.flydean.ChatServer - hello 小師妹！
client收到: INFO com.flydean.ChatClient - response is 你是個好人
server收到： INFO com.flydean.ChatServer - 能不能？
client收到： INFO com.flydean.ChatClient - response is 再見

解釋一下整個流程：志偉跟小師妹創建了一個鏈接，志偉向小師妹打了一個招呼，小師妹給志偉發了一張好人卡。志偉不死心，想繼續糾纏，小師妹回覆再見，而後本身關閉了通道。