reactor模式:主从式reactor

前面两篇文章提到

reactor模式:单线程的reactor模式

reactor模式:多线程的reactor模式

NIO的server模式只有5个阶段,但是,NIO的selectionkey里确实有个accept事件,所以,为了区别,衍生出了主reactor和从reactor

并且,从reactor可以根据服务器的负荷,新增多个从reactor进行请求处理

服务器架构如下图

这个就是完整版的reactor模式的架构图了,目前使用到了reactor模式的框架(如netty),基本用的模式就是这个

代码实现:

 

    // Reactor線程  
    package server;  
      
    import java.io.IOException;  
    import java.net.InetSocketAddress;  
    import java.nio.channels.SelectionKey;  
    import java.nio.channels.Selector;  
    import java.nio.channels.ServerSocketChannel;  
    import java.util.Iterator;  
    import java.util.Set;  
      
    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();  
            InetSocketAddress addr = new InetSocketAddress(port);  
            ssc.socket().bind(addr); // 在ServerSocketChannel綁定監聽端口  
            ssc.configureBlocking(false); // 設置ServerSocketChannel為非阻塞  
            SelectionKey sk = ssc.register(selector, SelectionKey.OP_ACCEPT); // ServerSocketChannel向selector註冊一個OP_ACCEPT事件,然後返回該通道的key  
            sk.attach(new Acceptor(ssc)); // 給定key一個附加的Acceptor對象  
        }  
      
        @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((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();  
        }  
      
    }  

 

    // 接受連線請求線程  
    package server;  
      
    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();  
            }  
        }  
      
    }  

 

    package server;  
      
    import java.io.IOException;  
    import java.nio.channels.SelectionKey;  
    import java.nio.channels.Selector;  
    import java.nio.channels.ServerSocketChannel;  
    import java.util.Iterator;  
    import java.util.Set;  
      
    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;  
        }  
    }  

 

    // Handler線程  
    package server;  
      
    import java.io.IOException;  
    import java.nio.channels.SelectionKey;  
    import java.nio.channels.SocketChannel;  
    import java.util.concurrent.LinkedBlockingQueue;  
    import java.util.concurrent.ThreadPoolExecutor;  
    import java.util.concurrent.TimeUnit;  
      
    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, 10, TimeUnit.SECONDS,  
                new LinkedBlockingQueue<Runnable>()); // 線程池  
      
        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;  
        }  
    }  

 

    package server;  
      
    import java.io.IOException;  
    import java.nio.channels.SelectionKey;  
    import java.nio.channels.SocketChannel;  
    import java.util.concurrent.ThreadPoolExecutor;  
      
    public interface HandlerState {  
      
        public void changeState(TCPHandler h);  
      
        public void handle(TCPHandler h, SelectionKey sk, SocketChannel sc,  
                ThreadPoolExecutor pool) throws IOException ;  
    }  

 

    package server;  
      
    import java.io.IOException;  
    import java.nio.ByteBuffer;  
    import java.nio.channels.SelectionKey;  
    import java.nio.channels.SocketChannel;  
    import java.util.concurrent.ThreadPoolExecutor;  
      
    public class ReadState implements HandlerState{  
      
        private SelectionKey sk;  
          
        public ReadState() {  
        }  
          
        @Override  
        public void changeState(TCPHandler h) {  
            // TODO Auto-generated method stub  
            h.setState(new WorkState());  
        }  
      
        @Override  
        public void handle(TCPHandler h, SelectionKey sk, SocketChannel sc,  
                ThreadPoolExecutor pool) throws IOException { // read()  
            this.sk = sk;  
            // non-blocking下不可用Readers,因為Readers不支援non-blocking  
            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.");  
                h.closeChannel();  
                return;  
            }  
            String str = new String(arr); // 將讀取到的byte內容轉為字符串型態  
            if ((str != null) && !str.equals(" ")) {  
                h.setState(new WorkState()); // 改變狀態(READING->WORKING)  
                pool.execute(new WorkerThread(h, str)); // do process in worker thread  
                System.out.println(sc.socket().getRemoteSocketAddress().toString()  
                        + " > " + str);  
            }  
              
        }  
          
        /* 
         * 執行邏輯處理之函數 
         */  
        synchronized void process(TCPHandler h, String str) {  
            // do process(decode, logically process, encode)..  
            // ..  
            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);  
            }  
      
        }  
    }  

 

    package server;  
      
    import java.io.IOException;  
    import java.nio.channels.SelectionKey;  
    import java.nio.channels.SocketChannel;  
    import java.util.concurrent.ThreadPoolExecutor;  
      
    public class WorkState implements HandlerState {  
      
        public WorkState() {  
        }  
          
        @Override  
        public void changeState(TCPHandler h) {  
            // TODO Auto-generated method stub  
            h.setState(new WriteState());  
        }  
      
        @Override  
        public void handle(TCPHandler h, SelectionKey sk, SocketChannel sc,  
                ThreadPoolExecutor pool) throws IOException {  
            // TODO Auto-generated method stub  
              
        }  
      
    }  

 

    package server;  
      
    import java.io.IOException;  
    import java.nio.ByteBuffer;  
    import java.nio.channels.SelectionKey;  
    import java.nio.channels.SocketChannel;  
    import java.util.concurrent.ThreadPoolExecutor;  
      
    public class WriteState implements HandlerState{  
      
        public WriteState() {  
        }  
          
        @Override  
        public void changeState(TCPHandler h) {  
            // TODO Auto-generated method stub  
            h.setState(new ReadState());  
        }  
      
        @Override  
        public void handle(TCPHandler h, SelectionKey sk, SocketChannel sc,  
                ThreadPoolExecutor pool) throws IOException { // send()  
            // 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操作立即返回  
        }  
    }  

 

    package server;  
      
    import java.io.IOException;  
      
    public class Main {  
      
          
        public static void main(String[] args) {  
            // TODO Auto-generated method stub  
            try {  
                TCPReactor reactor = new TCPReactor(1333);  
                new Thread(reactor).start();  
            } catch (IOException e) {  
                // TODO Auto-generated catch block  
                e.printStackTrace();  
            }  
        }  
      
    }  

 

 

 

 

 

 

 

总的来说,主从式reactor比多线程的reactor先进的地方在于:

1.主reactor是一个线程,负责监听外部的连线请求,并派发给Acceptor处理。故Main Reactor中的selector只有注册OP_ACCEPT事件,也只能监听OP_ACCEPT事件。

   而处理请求是其他N个不同的线程,即从reactor

2.可以根据请求的密集度来调控从reactor的个数

 

参考文章:

https://blog.csdn.net/yehjordan/article/details/51026045

版权声明:本文为mweibiao原创文章,遵循 CC 4.0 BY-SA 版权协议,转载请附上原文出处链接和本声明。
本文链接:https://blog.csdn.net/mweibiao/article/details/79713698