成人免费xxxxx在线视频软件_久久精品久久久_亚洲国产精品久久久_天天色天天色_亚洲人成一区_欧美一级欧美三级在线观看

Netty源碼之Reactor模式

作者：架構修煉者
開發 前端
每個新連接創建一個線程來處理。對于長連接服務，如果一個client和server保持一個連接的話，有多少個client接入，server就需要創建同等的線程來處理。線程上下文切換，數據同步和內存消耗，對server來說，將是非常大的開銷。

 [[410505]]

學習目標

  • 什么是Reactor模式?
  • Reactor模式由什么組成的?
  • Reactor模式解決什么問題?
  • Reactor模式線程模型有哪些?演進過程?

web處理請求架構

大多數web請求處理流程可以抽象成這幾個步驟：讀取(read)，解碼(decode)，處理(process)，編碼(encode)，發送(send)，如下圖所示：

 

同時，處理web請求通常有兩種架構：傳統基于線程架構和事件驅動架構。

傳統基于線程架構

概念

每個新連接創建一個線程來處理。對于長連接服務，如果一個client和server保持一個連接的話，有多少個client接入，server就需要創建同等的線程來處理。線程上下文切換，數據同步和內存消耗，對server來說，將是非常大的開銷。

代碼實現

傳統基于線程架構通常采用BIO的方式來實現，代碼如下：

  1. public class Server implements Runnable { 
  2.  
  3.     int port; 
  4.  
  5.     public Server(int port) { 
  6.         this.port = port; 
  7.     } 
  8.  
  9.     @Override 
  10.     public void run() { 
  11.         try { 
  12.             ServerSocket serverSocket = new ServerSocket(port); 
  13.             while (true){ 
  14.                 System.out.println("等待新連接..."); 
  15.                 new Thread(new Handler(serverSocket.accept())).start(); 
  16.             } 
  17.         } catch (IOException e) { 
  18.             e.printStackTrace(); 
  19.         } 
  20.     } 
  21.  
  22.     static class Handler implements Runnable{ 
  23.  
  24.         private Socket socket; 
  25.  
  26.         public Handler(Socket socket){ 
  27.             this.socket = socket; 
  28.         } 
  29.  
  30.         @Override 
  31.         public void run() { 
  32.             try { 
  33.                 byte[] input = new byte[1024]; 
  34.  
  35.                 this.socket.getInputStream().read(input); 
  36.                 byte[] output = process(input); 
  37.                 this.socket.getOutputStream().write(output); 
  38.                 this.socket.getOutputStream().flush(); 
  39.                 this.socket.close(); 
  40.                 System.out.println("響應完成！"); 
  41.             } catch (IOException e) { 
  42.                 e.printStackTrace(); 
  43.             } 
  44.         } 
  45.  
  46.         private byte[] process(byte[] input) { 
  47.             System.out.println("讀取內容：" + new String(input)); 
  48.             return input; 
  49.         } 
  50.     } 
  51.  
  52.     public static void main(String[] args) throws InterruptedException { 
  53.         Thread thread = new Thread(new Server(2021)); 
  54.         thread.setDaemon(true); 
  55.         thread.start(); 
  56.  
  57.         synchronized (Server.class) { 
  58.             Server.class.wait(); 
  59.         } 
  60.     } 
  61. } 

為了避免線程創建銷毀的開銷，我們通常會采用線程池，但是同樣也有很大的弊端：

  • 同步阻塞IO，讀寫阻塞，線程等待時間過長
  • 在制定線程策略的時候，只能根據CPU的數目來限定可用線程資源，不能根據連接并發數目來制定，也就是連接有限制。否則很難保證對客戶端請求的高效和公平。
  • 多線程之間的上下文切換，造成線程使用效率并不高，并且不易擴展
  • 狀態數據以及其他需要保持一致的數據，需要采用并發同步控制

應用場景

既然傳統基于線程架構弊端這么多，它存在還有什么價值?它的應用場景是什么?

傳統基于線程架構適用于連接數目比較小且一次傳輸大量數據的場景，比如上傳，下載。

事件驅動架構

事件驅動架構：可以把線程和連接解耦，線程只用于執行事件注冊的回調函數。事件驅動架構由事件生產者和事件消費者組成，前者是事件的來源，它只負責監聽哪些事件發生;后者是直接處理事件或者事件發生時，響應事件的實體。

Reactor模式

什么是Reactor模式?

Reactor模式是事件驅動架構的一種具體實現方法，簡而言之，就是一個單線程進行循環監聽就緒IO事件，并將就緒IO事件分發給對應的回調函數。

Reactor模式由什么組成的?

Reactor模式分為兩個重要組成部分，Reactor和Handler。 Reactor(反應器)：循環監聽就緒IO事件，并分發給回調函數。 Handler(回調函數)：執行對應IO事件的實際業務邏輯。

Reactor模式解決什么問題?

反應器模式可以實現同步的多路復用，同步是指按照事件到達的順序分發處理。反應器 接收來自不同的客戶端的消息、請求和連接，盡管客戶端是并發的，但是反應器可以按照事件到達的順序觸發回調函數。因此，Reactor模式將連接和線程解耦，不需要為每個連接創建單獨線程。這個問題和C10K問題相同，提供了一個解決思路。

Reactor模式下的三種模型

單線程模型：IO事件輪詢，分發(accept)和IO事件執行(read,decode,compute,encode,send)都在一個線程中完成，如下圖所示：

在單線程模型下，不僅IO操作在Reactor線程上，而非IO操作(handlder中process()方法)也在Reactor線程上執行了，當非IO操作執行慢的話，這會大大延遲IO請求響應，所以應該把非IO操作拆出來，來加速Reactor線程對IO請求響應，就出現多線程模型。

單線程模型實現：

  1. // reactor 
  2. public class Reactor implements Runnable { 
  3.  
  4.     int port; 
  5.     Selector selector; 
  6.     ServerSocketChannel serverSocket; 
  7.  
  8.     public Reactor(int port) throws IOException { 
  9.         this.port = port; 
  10.         // 創建serverSocket對象 
  11.         serverSocket = ServerSocketChannel.open(); 
  12.         // 綁定端口 
  13.         serverSocket.socket().bind(new InetSocketAddress(port)); 
  14.         // 配置非阻塞 
  15.         serverSocket.configureBlocking(false); 
  16.         // 創建selector對象 
  17.         selector = Selector.open(); 
  18.         // serversocket注冊到selector上，幫忙監聽accpet事件 
  19.         serverSocket.register(selector, SelectionKey.OP_ACCEPT, new Acceptor(serverSocket,selector)); 
  20.         /** 還可以使用 SPI provider，來創建selector和serversocket對象 
  21.         SelectorProvider p = SelectorProvider.provider(); 
  22.         selector = p.openSelector(); 
  23.         serverSocket = p.openServerSocketChannel(); 
  24.         */ 
  25.     } 
  26.  
  27.     @Override 
  28.     public void run() { 
  29.         try { 
  30.             while (!Thread.interrupted()) { 
  31.                 System.out.println("start select event..."); 
  32.                 selector.select(); 
  33.                 Set selectedKeys = selector.selectedKeys(); 
  34.                 Iterator it = selectedKeys.iterator(); 
  35.                 while (it.hasNext()) { 
  36.                     dispatch((SelectionKey)it.next()); 
  37.                 } 
  38.                 selectedKeys.clear(); 
  39.             } 
  40.         } catch (IOException e) { 
  41.             e.printStackTrace(); 
  42.         } 
  43.     } 
  44.  
  45.     private void dispatch(SelectionKey key) { 
  46.         Runnable r = (Runnable) key.attachment(); 
  47.         if (r != null) { 
  48.             r.run(); 
  49.         } 
  50.     } 
  51.  
  52.  
  53.     public static void main(String[] args) throws IOException, InterruptedException { 
  54.         Thread thread = new Thread(new Reactor(2021)); 
  55.         thread.start(); 
  56.         synchronized (Reactor.class) { 
  57.             Reactor.class.wait(); 
  58.         } 
  59.     } 
  60. } 
  61. // acceptor調度器 
  62. public class Acceptor implements Runnable { 
  63.  
  64.     ServerSocketChannel serverSocket; 
  65.     Selector selector; 
  66.  
  67.     public Acceptor(ServerSocketChannel serverSocket,Selector selector) { 
  68.         this.serverSocket = serverSocket; 
  69.         this.selector = selector; 
  70.     } 
  71.  
  72.     @Override 
  73.     public void run() { 
  74.         try { 
  75.             SocketChannel socket = this.serverSocket.accept(); 
  76.             if (socket != null) { 
  77.                 new Handler(selector,socket); 
  78.             } 
  79.  
  80.         } catch (IOException e) { 
  81.             e.printStackTrace(); 
  82.         } 
  83.     } 
  84. } 
  85. // 回調函數handler 
  86. public class Handler implements Runnable { 
  87.  
  88.     Selector selector; 
  89.     SocketChannel socket; 
  90.     SelectionKey sk; 
  91.  
  92.     ByteBuffer input = ByteBuffer.allocate(1024); 
  93.     ByteBuffer output = ByteBuffer.allocate(1024); 
  94.     static final int READING = 0, SENDING = 1; 
  95.     int state = READING; 
  96.  
  97.  
  98.     public Handler(Selector selector, SocketChannel socket) throws IOException { 
  99.         this.selector = selector; 
  100.         this.socket = socket; 
  101.  
  102.         this.socket.configureBlocking(false); 
  103.         sk = this.socket.register(selector,0); 
  104.         sk.attach(this); 
  105.         sk.interestOps(SelectionKey.OP_READ); 
  106.         selector.wakeup(); 
  107.     } 
  108.  
  109.     @Override 
  110.     public void run() { 
  111.         try{ 
  112.             if (state == READING) { 
  113.                 read(); 
  114.             } else if (state == SENDING) { 
  115.                 send(); 
  116.             } 
  117.         } catch (IOException ex) { 
  118.             ex.printStackTrace(); 
  119.         } 
  120.     } 
  121.  
  122.     private void read() throws IOException { 
  123.         socket.read(input); 
  124.         if (inputIsComplete()) { 
  125.             // 執行業務邏輯代碼 
  126.             process(); 
  127.             state = SENDING; 
  128.             // Normally also do first write now 
  129.             sk.interestOps(SelectionKey.OP_WRITE); 
  130.         } 
  131.     } 
  132.  
  133.     private void send() throws IOException { 
  134.         socket.write(output); 
  135.         socket.close(); 
  136.         if (outputIsComplete()) sk.cancel(); 
  137.     } 
  138.  
  139.     boolean inputIsComplete() { return true;} 
  140.  
  141.     boolean outputIsComplete() {return true;} 
  142.     // 處理非IO操作(業務邏輯代碼) 
  143.     void process(){ 
  144.         String msg = new String(input.array()); 
  145.         System.out.println("讀取內容：" + msg); 
  146.         output.put(msg.getBytes()); 
  147.         output.flip(); 
  148.     } 
  149. } 
  • 多線程模型：與單線程模型不同的是添加一個業務線程池，將非IO操作(業務邏輯處理)交給業務線程池來處理，提高Reactor線程的IO響應，如圖所示：

 

在多線程模型下，雖然將非IO操作拆出去了，但是所有IO操作都在Reactor單線程中完成的。在高負載、高并發場景下，也會成為瓶頸，于是對Reactor單線程進行了優化，出現了主從線程模型。

多線程模型實現：

  1. public class Reactor implements Runnable { 
  2.  
  3.     int port; 
  4.     Selector selector; 
  5.     ServerSocketChannel serverSocket; 
  6.  
  7.  
  8.     public Reactor(int port) throws IOException { 
  9.         this.port = port; 
  10.  
  11.         // 創建serverSocket對象 
  12.         serverSocket = ServerSocketChannel.open(); 
  13.         // 綁定端口 
  14.         serverSocket.socket().bind(new InetSocketAddress(port)); 
  15.         // 配置非阻塞 
  16.         serverSocket.configureBlocking(false); 
  17.  
  18.         // 創建selector對象 
  19.         selector = Selector.open(); 
  20.         // serversocket注冊到selector上，幫忙監聽accpet事件 
  21.         serverSocket.register(selector, SelectionKey.OP_ACCEPT, new Acceptor("Acceptor",serverSocket,selector)); 
  22.  
  23.         /** 還可以使用 SPI provider，來創建selector和serversocket對象 
  24.         SelectorProvider p = SelectorProvider.provider(); 
  25.         selector = p.openSelector(); 
  26.         serverSocket = p.openServerSocketChannel(); 
  27.         */ 
  28.     } 
  29.  
  30.     @Override 
  31.     public void run() { 
  32.         try { 
  33.             while (!Thread.interrupted()) { 
  34.                 System.out.println("start select event..."); 
  35.                 selector.select(); 
  36.                 Set selectedKeys = selector.selectedKeys(); 
  37.                 Iterator it = selectedKeys.iterator(); 
  38.                 while (it.hasNext()) { 
  39.                     dispatch((SelectionKey)it.next()); 
  40.                 } 
  41.                 selectedKeys.clear(); 
  42.             } 
  43.         } catch (IOException e) { 
  44.             e.printStackTrace(); 
  45.         } 
  46.     } 
  47.  
  48.     private void dispatch(SelectionKey key) { 
  49.         SelfRunable r = (SelfRunable) key.attachment(); 
  50.         if (r != null) { 
  51.             System.out.println("dispatch to " + r.getName() + "===="); 
  52.             r.run(); 
  53.         } 
  54.     } 
  55.  
  56.  
  57.     public static void main(String[] args) throws IOException, InterruptedException { 
  58.  
  59.         Thread thread = new Thread(new Reactor(2021)); 
  60.         thread.start(); 
  61.  
  62.         synchronized (Reactor.class) { 
  63.             Reactor.class.wait(); 
  64.         } 
  65.  
  66.  
  67.     } 
  68. } 
  69.  
  70. public class Acceptor implements SelfRunable { 
  71.     ServerSocketChannel serverSocket; 
  72.     Selector selector; 
  73.     String name; 
  74.     public Acceptor(String name, ServerSocketChannel serverSocket,Selector selector) { 
  75.         this.name = name; 
  76.         this.serverSocket = serverSocket; 
  77.         this.selector = selector; 
  78.     } 
  79.  
  80.     @Override 
  81.     public void run() { 
  82.         try { 
  83.             SocketChannel socket = this.serverSocket.accept(); 
  84.             if (socket != null) { 
  85.                 new Handler("handler_" + ((InetSocketAddress)socket.getLocalAddress()).getPort(), selector,socket); 
  86.             } 
  87.  
  88.         } catch (IOException e) { 
  89.             e.printStackTrace(); 
  90.         } 
  91.     } 
  92.  
  93.     @Override 
  94.     public String getName() { 
  95.         return this.name; 
  96.     } 
  97. } 
  98.  
  99. public class Handler implements SelfRunable { 
  100.     String name; 
  101.     Selector selector; 
  102.     SocketChannel socket; 
  103.     SelectionKey sk; 
  104.     ByteBuffer input = ByteBuffer.allocate(1024); 
  105.     ByteBuffer output = ByteBuffer.allocate(1024); 
  106.     static final int READING = 0, SENDING = 1,  PROCESSING = 3; 
  107.     volatile int state = READING; 
  108.     static ExecutorService poolExecutor = Executors.newFixedThreadPool(5); 
  109.  
  110.     public Handler(String name, Selector selector, SocketChannel socket) throws IOException { 
  111.         this.selector = selector; 
  112.         this.socket = socket; 
  113.         this.name = name; 
  114.  
  115.         this.socket.configureBlocking(false); 
  116.         sk = this.socket.register(selector,0); 
  117.         sk.attach(this); 
  118.         sk.interestOps(SelectionKey.OP_READ); 
  119.         selector.wakeup(); 
  120.     } 
  121.  
  122.     @Override 
  123.     public void run() { 
  124.         try{ 
  125.             System.out.println("state:" + state); 
  126.             if (state == READING) { 
  127.                 read(); 
  128.             } else if (state == SENDING) { 
  129.                 send(); 
  130.             } 
  131.         } catch (IOException ex) { 
  132.             ex.printStackTrace(); 
  133.         } 
  134.     } 
  135.  
  136.     synchronized void read() throws IOException { 
  137.         socket.read(input); 
  138.         if (inputIsComplete()) { 
  139.             state = PROCESSING; 
  140.            poolExecutor.execute(new Processer()); 
  141.         } 
  142.     } 
  143.  
  144.     synchronized void processAndHandOff() { 
  145.         System.out.println("processAndHandOff========="); 
  146.         process(); 
  147.         state = SENDING; // or rebind attachment 
  148.         sk.interestOps(SelectionKey.OP_WRITE); 
  149.         selector.wakeup(); 
  150.         System.out.println("processAndHandOff finish ! ========="); 
  151.     } 
  152.  
  153.     private void send() throws IOException { 
  154.         System.out.println("start send ..."); 
  155.         socket.write(output); 
  156.         socket.close(); 
  157.         System.out.println("start send finish!"); 
  158.         if (outputIsComplete()) sk.cancel(); 
  159.     } 
  160.  
  161.     boolean inputIsComplete() { return true;} 
  162.  
  163.     boolean outputIsComplete() {return true;} 
  164.  
  165.     void process(){ 
  166.         String msg = new String(input.array()); 
  167.         System.out.println("讀取內容：" + msg); 
  168.         output.put(msg.getBytes()); 
  169.         output.flip(); 
  170.     } 
  171.  
  172.     @Override 
  173.     public String getName() { 
  174.         return this.name; 
  175.     } 
  176.  
  177.     class Processer implements Runnable { 
  178.         public void run() { processAndHandOff(); } 
  179.     } 
  180. } 
  • 主從線程模型： 相比多線程模型而言，對于多核cpu，為了充分利用資源，將Reactor拆分成了mainReactor 和 subReactor，但是，主從線程模型也有弊端，不適合大量數據傳輸。 mainReactor：負責監聽接收(accpet)新連接，將新連接后續操作交給subReactor來處理，通常由一個線程處理。 subReactor： 負責處理IO的讀寫操作，通常由多個線程處理。 非IO操作依然由業務線程池來處理。

主從線程模型實現：

  1. public class Reactor implements Runnable { 
  2.  
  3.     int port; 
  4.     Selector selector; 
  5.     ServerSocketChannel serverSocket; 
  6.     int SUBREACTOR_SIZE = 1; 
  7.     SubReactor[] subReactorPool = new SubReactor[SUBREACTOR_SIZE]; 
  8.  
  9.  
  10.     public Reactor(int port) throws IOException { 
  11.         this.port = port; 
  12.  
  13.         // 創建serverSocket對象 
  14.         serverSocket = ServerSocketChannel.open(); 
  15.         // 綁定端口 
  16.         serverSocket.socket().bind(new InetSocketAddress(port)); 
  17.         // 配置非阻塞 
  18.         serverSocket.configureBlocking(false); 
  19.  
  20.         // 創建selector對象 
  21.         selector = Selector.open(); 
  22.         // serversocket注冊到selector上，幫忙監聽accpet事件 
  23.         serverSocket.register(selector, SelectionKey.OP_ACCEPT, new Acceptor("Acceptor",serverSocket,subReactorPool)); 
  24.  
  25.         // 初始化subreactor pool 
  26.         initSubReactorPool(); 
  27.  
  28.  
  29.         /** 還可以使用 SPI provider，來創建selector和serversocket對象 
  30.         SelectorProvider p = SelectorProvider.provider(); 
  31.         selector = p.openSelector(); 
  32.         serverSocket = p.openServerSocketChannel(); 
  33.         */ 
  34.     } 
  35.  
  36.     @Override 
  37.     public void run() { 
  38.         try { 
  39.             while (!Thread.interrupted()) { 
  40.                 System.out.println("mainReactor start select event..."); 
  41.                 selector.select(); 
  42.                 Set selectedKeys = selector.selectedKeys(); 
  43.                 Iterator it = selectedKeys.iterator(); 
  44.                 while (it.hasNext()) { 
  45.                     dispatch((SelectionKey)it.next()); 
  46.                 } 
  47.                 selectedKeys.clear(); 
  48.             } 
  49.         } catch (IOException e) { 
  50.             e.printStackTrace(); 
  51.         } 
  52.     } 
  53.  
  54.     void initSubReactorPool() { 
  55.         try { 
  56.             for (int i = 0; i < SUBREACTOR_SIZE; i++) { 
  57.                 subReactorPool[i] = new SubReactor("SubReactor" + i); 
  58.             } 
  59.         } catch (IOException ex) { /* ... */ } 
  60.     } 
  61.  
  62.     private void dispatch(SelectionKey key) { 
  63.         SelfRunable r = (SelfRunable) key.attachment(); 
  64.         if (r != null) { 
  65.             System.out.println("mainReactor dispatch to " + r.getName() + "===="); 
  66.             r.run(); 
  67.         } 
  68.     } 
  69.  
  70.  
  71.     public static void main(String[] args) throws IOException, InterruptedException { 
  72.  
  73.         Thread thread = new Thread(new Reactor(2021)); 
  74.         thread.start(); 
  75.  
  76.         synchronized (Reactor.class) { 
  77.             Reactor.class.wait(); 
  78.         } 
  79.     } 
  80. } 
  81.  
  82. public class SubReactor implements SelfRunable { 
  83.  
  84.     private Selector selector; 
  85.     private String name; 
  86.     private List<SelfRunable> task = new ArrayList<SelfRunable>(); 
  87.  
  88.     public SubReactor(String name) throws IOException { 
  89.         this.name = name; 
  90.         selector = Selector.open(); 
  91.         new Thread(this).start(); 
  92.     } 
  93.  
  94.     @Override 
  95.     public String getName() { 
  96.         return this.name; 
  97.     } 
  98.  
  99.     @Override 
  100.     public void run() { 
  101.         try { 
  102.             while (!Thread.interrupted()) { 
  103.                 System.out.println("subReactor start select event..."); 
  104.                 selector.select(5000); 
  105.                 Set selectedKeys = selector.selectedKeys(); 
  106.                 Iterator it = selectedKeys.iterator(); 
  107.                 while (it.hasNext()) { 
  108.                     dispatch((SelectionKey)it.next()); 
  109.                 } 
  110.                 selectedKeys.clear(); 
  111.  
  112.             } 
  113.         } catch (IOException e) { 
  114.             e.printStackTrace(); 
  115.         } 
  116.     } 
  117.  
  118.     private void dispatch(SelectionKey key) { 
  119.         SelfRunable r = (SelfRunable) key.attachment(); 
  120.         if (r != null) { 
  121.             System.out.println("subReactor dispatch to " + r.getName() + "===="); 
  122.             r.run(); 
  123.         } 
  124.     } 
  125.  
  126.     public Selector getSelector(){ 
  127.         return this.selector; 
  128.     } 
  129.  
  130.     public void submit(SelfRunable runnable) { 
  131.         task.add(runnable); 
  132.     } 
  133.  
  134. } 
  135.  
  136. public class Acceptor implements SelfRunable { 
  137.  
  138.     int next = 0; 
  139.     String name; 
  140.     SubReactor[] subReactorPool; 
  141.     ServerSocketChannel serverSocket; 
  142.  
  143.     public Acceptor(String name, ServerSocketChannel serverSocket,SubReactor[] subReactorPool) { 
  144.         this.name = name; 
  145.         this.serverSocket = serverSocket; 
  146.         this.subReactorPool = subReactorPool; 
  147.     } 
  148.  
  149.     @Override 
  150.     public void run() { 
  151.         try { 
  152.             SocketChannel socket = this.serverSocket.accept(); 
  153.             if (socket != null) { 
  154.                 new Handler("handler", subReactorPool[next].getSelector(),socket); 
  155.             } 
  156.             if (++next == subReactorPool.length) {next=0;} 
  157.  
  158.         } catch (IOException e) { 
  159.             e.printStackTrace(); 
  160.         } 
  161.     } 
  162.  
  163.     @Override 
  164.     public String getName() { 
  165.         return this.name; 
  166.     } 
  167. } 
  168.  
  169. public class Handler implements SelfRunable { 
  170.  
  171.     String name; 
  172.     Selector selector; 
  173.     SocketChannel socket; 
  174.     SelectionKey sk; 
  175.  
  176.     ByteBuffer input = ByteBuffer.allocate(1024); 
  177.     ByteBuffer output = ByteBuffer.allocate(1024); 
  178.     static final int READING = 0, SENDING = 1,  PROCESSING = 3; 
  179.     volatile int state = READING; 
  180.  
  181.     static ExecutorService poolExecutor = Executors.newFixedThreadPool(5); 
  182.  
  183.     public Handler(String name, Selector selector, SocketChannel socket) throws IOException { 
  184.         this.selector = selector; 
  185.         this.socket = socket; 
  186.         this.name = name; 
  187.  
  188.         this.socket.configureBlocking(false); 
  189.         sk = this.socket.register(this.selector,0); 
  190.         sk.attach(this); 
  191.         sk.interestOps(SelectionKey.OP_READ); 
  192.         selector.wakeup(); 
  193.     } 
  194.  
  195.     @Override 
  196.     public void run() { 
  197.         try{ 
  198.             System.out.println("state:" + state); 
  199.             if (state == READING) { 
  200.                 read(); 
  201.             } else if (state == SENDING) { 
  202.                 send(); 
  203.             } 
  204.         } catch (IOException ex) { 
  205.             ex.printStackTrace(); 
  206.         } 
  207.     } 
  208.  
  209.     synchronized void read() throws IOException { 
  210.         socket.read(input); 
  211.         if (inputIsComplete()) { 
  212.             state = PROCESSING; 
  213.            poolExecutor.execute(new Processer()); 
  214.         } 
  215.     } 
  216.  
  217.     synchronized void processAndHandOff() { 
  218.         System.out.println("processAndHandOff========="); 
  219.         process(); 
  220.         state = SENDING; // or rebind attachment 
  221.         sk.interestOps(SelectionKey.OP_WRITE); 
  222.         selector.wakeup(); 
  223.         System.out.println("processAndHandOff finish ! ========="); 
  224.     } 
  225.  
  226.     private void send() throws IOException { 
  227.         System.out.println("start send ..."); 
  228.         socket.write(output); 
  229.         socket.close(); 
  230.         System.out.println("start send finish!"); 
  231.         if (outputIsComplete()) sk.cancel(); 
  232.     } 
  233.  
  234.     boolean inputIsComplete() { return true;} 
  235.  
  236.     boolean outputIsComplete() {return true;} 
  237.  
  238.     void process(){ 
  239.         String msg = new String(input.array()); 
  240.         System.out.println("讀取內容：" + msg); 
  241.         output.put(msg.getBytes()); 
  242.         output.flip(); 
  243.     } 
  244.  
  245.     @Override 
  246.     public String getName() { 
  247.         return this.name; 
  248.     } 
  249.  
  250.     class Processer implements Runnable { 
  251.         public void run() { processAndHandOff(); } 
  252.     } 
  253. } 

Reactor線程模型演進

模型

簡介

弊端

單線程模型

IO/非IO操作都在Reactor單線程中完成

非IO操作執行慢，影響IO操作響應延遲

多線程模型

拆分非IO操作交給業務線程池執行，IO操作由Reator單線程執行

高并發，高負載場景下，Reactor單線程會成為瓶頸

主從線程模型

Reactor單線程拆分為mainReactor和subReactor

不適合大量數據傳輸

Netty線程模型

Reactor主從線程模型-抽象模型

  • 創建ServerSocketChannel過程(創建channel，配置非阻塞)
  • ServerSocketChannel注冊到mainReactor的selector對象上，監聽accept事件
  • mainReactor的selector監聽到新連接SocketChannel，將SocketChannel注冊到subReactor的selector對象上，監聽read/write事件
  • subReactor的selector監聽到read/write事件，移交給業務線程池(對應netty的pipeline)

Netty線程模型

我們再好好看看mainReactor和subReactor，其實這兩個類功能非常相似，所以Netty將mainReactor和subReactor統一成了EventLoop。對于Netty零基礎的，請參考這個Reactor主從線程模型-抽象模型和下面這張圖來理解EventLoop。

 

 

責任編輯：武曉燕 來源： 今日頭條
Reactor模式Netty
相關推薦

2022-03-04 08:10:35

NettyIO模型Reactor

2022-03-06 12:15:38

NettyReactor線程

2025-05-08 10:25:00

Netty網絡編程框架

2021-06-16 14:18:37

NettyReactor線程模型

2022-10-25 08:23:09

Reactor模式I/O

2012-08-24 09:58:09

ReactorDSSC

2022-03-10 07:58:12

ReactorNetty運轉架構

2022-09-29 15:39:10

服務器NettyReactor

2020-12-11 11:04:07

NettyIO

2022-02-09 09:37:54

ReactorNettyI/O

2024-11-22 08:00:00

Netty開發

2019-01-15 10:54:03

高性能ServerReactor

2022-05-24 15:46:51

Wi-FiSTA模式

2024-10-24 20:48:04

Netty線程Java

2020-08-21 07:23:50

工廠模式設計

2021-04-26 17:38:40

ReactorProactor網絡

2021-09-27 08:56:44

NettyChannelHand架構

2015-03-31 18:26:43

陌陌社交

2021-06-09 08:53:34

設計模式策略模式工廠模式

2012-02-29 09:41:14

JavaScript
點贊
收藏

51CTO技術棧公眾號

主站蜘蛛池模板： 美美女高清毛片视频免费观看 | 亚洲国产精品一区二区三区 | 日韩在线视频网址 | 欧美日韩一区在线 | 久视频在线 | 久久成人精品一区二区三区 | 亚洲一级毛片 | 色狠狠桃花综合 | 日韩精品一区二 | 亚洲欧美在线视频 | 91正在播放 | 色久影院 | 免费一区二区 | 久久国产精品-久久精品 | 91麻豆精品国产91久久久更新资源速度超快 | 久久久久久久久一区 | 亚洲网站在线观看 | 成人免费毛片在线观看 | 精品免费国产一区二区三区 | 日韩喷潮| 一区精品在线观看 | 国产九九精品视频 | 欧美性乱| 国产成人在线一区二区 | 欧美片网站免费 | 一级毛片在线播放 | 中文字幕国产日韩 | 欧美日韩国产不卡 | 亚洲免费在线 | 国产美女一区 | 日日爱av | 色悠悠久 | 久久精品国产99国产精品 | www.成人.com | 日韩伦理一区二区 | 免费精品久久久久久中文字幕 | 日韩精品一区二区三区在线播放 | www久久av| 亚洲第一成年免费网站 | 国产免费一区二区三区 | 亚洲精品国产电影 |