Scalable IO in Java

Scalable IO in Java

http://gee.cs.oswego.edu/dl/cpjslides/nio.pdf

基本上全部的網絡處理程序都有如下基本的處理過程:
Read request
Decode request
Process service
Encode reply
Send reply

Classic Service Designs

簡單的代碼實現：

class Server implements Runnable {
    public void run() {
        try {
            ServerSocket ss = new ServerSocket(PORT);
            while (!Thread.interrupted())
            new Thread(new Handler(ss.accept())).start(); //建立新線程來handle
            // or, single-threaded, or a thread pool
        } catch (IOException ex) { /* ... */ }
    }
    
    static class Handler implements Runnable {
        final Socket socket;
        Handler(Socket s) { socket = s; }
        public void run() {
            try {
                byte[] input = new byte[MAX_INPUT];
                socket.getInputStream().read(input);
                byte[] output = process(input);
                socket.getOutputStream().write(output);
            } catch (IOException ex) { /* ... */ }
        }       
        private byte[] process(byte[] cmd) { /* ... */ }
    }
}

對於每個請求都分發給一個線程，每一個線程中都獨自處理上面的流程。

這種模型因爲IO在阻塞時會一直等待，所以在用戶負載增長時，性能降低的很是快。

server致使阻塞的緣由：

一、serversocket的accept方法，阻塞等待client鏈接，直到client鏈接成功。

二、線程從socket inputstream讀入數據，會進入阻塞狀態，直到所有數據讀完。

三、線程向socket outputstream寫入數據，會阻塞直到所有數據寫完。

client致使阻塞的緣由：

一、client創建鏈接時會阻塞，直到鏈接成功。

二、線程從socket輸入流讀入數據，若是沒有足夠數據讀完會進入阻塞狀態，直到有數據或者讀到輸入流末尾。

三、線程從socket輸出流寫入數據，直到輸出全部數據。

四、socket.setsolinger()設置socket的延遲時間，當socket關閉時，會進入阻塞狀態，直到所有數據都發送完或者超時。

改進：採用基於事件驅動的設計，當有事件觸發時，纔會調用處理器進行數據處理。

Basic Reactor Design

 代碼實現：

class Reactor implements Runnable { 
    final Selector selector;
    final ServerSocketChannel serverSocket;
    Reactor(int port) throws IOException { //Reactor初始化
        selector = Selector.open();
        serverSocket = ServerSocketChannel.open();
        serverSocket.socket().bind(new InetSocketAddress(port));
        serverSocket.configureBlocking(false); //非阻塞
        SelectionKey sk = serverSocket.register(selector, SelectionKey.OP_ACCEPT); //分步處理,第一步,接收accept事件
        sk.attach(new Acceptor()); //attach callback object, Acceptor
    }
    
    public void run() { 
        try {
            while (!Thread.interrupted()) {
                selector.select();
                Set selected = selector.selectedKeys();
                Iterator it = selected.iterator();
                while (it.hasNext())
                    dispatch((SelectionKey)(it.next()); //Reactor負責dispatch收到的事件
                selected.clear();
            }
        } catch (IOException ex) { /* ... */ }
    }
    
    void dispatch(SelectionKey k) {
        Runnable r = (Runnable)(k.attachment()); //調用以前註冊的callback對象
        if (r != null)
            r.run();
    }
    
    class Acceptor implements Runnable { // inner
        public void run() {
            try {
                SocketChannel c = serverSocket.accept();
                if (c != null)
                new Handler(selector, c);
            }
            catch(IOException ex) { /* ... */ }
        }
    }
}

final class Handler implements Runnable {
    final SocketChannel socket;
    final SelectionKey sk;
    ByteBuffer input = ByteBuffer.allocate(MAXIN);
    ByteBuffer output = ByteBuffer.allocate(MAXOUT);
    static final int READING = 0, SENDING = 1;
    int state = READING;
    
    Handler(Selector sel, SocketChannel c) throws IOException {
        socket = c; c.configureBlocking(false);
        // Optionally try first read now
        sk = socket.register(sel, 0);
        sk.attach(this); //將Handler做爲callback對象
        sk.interestOps(SelectionKey.OP_READ); //第二步,接收Read事件
        sel.wakeup();
    }
    boolean inputIsComplete() { /* ... */ }
    boolean outputIsComplete() { /* ... */ }
    void process() { /* ... */ }
    
    public void run() {
        try {
            if (state == READING) read();
            else if (state == SENDING) send();
        } catch (IOException ex) { /* ... */ }
    }
    
    void read() throws IOException {
        socket.read(input);
        if (inputIsComplete()) {
            process();
            state = SENDING;
            // Normally also do first write now
            sk.interestOps(SelectionKey.OP_WRITE); //第三步,接收write事件
        }
    }
    void send() throws IOException {
        socket.write(output);
        if (outputIsComplete()) sk.cancel(); //write完就結束了, 關閉select key
    }
}

//上面 的實現用Handler來同時處理Read和Write事件, 因此裏面出現狀態判斷
//咱們能夠用State-Object pattern來更優雅的實現
class Handler { // ...
    public void run() { // initial state is reader
        socket.read(input);
        if (inputIsComplete()) {
            process();
            sk.attach(new Sender());  //狀態遷移, Read後變成write, 用Sender做爲新的callback對象
            sk.interest(SelectionKey.OP_WRITE);
            sk.selector().wakeup();
        }
    }
    class Sender implements Runnable {
        public void run(){ // ...
            socket.write(output);
            if (outputIsComplete()) sk.cancel();
        }
    }
}

這裏用到了Reactor模式。

關於Reactor模式的一些概念：

Reactor：負責響應IO事件，當檢測到一個新的事件，將其發送給相應的Handler去處理。

Handler：負責處理非阻塞的行爲，標識系統管理的資源；同時將handler與事件綁定。

Reactor爲單個線程，須要處理accept鏈接，同時發送請求處處理器中。

因爲只有單個線程，因此處理器中的業務須要可以快速處理完。

改進：使用多線程處理業務邏輯。

Worker Thread Pools

 參考代碼：

class Handler implements Runnable {
    // uses util.concurrent thread pool
    static PooledExecutor pool = new PooledExecutor(...);
    static final int PROCESSING = 3;
    // ...
    synchronized void read() { // ...
        socket.read(input);
        if (inputIsComplete()) {
            state = PROCESSING;
            pool.execute(new Processer()); //使用線程pool異步執行
        }
    }
    
    synchronized void processAndHandOff() {
        process();
        state = SENDING; // or rebind attachment
        sk.interest(SelectionKey.OP_WRITE); //process完,開始等待write事件
    }
    
    class Processer implements Runnable {
        public void run() { processAndHandOff(); }
    }
}

將處理器的執行放入線程池，多線程進行業務處理。但Reactor仍爲單個線程。

繼續改進：對於多個CPU的機器，爲充分利用系統資源，將Reactor拆分爲兩部分。

Using Multiple Reactors

參考代碼：

Selector[] selectors; //subReactors集合, 一個selector表明一個subReactor
int next = 0;
class Acceptor { // ...
    public synchronized void run() { ...
        Socket connection = serverSocket.accept(); //主selector負責accept
        if (connection != null)
            new Handler(selectors[next], connection); //選個subReactor去負責接收到的connection
        if (++next == selectors.length) next = 0;
    }
}

mainReactor負責監聽鏈接，accept鏈接給subReactor處理，爲何要單獨分一個Reactor來處理監聽呢？由於像TCP這樣須要通過3次握手才能創建鏈接，這個創建鏈接的過程也是要耗時間和資源的，單獨分一個Reactor來處理，能夠提升性能。