一、单线程模式

1.1 模型图

image.png

1.2 代码实现

  1. public class Main {
  3.     public static void main(String[] args) throws IOException {
  4.         TCPReactor tcpReactor = new TCPReactor(9090);
  5.         tcpReactor.run();
  6.     }
  8. }
  10. public class TCPReactor implements Runnable {
  12.     private ServerSocketChannel serverSocketChannel;
  13.     private Selector selector;
  15.     public TCPReactor(int port) throws IOException {
  16.         selector = Selector.open();
  17.         serverSocketChannel = ServerSocketChannel.open();
  18.         ServerSocket socket = serverSocketChannel.socket();
  19.         socket.bind(new InetSocketAddress(port)); // 绑定端口
  20.         serverSocketChannel.configureBlocking(false); // 设置非阻塞
  21.         SelectionKey selectionKey = serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT); // 处理 accept 事件
  22.         selectionKey.attach(new Acceptor(selector, serverSocketChannel)); // 附件一个对象
  23.     }
  25.     @Override
  26.     public void run() {
  27.         while (!Thread.interrupted()) {
  28.             System.out.println("Waiting for new event on port: " + serverSocketChannel.socket().getLocalPort());
  29.             try {
  30.                 if (selector.select() == 0) {
  31.                     continue;
  32.                 }
  33.             } catch (IOException e) {
  34.                 e.printStackTrace();
  35.             }
  37.             Set<SelectionKey> selectionKeys = selector.selectedKeys();
  38.             Iterator<SelectionKey> iterator = selectionKeys.iterator();
  39.             while (iterator.hasNext()) {
  40.                 SelectionKey selectionKey = iterator.next();
  41.                 dispatch(selectionKey);
  42.                 iterator.remove();
  43.             }
  44.         }
  45.     }
  47.     private void dispatch(SelectionKey selectionKey) {
  48.         // 第一次是 accept 事件（Acceptor）
  49.         // 后面就有可能是 read 事件（TCPHandler），根据需要进行 dispatch
  50.         Runnable r = (Runnable) selectionKey.attachment();
  51.         if (r != null) {
  52.             r.run();
  53.         }
  54.     }
  56. }
  58. public class Acceptor implements Runnable {
  60.     Selector selector;
  61.     ServerSocketChannel serverSocketChannel;
  63.     public Acceptor(Selector selector, ServerSocketChannel serverSocketChannel) {
  64.         this.selector = selector;
  65.         this.serverSocketChannel = serverSocketChannel;
  66.     }
  68.     @Override
  69.     public void run() {
  70.         try {
  71.             SocketChannel socketChannel = serverSocketChannel.accept(); // 接收客户端请求
  72.             System.out.println(socketChannel.socket().getRemoteSocketAddress().toString() + " is connected");
  73.             if (socketChannel != null) {
  74.                 socketChannel.configureBlocking(false); // 设置非阻塞
  75.                 SelectionKey selectionKey = socketChannel.register(selector, SelectionKey.OP_READ);
  76.                 selectionKey.attach(new TCPHandler(selectionKey, socketChannel));
  77.                 selector.wakeup(); // 让一个阻塞的 selector 操作立即返回
  78.             }
  79.         } catch (IOException e) {
  80.             e.printStackTrace();
  81.         }
  82.     }
  83. }
  85. public class TCPHandler implements Runnable {
  87.     SelectionKey selectionKey;
  88.     SocketChannel socketChannel;
  89.     int state = 0;
  91.     public TCPHandler(SelectionKey selectionKey, SocketChannel socketChannel) {
  92.         this.selectionKey = selectionKey;
  93.         this.socketChannel = socketChannel;
  94.         state = 0; // 初始状态设置未 READING
  95.     }
  97.     @Override
  98.     public void run() {
  99.         try {
  100.             if (state == 0) {
  101.                 read(); // 读取数据
  102.             } else {
  103.                 send(); // 发送数据
  104.             }
  105.         } catch (Exception ex) {
  106.             System.out.println("[warning] A client has been closed");
  107.             closeChannel();
  108.         }
  109.     }
  111.     private void closeChannel() {
  112.         try {
  113.             socketChannel.close();
  114.         } catch (IOException e) {
  115.             e.printStackTrace();
  116.         }
  117.     }
  119.     private synchronized void read() throws IOException {
  120.         ByteBuffer buffer = ByteBuffer.allocate(1024);
  122.         int len = socketChannel.read(buffer);
  123.         if (len == -1) {
  124.             System.out.println("[warning] A client has been closed");
  125.             closeChannel();
  126.             return;
  127.         }
  129.         String content = new String(buffer.array());
  130.         if (!content.equals(" ")) {
  131.             System.out.println(socketChannel.socket().getRemoteSocketAddress().toString() + " > " + content);
  132.             process(content);
  133.             state = 1; // 改变状态
  134.             selectionKey.interestOps(SelectionKey.OP_WRITE); // 通过 key 改变通道注册事件
  135.             selectionKey.selector().wakeup(); // 唤醒 selector
  136.         }
  137.     }
  139.     private void send() throws IOException {
  140.         String response = "Your message has send to " + socketChannel.socket().getRemoteSocketAddress().toString() + "\r\n";
  141.         ByteBuffer buffer = ByteBuffer.wrap(response.getBytes());
  142.         while (buffer.hasRemaining()) {
  143.             socketChannel.write(buffer);
  144.         }
  145.         state = 0;
  146.         selectionKey.interestOps(SelectionKey.OP_READ); // 通过 key 改变通道注册事件
  147.         selectionKey.selector().wakeup(); // 唤醒 selector
  148.     }
  150.     private void process(String content) {
  151.         try {
  152.             TimeUnit.SECONDS.sleep(2);
  153.         } catch (InterruptedException e) {
  154.             e.printStackTrace();
  155.         }
  156.     }
  158. }

二、多线程模型

2.1 模型图

image.png

2.2 代码实现

  1. public class Main {
  2.     public static void main(String[] args) throws IOException {
  3.         TCPReactor tcpReactor = new TCPReactor(9090);
  4.         tcpReactor.run();
  5.     }
  6. }
  8. public class TCPReactor implements Runnable {
  10.     private ServerSocketChannel serverSocketChannel;
  11.     private Selector selector;
  13.     public TCPReactor(int port) throws IOException {
  14.         selector = Selector.open();
  15.         serverSocketChannel = ServerSocketChannel.open();
  16.         ServerSocket socket = serverSocketChannel.socket();
  17.         socket.bind(new InetSocketAddress(port)); // 绑定端口
  18.         serverSocketChannel.configureBlocking(false); // 设置非阻塞
  19.         SelectionKey selectionKey = serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT); // 处理 accept 事件
  20.         selectionKey.attach(new Acceptor(selector, serverSocketChannel)); // 附件一个对象
  21.     }
  23.     @Override
  24.     public void run() {
  25.         while (!Thread.interrupted()) {
  26.             System.out.println("Waiting for new event on port: " + serverSocketChannel.socket().getLocalPort());
  27.             try {
  28.                 if (selector.select() == 0) {
  29.                     continue;
  30.                 }
  31.             } catch (IOException e) {
  32.                 e.printStackTrace();
  33.             }
  35.             Set<SelectionKey> selectionKeys = selector.selectedKeys();
  36.             Iterator<SelectionKey> iterator = selectionKeys.iterator();
  37.             while (iterator.hasNext()) {
  38.                 SelectionKey selectionKey = iterator.next();
  39.                 dispatch(selectionKey);
  40.                 iterator.remove();
  41.             }
  42.         }
  43.     }
  45.     private void dispatch(SelectionKey selectionKey) {
  46.         // 第一次是 accept 事件（Acceptor）
  47.         // 后面就有可能是 read 事件（TCPHandler），根据需要进行 dispatch
  48.         Runnable r = (Runnable) selectionKey.attachment();
  49.         if (r != null) {
  50.             r.run();
  51.         }
  52.     }
  54. }
  56. public class TCPHandler implements Runnable {
  58.     SelectionKey selectionKey;
  59.     SocketChannel socketChannel;
  60.     HandlerState state;
  61.     private static final int THREAD_COUNTING = 10;
  62.     static ThreadPoolExecutor pool = new ThreadPoolExecutor(THREAD_COUNTING, THREAD_COUNTING,
  63.             120, TimeUnit.SECONDS, new LinkedBlockingDeque<>());
  65.     public TCPHandler(SelectionKey selectionKey, SocketChannel socketChannel) {
  66.         this.selectionKey = selectionKey;
  67.         this.socketChannel = socketChannel;
  68.         state = new ReadState(); // 初始状态设置未 READING
  69.         pool.setMaximumPoolSize(32);
  70.     }
  72.     @Override
  73.     public void run() {
  74.         try {
  75.             state.handle(this, selectionKey, socketChannel, pool);
  76.         } catch (Exception ex) {
  77.             System.out.println("[warning] A client has been closed");
  78.             closeChannel();
  79.         }
  80.     }
  82.     public void closeChannel() {
  83.         try {
  84.             socketChannel.close();
  85.         } catch (IOException e) {
  86.             e.printStackTrace();
  87.         }
  88.     }
  90.     public void setState(HandlerState state) {
  91.         this.state = state;
  92.     }
  94. }
  96. public interface HandlerState {
  98.     void changeState(TCPHandler handler);
  100.     void handle(TCPHandler handler, SelectionKey key, SocketChannel sc, ThreadPoolExecutor pool) throws IOException;
  102. }
  104. public class ReadState implements HandlerState {
  106.     private SelectionKey sk;
  108.     @Override
  109.     public void changeState(TCPHandler handler) {
  110.         handler.setState(new WorkState());
  111.     }
  113.     @Override
  114.     public void handle(TCPHandler handler, SelectionKey sk, SocketChannel sc, ThreadPoolExecutor pool) throws IOException {
  115.         this.sk = sk;
  116.         byte[] arr = new byte[1024];
  117.         ByteBuffer buf = ByteBuffer.wrap(arr);
  119.         int numBytes = sc.read(buf); // 讀取字符串
  120.         if (numBytes == -1) {
  121.             System.out.println("[Warning!] A client has been closed.");
  122.             handler.closeChannel();
  123.             return;
  124.         }
  125.         String str = new String(arr);
  126.         if (!str.equals(" ")) {
  127.             handler.setState(new WorkState()); // 改變狀態(READING->WORKING)
  128.             pool.execute(new WorkerThread(handler, str)); // do process in worker thread
  129.             System.out.println(sc.socket().getRemoteSocketAddress().toString() + " > " + str);
  130.         }
  131.     }
  133.     /*
  134.      * 執行邏輯處理之函數
  135.      */
  136.     synchronized void process(TCPHandler h, String str) {
  137.         try {
  138.             TimeUnit.SECONDS.sleep(2);
  139.         } catch (InterruptedException e) {
  140.             e.printStackTrace();
  141.         }
  142.         h.setState(new WriteState()); // 改變狀態(WORKING->SENDING)
  143.         this.sk.interestOps(SelectionKey.OP_WRITE); // 通過key改變通道註冊的事件
  144.         this.sk.selector().wakeup(); // 使一個阻塞住的selector操作立即返回
  145.     }
  147.     class WorkerThread implements Runnable {
  149.         TCPHandler h;
  150.         String str;
  152.         public WorkerThread(TCPHandler h, String str) {
  153.             this.h = h;
  154.             this.str = str;
  155.         }
  157.         @Override
  158.         public void run() {
  159.             process(h, str);
  160.         }
  162.     }
  163. }
  165. public class WorkState implements HandlerState {
  167.     @Override
  168.     public void changeState(TCPHandler h) {
  169.         h.setState(new WriteState());
  170.     }
  172.     @Override
  173.     public void handle(TCPHandler h, SelectionKey sk, SocketChannel sc,
  174.                        ThreadPoolExecutor pool) throws IOException {
  175.     }
  176. }
  178. public class WriteState implements HandlerState {
  180.     @Override
  181.     public void changeState(TCPHandler h) {
  182.         h.setState(new ReadState());
  183.     }
  185.     @Override
  186.     public void handle(TCPHandler h, SelectionKey sk, SocketChannel sc,
  187.                        ThreadPoolExecutor pool) throws IOException {
  188.         // get message from message queue
  190.         String str = "Your message has sent to " + sc.socket().getLocalSocketAddress().toString() + "\r\n";
  191.         ByteBuffer buf = ByteBuffer.wrap(str.getBytes()); // wrap自動把buf的position設為0，所以不需要再flip()
  193.         while (buf.hasRemaining()) {
  194.             sc.write(buf); // 回傳給client回應字符串，發送buf的position位置 到limit位置為止之間的內容
  195.         }
  197.         h.setState(new ReadState()); // 改變狀態(SENDING->READING)
  198.         sk.interestOps(SelectionKey.OP_READ); // 通過key改變通道註冊的事件
  199.         sk.selector().wakeup(); // 使一個阻塞住的selector操作立即返回
  200.     }
  201. }

三、主从模型

3.1 模型图

image.png

3.2 代码实现

public class Main {

    public static void main(String[] args) {
        try {
            TCPReactor reactor = new TCPReactor(9090);
            // reactor.run();
            Thread thread = new Thread(reactor);
            thread.start();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

}

public class TCPReactor implements Runnable {

    private final ServerSocketChannel ssc;
    private final Selector selector; // mainReactor用的selector

    public TCPReactor(int port) throws IOException {
        selector = Selector.open();
        ssc = ServerSocketChannel.open();
        ssc.configureBlocking(false); // 設置ServerSocketChannel為非阻塞

        Acceptor acceptor = new Acceptor(ssc);

        SelectionKey sk = ssc.register(selector, SelectionKey.OP_ACCEPT); // ServerSocketChannel向selector註冊一個OP_ACCEPT事件，然後返回該通道的key
        sk.attach(acceptor); // 給定key一個附加的Acceptor對象

        InetSocketAddress addr = new InetSocketAddress(port);
        ssc.socket().bind(addr); // 在ServerSocketChannel綁定監聽端口
    }

    @Override
    public void run() {
        while (!Thread.interrupted()) { // 在線程被中斷前持續運行
            System.out.println("mainReactor waiting for new event on port: " + ssc.socket().getLocalPort() + "...");
            try {
                if (selector.select() == 0) // 若沒有事件就緒則不往下執行
                    continue;
            } catch (IOException e) {
                e.printStackTrace();
            }
            Set<SelectionKey> selectedKeys = selector.selectedKeys(); // 取得所有已就緒事件的key集合
            Iterator<SelectionKey> it = selectedKeys.iterator();
            while (it.hasNext()) {
                dispatch(it.next()); // 根據事件的key進行調度
                it.remove();
            }
        }
    }

    /*
     * name: dispatch(SelectionKey key)
     * description: 調度方法，根據事件綁定的對象開新線程
     */
    private void dispatch(SelectionKey key) {
        Runnable r = (Runnable) (key.attachment()); // 根據事件之key綁定的對象開新線程
        if (r != null)
            r.run();
    }

}

// 接受連線請求線程
package org.wesoft.study.reactor.masterandslave;

import java.io.IOException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;

public class Acceptor implements Runnable {

    private final ServerSocketChannel ssc; // mainReactor監聽的socket通道
    private final int cores = Runtime.getRuntime().availableProcessors(); // 取得CPU核心數
    private final Selector[] selectors = new Selector[cores]; // 創建核心數個selector給subReactor用
    private int selIdx = 0; // 當前可使用的subReactor索引
    private TCPSubReactor[] r = new TCPSubReactor[cores]; // subReactor線程
    private Thread[] t = new Thread[cores]; // subReactor線程

    public Acceptor(ServerSocketChannel ssc) throws IOException {
        this.ssc = ssc;
        // 創建多個selector以及多個subReactor線程
        for (int i = 0; i < cores; i++) {
            selectors[i] = Selector.open();
            r[i] = new TCPSubReactor(selectors[i], ssc, i);
            t[i] = new Thread(r[i]);
            t[i].start();
        }
    }

    @Override
    public synchronized void run() {
        try {
            SocketChannel sc = ssc.accept(); // 接受client連線請求
            System.out.println(sc.socket().getRemoteSocketAddress().toString() + " is connected.");

            if (sc != null) {
                sc.configureBlocking(false); // 設置為非阻塞
                r[selIdx].setRestart(true); // 暫停線程
                selectors[selIdx].wakeup(); // 使一個阻塞住的selector操作立即返回
                SelectionKey sk = sc.register(selectors[selIdx],
                        SelectionKey.OP_READ); // SocketChannel向selector[selIdx]註冊一個OP_READ事件，然後返回該通道的key
                selectors[selIdx].wakeup(); // 使一個阻塞住的selector操作立即返回
                r[selIdx].setRestart(false); // 重啟線程
                sk.attach(new TCPHandler(sk, sc)); // 給定key一個附加的TCPHandler對象
                if (++selIdx == selectors.length)
                    selIdx = 0;
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

}

public class TCPSubReactor implements Runnable {

    private final ServerSocketChannel ssc;
    private final Selector selector;
    private boolean restart = false;
    int num;

    public TCPSubReactor(Selector selector, ServerSocketChannel ssc, int num) {
        this.ssc = ssc;
        this.selector = selector;
        this.num = num;
    }

    @Override
    public void run() {
        while (!Thread.interrupted()) { // 在線程被中斷前持續運行  
            //System.out.println("ID:" + num  
            //      + " subReactor waiting for new event on port: "  
            //      + ssc.socket().getLocalPort() + "...");  
            System.out.println("waiting for restart");
            while (!Thread.interrupted() && !restart) { // 在線程被中斷前以及被指定重啟前持續運行  
                try {
                    if (selector.select() == 0)
                        continue; // 若沒有事件就緒則不往下執行  
                } catch (IOException e) {
                    e.printStackTrace();
                }
                Set<SelectionKey> selectedKeys = selector.selectedKeys(); // 取得所有已就緒事件的key集合  
                Iterator<SelectionKey> it = selectedKeys.iterator();
                while (it.hasNext()) {
                    dispatch((SelectionKey) (it.next())); // 根據事件的key進行調度  
                    it.remove();
                }
            }
        }
    }

    /*
     * name: dispatch(SelectionKey key) description: 調度方法，根據事件綁定的對象開新線程
     */
    private void dispatch(SelectionKey key) {
        Runnable r = (Runnable) (key.attachment()); // 根據事件之key綁定的對象開新線程  
        if (r != null)
            r.run();
    }

    public void setRestart(boolean restart) {
        this.restart = restart;
    }
}

public class TCPHandler implements Runnable {

    private final SelectionKey sk;
    private final SocketChannel sc;
    private static final int THREAD_COUNTING = 10;
    private static ThreadPoolExecutor pool = new ThreadPoolExecutor(THREAD_COUNTING, THREAD_COUNTING,
            120, TimeUnit.SECONDS, new LinkedBlockingQueue<>()); // 線程池
    HandlerState state; // 以狀態模式實現Handler

    public TCPHandler(SelectionKey sk, SocketChannel sc) {
        this.sk = sk;
        this.sc = sc;
        state = new ReadState(); // 初始狀態設定為READING
        pool.setMaximumPoolSize(32); // 設置線程池最大線程數
    }

    @Override
    public void run() {
        try {
            state.handle(this, sk, sc, pool);
        } catch (IOException e) {
            System.out.println("[Warning!] A client has been closed.");
            closeChannel();
        }
    }

    public void closeChannel() {
        try {
            sk.cancel();
            sc.close();
        } catch (IOException e1) {
            e1.printStackTrace();
        }
    }

    public void setState(HandlerState state) {
        this.state = state;
    }
}

public interface HandlerState {

    void changeState(TCPHandler handler);

    void handle(TCPHandler handler, SelectionKey key, SocketChannel sc, ThreadPoolExecutor pool) throws IOException;

}

public class ReadState implements HandlerState {

    private SelectionKey sk;

    @Override
    public void changeState(TCPHandler handler) {
        handler.setState(new WorkState());
    }

    @Override
    public void handle(TCPHandler handler, SelectionKey sk, SocketChannel sc, ThreadPoolExecutor pool) throws IOException {
        this.sk = sk;
        byte[] arr = new byte[1024];
        ByteBuffer buf = ByteBuffer.wrap(arr);

        int numBytes = sc.read(buf); // 讀取字符串
        if (numBytes == -1) {
            System.out.println("[Warning!] A client has been closed.");
            handler.closeChannel();
            return;
        }
        String str = new String(arr);
        if (!str.equals(" ")) {
            handler.setState(new WorkState()); // 改變狀態(READING->WORKING)
            pool.execute(new WorkerThread(handler, str)); // do process in worker thread
            System.out.println(sc.socket().getRemoteSocketAddress().toString() + " > " + str);
        }
    }

    /*
     * 執行邏輯處理之函數
     */
    synchronized void process(TCPHandler h, String str) {
        try {
            TimeUnit.SECONDS.sleep(2);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        h.setState(new WriteState()); // 改變狀態(WORKING->SENDING)
        this.sk.interestOps(SelectionKey.OP_WRITE); // 通過key改變通道註冊的事件
        this.sk.selector().wakeup(); // 使一個阻塞住的selector操作立即返回
    }

    class WorkerThread implements Runnable {

        TCPHandler h;
        String str;

        public WorkerThread(TCPHandler h, String str) {
            this.h = h;
            this.str = str;
        }

        @Override
        public void run() {
            process(h, str);
        }

    }
}

public class WriteState implements HandlerState {

    @Override
    public void changeState(TCPHandler h) {
        h.setState(new ReadState());
    }

    @Override
    public void handle(TCPHandler h, SelectionKey sk, SocketChannel sc,
                       ThreadPoolExecutor pool) throws IOException {
        // get message from message queue

        String str = "Your message has sent to " + sc.socket().getLocalSocketAddress().toString() + "\r\n";
        ByteBuffer buf = ByteBuffer.wrap(str.getBytes()); // wrap自動把buf的position設為0，所以不需要再flip()

        while (buf.hasRemaining()) {
            sc.write(buf); // 回傳給client回應字符串，發送buf的position位置 到limit位置為止之間的內容
        }

        h.setState(new ReadState()); // 改變狀態(SENDING->READING)
        sk.interestOps(SelectionKey.OP_READ); // 通過key改變通道註冊的事件
        sk.selector().wakeup(); // 使一個阻塞住的selector操作立即返回
    }
}