2. 程式人生 > >Reactor模型-單執行緒版

Reactor模型-單執行緒版

阿新 • • 發佈：2018-12-19

Reactor模型是典型的事件驅動模型。在網路程式設計中，所謂的事件當然就是read、write、bind、connect、close等這些動作了。Reactor模型的實現有很多種，下面介紹最基本的三種：

  • 單執行緒版
  • 多執行緒版
  • 主從多執行緒版
Key Word：Java NIO，Reactor模型，Java併發程式設計，Event-Driven

單執行緒版本

結構圖（引用自Doug Lea的Scalable IO in Java）如下：

  Reactor模型圖
上圖中Reactor是一個典型的事件驅動中心，客戶端發起請求並建立連線時，會觸發註冊在多路複用器Selector上的SelectionKey.OP_ACCEPT事件，繫結在該事件上的Acceptor物件的職責就是接受請求，為接下來的讀寫操作做準備。

 

Reactor設計如下：

/**
 * Reactor
 * 
 * @author wqx
 *
 */
public class Reactor implements Runnable { private static final Logger LOG = LoggerFactory.getLogger(Reactor.class); private Selector selector; private ServerSocketChannel ssc; private Handler DEFAULT_HANDLER = new Handler(){ @Override public void processRequest(Processor processor, ByteBuffer msg) { //NOOP } }; private Handler handler = DEFAULT_HANDLER; /** * 啟動階段 * @param port * @throws IOException */ public Reactor(int port, int maxClients, Handler serverHandler) throws IOException{ selector = Selector.open(); ssc = ServerSocketChannel.open(); ssc.configureBlocking(false); ssc.socket().bind(new InetSocketAddress(port)); this.handler = serverHandler; SelectionKey sk = ssc.register(selector, SelectionKey.OP_ACCEPT); sk.attach(new Acceptor()); } /** * 輪詢階段 */ @Override public void run() { while(!ssc.socket().isClosed()){ try { selector.select(1000); Set<SelectionKey> keys; synchronized(this){ keys = selector.selectedKeys(); } Iterator<SelectionKey> it = keys.iterator(); while(it.hasNext()){ SelectionKey key = it.next(); dispatch(key); it.remove(); } } catch (IOException e) { e.printStackTrace(); } } close(); } public void dispatch(SelectionKey key){ Runnable r = (Runnable)key.attachment(); if(r != null) r.run(); } /** * 用於接受TCP連線的Acceptor * */ class Acceptor implements Runnable{ @Override public void run() { SocketChannel sc; try { sc = ssc.accept(); if(sc != null){ new Processor(Reactor.this,selector,sc); } } catch (IOException e) { e.printStackTrace(); } } } public void close(){ try { selector.close(); if(LOG.isDebugEnabled()){ LOG.debug("Close selector"); } } catch (IOException e) { LOG.warn("Ignoring exception during close selector, e=" + e); } } public void processRequest(Processor processor, ByteBuffer msg){ if(handler != DEFAULT_HANDLER){ handler.processRequest(processor, msg); } } }

上面是典型的單執行緒版本的Reactor實現，例項化Reactor物件的過程中，在當前多路複用器Selector上註冊了OP_ACCEPT事件，當OP_ACCEPT事件發生後，Reactor通過dispatch方法執行Acceptor的run方法，Acceptor類的主要功能就是接受請求，建立連線，並將代表連線建立的SocketChannel以引數的形式構造Processor物件。

Processor的任務就是進行I/O操作。

下面是Processor的原始碼：

/**
 * Server Processor
 * 
 * @author wqx
 */
public class Processor implements Runnable { private static final Logger LOG = LoggerFactory.getLogger(Processor.class); Reactor reactor; private SocketChannel sc; private final SelectionKey sk; private final ByteBuffer lenBuffer = ByteBuffer.allocate(4); private ByteBuffer inputBuffer = lenBuffer; private ByteBuffer outputDirectBuffer = ByteBuffer.allocateDirect(1024 * 64); private LinkedBlockingQueue<ByteBuffer> outputQueue = new LinkedBlockingQueue<ByteBuffer>(); public Processor(Reactor reactor, Selector sel,SocketChannel channel) throws IOException{ this.reactor = reactor; sc = channel; sc.configureBlocking(false); sk = sc.register(sel, SelectionKey.OP_READ); sk.attach(this); sel.wakeup(); } @Override public void run() { if(sc.isOpen() && sk.isValid()){ if(sk.isReadable()){ doRead(); }else if(sk.isWritable()){ doSend(); } }else{ LOG.error("try to do read/write operation on null socket"); try { if(sc != null) sc.close(); } catch (IOException e) {} } } private void doRead(){ try { int byteSize = sc.read(inputBuffer); if(byteSize < 0){ LOG.error("Unable to read additional data"); } if(!inputBuffer.hasRemaining()){ if(inputBuffer == lenBuffer){ //read length inputBuffer.flip(); int len = inputBuffer.getInt(); if(len < 0){ throw new IllegalArgumentException("Illegal data length"); } //prepare for receiving data inputBuffer = ByteBuffer.allocate(len); }else{ //read data if(inputBuffer.hasRemaining()){ sc.read(inputBuffer); } if(!inputBuffer.hasRemaining()){ inputBuffer.flip(); processRequest(); //clear lenBuffer and waiting for next reading operation lenBuffer.clear(); inputBuffer = lenBuffer; } } } } catch (IOException e) { LOG.error("Unexcepted Exception during read. e=" + e); try { if(sc != null) sc.close(); } catch (IOException e1) { LOG.warn("Ignoring exception when close socketChannel"); } } } /** * process request and get response * * @param request * @return */ private void processRequest(){ reactor.processRequest(this,inputBuffer); } private void doSend(){ try{ /** * write data to channel： * step 1: write the length of data(occupy 4 byte) * step 2: data content */ if(outputQueue.size() > 0){ ByteBuffer directBuffer = outputDirectBuffer; directBuffer.clear(); for(ByteBuffer buf : outputQueue){ buf.flip(); if(buf.remaining() > directBuffer.remaining()){ //prevent BufferOverflowException buf = (ByteBuffer) buf.slice().limit(directBuffer.remaining()); } //transfers the bytes remaining in buf into directBuffer int p = buf.position(); directBuffer.put(buf); //reset position buf.position(p); if(!directBuffer.hasRemaining()){ break; } } directBuffer.flip(); int sendSize = sc.write(directBuffer); while(!outputQueue.isEmpty()){ ByteBuffer buf = outputQueue.peek(); int left = buf.remaining() - sendSize; if(left > 0){ buf.position(buf.position() + sendSize); break; } sendSize -= buf.remaining(); outputQueue.remove(); } } synchronized(reactor){ if(outputQueue.size() == 0){ //disable write disableWrite(); }else{ //enable write enableWrite(); } } } catch (CancelledKeyException e) { LOG.warn("CancelledKeyException occur e=" + e); } catch (IOException e) { LOG.warn("Exception causing close, due to " + e); } } public void sendBuffer(ByteBuffer bb){ try{ synchronized(this.reactor){ if(LOG.isDebugEnabled()){ LOG.debug("add sendable bytebuffer into outputQueue"); } //wrap ByteBuffer with length header ByteBuffer wrapped = wrap(bb); outputQueue.add(wrapped); enableWrite(); } }catch(Exception e){ LOG.error("Unexcepted Exception: ", e); } } private ByteBuffer wrap(ByteBuffer bb){ bb.flip(); lenBuffer.clear(); int len = bb.remaining(); lenBuffer.putInt(len); ByteBuffer resp = ByteBuffer.allocate(len+4); lenBuffer.flip(); resp.put(lenBuffer); resp.put(bb); return resp; } private void enableWrite(){ int i = sk.interestOps(); if((i & SelectionKey.OP_WRITE) == 0){ sk.interestOps(i | SelectionKey.OP_WRITE); } } private void disableWrite(){ int i = sk.interestOps(); if((i & SelectionKey.OP_WRITE) == 1){ sk.interestOps(i & (~SelectionKey.OP_WRITE)); } } }

其實Processor要做的事情很簡單，就是向selector註冊感興趣的讀寫時間，OP_READ或OP_WRITE，然後等待事件觸發，做相應的操作。

    @Override
    public void run() { if(sc.isOpen() && sk.isValid()){ if(sk.isReadable()){ doRead(); }else if(sk.isWritable()){ doSend(); } }else{ LOG.error("try to do read/write operation on null socket"); try { if(sc != null) sc.close(); } catch (IOException e) {} } }

而doRead()和doSend()方法稍微複雜了一點，這裡其實處理了用TCP協議進行通訊時必須要解決的問題：TCP粘包拆包問題

TCP粘包拆包問題

我們都知道TCP協議是面向位元組流的，而位元組流是連續的，無法有效識別應用層資料的邊界。如下圖：


  粘包拆包示意圖

上圖顯示的應用層有三個資料包，D1，D2，D3.當應用層資料傳到傳輸層後，可能會出現粘包拆包現象。

TCP協議的基本傳輸單位是報文段，而每個報文段最大有效載荷是有限制的,一般乙太網MTU為1500，去除IP頭20B，TCP頭20B，那麼剩下的1460B就是傳輸層最大報文段的有效載荷。如果應用層資料大於該值（如上圖中的資料塊D2），那麼傳輸層就會進行拆分重組。

解決方案

  1. 訊息定長（通訊雙方傳送的訊息固定長度，缺點很明顯：浪費可恥！！！）
  2. 每個訊息之間加分割符（缺點：訊息編解碼耗時，並且如果訊息體中本省就包含分隔字元，需要進行轉義，效率低）
  3. 每個資料包加個Header！！！（header中指定後面資料的長度，這不就是Tcp、Ip協議通用的做法麼。。。哈哈）

採用方案三

示意圖如下：


  資料包結構

header區佔用4B，內容為資料的長度。too simple。。。-_-

理論有了，下面具體分析下Read、Write的實現過程：

doRead
inputBuffer負責接受資料，lenBuffer負責接受資料長度，初始化的時候，將lenBuffer賦給inputBuffer，定義如下：

private final ByteBuffer lenBuffer = ByteBuffer.allocate(4);
private ByteBuffer inputBuffer = lenBuffer;
  1. 如果inputBuffer == lenBuffer,那麼從inputBuffer中讀取出一個整型值len，這個值就是接下來要接受的資料的長度。同時分配一個大小為len的記憶體空間，並複製給inputBuffer，準備接受資料！！！
    private void doRead(){ try { int byteSize = sc.read(inputBuffer); if(byteSize < 0){ LOG.error("Unable to read additional data"); } if(!inputBuffer.hasRemaining()){ if(inputBuffer == lenBuffer){ //read length inputBuffer.flip(); int len = inputBuffer.getInt(); if(len < 0){ throw new IllegalArgumentException("Illegal data length"); } //prepare for receiving data inputBuffer = ByteBuffer.allocate(len); else{...}
  1. 如果inputBuffer ！= lenBuffer，那麼開始接受資料吧！
if(inputBuffer == lenBuffer){
        //。。。
}else{
    //read data
    if(inputBuffer.hasRemaining()){ sc.read(inputBuffer); } if(!inputBuffer.hasRemaining()){ inputBuffer.flip(); processRequest(); //clear lenBuffer and waiting for next reading operation lenBuffer.clear(); inputBuffer = lenBuffer; } }

注意：

  1. 必須保證緩衝區是滿的，即inputBuffer.hasRemaining()=false
  2. processRequest後，將inputBuffer重新賦值為lenBuffer，為下一次讀操作做準備。

doWrite

使用者呼叫sendBuffer方法傳送資料，其實就是將資料加入outputQueue，這個outputQueue就是一個傳送緩衝佇列。

public void sendBuffer(ByteBuffer bb){ try{ synchronized(this.reactor){ if(LOG.isDebugEnabled()){ LOG.debug("add sendable bytebuffer into outputQueue"); } //wrap ByteBuffer with length header ByteBuffer wrapped = wrap(bb); outputQueue.add(wrapped); enableWrite(); } }catch(Exception e){ LOG.error("Unexcepted Exception: ", e); } }

doSend方法就很好理解了，無非就是不斷從outputQueue中取資料，然後寫入channel中即可。過程如下：

將傳送佇列outputQueue中的資料寫入緩衝區outputDirectBuffer：

  1. 清空outputDirectBuffer，為傳送資料做準備
  2. 將outputQueue資料寫入outputDirectBuffer
  3. 呼叫socketChannel.write(outputDirectBuffer);將outputDirectBuffer寫入socket緩衝區

執行步驟2的時候，我們可能會遇到這麼幾種情況：

1.某個資料包大小超過了outputDirectBuffer剩餘空間大小

2.outputDirectBuffer已被填滿，但是outputQueue仍有待發送的資料

執行步驟3的時候，也可能出現下面兩種情況：

1.outputDirectBuffer被全部寫入socket緩衝區

2.outputDirectBuffer只有部分資料或者壓根就沒有資料被寫入socket緩衝區

實現過程可以結合原始碼，這裡重點分析下面幾個點：

為什麼需要重置buf的position

int p = buf.position();
directBuffer.put(buf);
//reset position
buf.position(p);

寫入directBuffer的資料是即將被寫入SocketChannel的資料，問題就在於：當我們呼叫

int sendSize = sc.write(directBuffer);

的時候，directBuffer中的資料都被寫入Channel了嗎？明顯是不確定的（具體可以看java.nio.channels.SocketChannel.write(ByteBuffer src)的doc文件）

上面的問題如何解決

思路很簡單，根據write方法返回值sendSize，遍歷outputQueue中的ByteBuffer，根據buf.remaining()和sendSize的大小，才可以確定buf是否真的被髮送了。如下所示：

while(!outputQueue.isEmpty()){
    ByteBuffer buf = outputQueue.peek();
    int left = buf.remaining() - sendSize;
    if(left > 0){
        buf.position(buf.position() + sendSize);
        break; } sendSize -= buf.remaining(); outputQueue.remove(); }

網路通訊基本解決，上面的處理思路是參照Zookeeper網路模組的實現，有興趣可以看Zookeeper相應原始碼。

測試

Server端：

public class ServerTest {

    private static int PORT = 8888; public static void main(String[] args) throws IOException, InterruptedException { Thread t = new Thread(new Reactor(PORT,1024,new MyHandler())); t.start(); System.out.println("server start"); t.join(); } }

使用者自定義Handler：

public class MyHandler implements Handler { @Override public void processRequest(Processor processor, ByteBuffer msg) { byte[] con = new byte[msg.remaining()]; msg.get(con); String str = new String(con,0,con.length); String resp = ""; switch(str){ case "request1":resp = "response1";break; case "request2":resp = "response2";break; case "request3":resp = "response3";break; default :resp = ""; } ByteBuffer buf = ByteBuffer.allocate(resp.getBytes().length); buf.put(resp.getBytes()); processor.sendBuffer(buf); } }

client端

public class ClientTest {

    private static String HOST = "localhost"; private static int PORT = 8888; public static void main(String[] args) throws IOException { Client client = new Client(); client.socket().setTcpNoDelay(true); client.connect( new InetSocketAddress(HOST,PORT)); ByteBuffer msg; for(int i = 1; i <= 3; i++){ msg = ByteBuffer.wrap(("request" + i).getBytes()); System.out.println("send-" + "request" + i); ByteBuffer resp = client.send(msg); byte[] retVal = new byte[resp.remaining()]; resp.get(retVal); System.out.println("receive-" + new String(retVal,0,retVal.length)); } } }

輸出：

send-request1
receive-response1
send-request2
receive-response2
send-request3
receive-response3

Client是一個客戶端工具類，簡單封裝了傳送ByteBuffer前，新增header的邏輯。詳見原始碼。Client.java

總結

在這種實現方式中，dispatch方法是同步阻塞的！！！所有的IO操作和業務邏輯處理都在NIO執行緒（即Reactor執行緒）中完成。如果業務處理很快，那麼這種實現方式沒什麼問題，不用切換到使用者執行緒。但是，想象一下如果業務處理很耗時（涉及很多資料庫操作、磁碟操作等），那麼這種情況下Reactor將被阻塞，這肯定是我們不希望看到的。解決方法很簡單，業務邏輯進行非同步處理,即交給使用者執行緒處理。

下面分析下單執行緒版的Reactor模型的缺點：

  • 自始自終都只有一個Reactor執行緒，缺點很明顯：Reactor意外掛了，整個系統也就無法正常工作，可靠性太差。
  • 單執行緒的另外一個問題是在大負載的情況下，Reactor的處理速度必然會成為系統性能的瓶頸。


作者：TopGun_Viper
連結：https://www.jianshu.com/p/eb28811421e3
來源：簡書
簡書著作權歸作者所有，任何形式的轉載都請聯絡作者獲得授權並註明出處。

相關推薦

Reactor模型-單執行緒版

Reactor模型是典型的事件驅動模型。在網路程式設計中，所謂的事件當然就是read、write、bind、connect、close等這些動作了。Reactor模型的實現有很多種，下面介紹最基本的三種： 單執行緒版 多執行緒版 主從多執行緒版 Ke

Reactor模型-多執行緒程版

1.概述 在Reactor單執行緒版本的設計中，I/O任務乃至業務邏輯都由Reactor執行緒來完成，這無疑增加了Reactor執行緒的負擔，高負載情況下必然會出現效能瓶頸。此外，對於多處理器的伺服器來說，單個Reactor執行緒也發揮不了多CPU的最大功效。下面我們對之前單執行緒版的Re

Redis之單執行緒模型

Redis客戶端對服務端的每次呼叫都經歷了傳送命令，執行命令，返回結果三個過程。其中執行命令階段，由於Redis是單執行緒來處理命令的，所有每一條到達服務端的命令不會立刻執行，所有的命令都會進入一個佇列中，然後逐個被執行。並且多個客戶端傳送的命令的執行順序是不確定的。但是可以確定的是不會有兩條命

ES6非同步操作概述- 單執行緒模型

單執行緒模型 單執行緒模型指的是，JavaScript 只在一個執行緒上執行。也就是說，JavaScript 同時只能執行一個任務，其他任務都必須在後面排隊等待。 注意，JavaScript 只在一個執行緒上執行，不代表 JavaScript 引擎只有一個執行緒。事實上，

node單執行緒非同步模型

提到nodejs都知道單執行緒非同步I/O，但是能說清楚為什麼單執行緒非同步I/O，為什麼能增加網路吞吐量，怎麼充分利用cpu資源嗯，這個知道的就不多了。 首先要說的是I/O，I/O是計算機的抽象概念，指的是鍵盤，滑鼠，印表機，套接字等和記憶體之間的資料交換，I/O的速度是很慢的，知道計算機存貯模型的都知道

單執行緒模型

單執行緒模型指的是，JavaScript只在一個執行緒上執行。也就是說 JavaScript同時只能執行一個任務，其他任務都必須在後面排隊等待。 注意，JavaScript只在一個執行緒上執行，不代表JavaScript引擎只有一個執行緒。時實上，  JavaScript引擎有

單執行緒、多執行緒以及事件驅動程式設計模型對比

中文版全文：http://blog.csdn.net/hanhuili/article/details/9389433#comments 事件驅動程式設計是一種程式設計正規化，這裡程式的執行流由外部事件來決定。它的特點是包含一個事件迴圈，當外部事件發生時使用回撥機制來

redis單執行緒模型分析

redis原理 redis採用自己實現的事件分離器，效率比較高，內部採用非阻塞的執行方式，吞吐能力比較大。 不過，因為一般的記憶體操作都是簡單存取操作，執行緒佔用時間相對較短，主要問題在io上，因此，redis這種模型是合適的，但是如果某一個執行緒出現問題導致執行緒佔用很長時間，那麼reids的單

關於單執行緒epoll模型伺服器的一點說明

現在的不少高效能伺服器都是用epoll模型單執行緒的模式 它的優點在於: 1. 單執行緒避免了多執行緒切換帶來的上下文切換開銷 2. epoll 模型優於select是因為無需阻塞在等待io等待上，而是去處理前一個已經就緒的事件(前一個請求的資料已經到達了或者說是就緒的)

多程序單執行緒模型與單程序多執行緒模型之爭

伺服器，事件 多程序單執行緒模型典型代表：nginx 單程序多執行緒模型典型代表：memcached 另外redis, mongodb也可以說是走的“多程序單執行緒模”模型（叢集），只不過作為資料庫伺服器，需要進行防寫，只提供了讀同步。 原因很簡單，因為伺服器的發展大部分都

單程序單執行緒，完成併發伺服器（select版）

# 單程序單執行緒，利用select函式完成併發伺服器 from select import * from socket import * def main(): serverSocket

Android單執行緒模型中Message、Handler、Message Queue、Looper之間的關係

四者之間的關係： Handler獲取當前執行緒中的looper物件，looper用來從存放Message的MessageQueue中取出Message，再有Handler進行Message的分發和處理. 簡單定義： 1、Message Queue(訊息佇列)： 用來存放通過Handler釋出的訊息，通常附屬

淺析 Node.js 單執行緒模型

總結筆記：對於每個使用者請求，由主執行緒接收並存放於一個事件佇列中（不做任何處理），當無請求發生時，即主執行緒空閒，主執行緒開始迴圈處理事件佇列中的任務： 對於非阻塞JS程式： 1、若某事件需要I/O操作，則主執行緒發出I/O請求，然後繼續執行，由底層

Redis 網路架構及單執行緒模型

最近略有閒暇時間，於是對Redis進行了一些學習，學習途徑除了官方文件還有Redis原始碼，我看的版本是2.8.13，Redis原始碼總行數不到5W行，不同元件拆分非常細緻，閱讀起來也很清晰。這篇部落格主要介紹我對Redis網路層架構以及執行緒模型的一些瞭解，希望能對大家

reactor模式：單執行緒的reactor模式

reactor模式稱之為響應器模式，常用於nio的網路通訊框架，其服務架構圖如下 不同於傳統IO的序列排程方式，NIO把整個服務請求分為五個階段 read：接收到請求，讀取資料 decode：解碼資料 compute：業務邏輯處理 encode：返回資料編碼

《【面試突擊】— Redis篇》-- Redis的執行緒模型瞭解嗎？為啥單執行緒效率還這麼高？

能堅持別人不能堅持的，才能擁有別人未曾擁有的。關注程式設計大道公眾號，讓我們一同堅持心中所想，一起成長！！ 《【面試突擊】— Redis篇》-- Redis的執行緒模型瞭解嗎？為啥單執行緒效率還這麼高？ 在這個系列裡，我會整理一些面試題與大家分享，幫助年後和我一樣想要在金三銀四準備

為什麼reids是單執行緒還那麼快？

背景介紹： redis一個非常大的特點就是快，官方給出說明： 1.單執行緒，減少執行緒切換時間。 2.純記憶體操作 3.I/O多路複用機制 理解起來單執行緒操作減少了執行緒切換的時間，以及減少了多執行緒的複雜度這個很好理解；純記憶體操作，減少了訪問硬碟的操作，但是I/O多路複

十一、JVM(HotSpot)Java記憶體模型與執行緒

注：本博文主要是基於JDK1.7會適當加入1.8內容。 1、Java記憶體模型 記憶體模型：在特定的操作協議下，對特定的記憶體或快取記憶體進行讀寫訪問的抽象過程。不同的物理機擁有不一樣的記憶體模型，而Java虛擬機器也擁有自己的記憶體模型。 主要目標：定義程式中各個變數的訪問規則，

JavaScript到底是不是單執行緒

JavaScript引擎 在瞭解計時器內部運作前，我們必須清楚一點，觸發和執行並不是同一概念，計時器的回撥函式一定會在指定delay的時間後被觸發，但並不一定立即執行，可能需要等待。所有JavaScript程式碼是在一個執行緒裡執行的，像滑鼠點選和計時器之類的事件只有在JS單執行緒空閒時才執行。

Kafka的單執行緒生產消費測試

程式碼: package com.weichai.kafka; import java.util.Properties; import kafka.javaapi.producer.Producer; import kafka.producer.KeyedMessage; import