日本综合一区二区|亚洲中文天堂综合|日韩欧美自拍一区|男女精品天堂一区|欧美自拍第6页亚洲成人精品一区|亚洲黄色天堂一区二区成人|超碰91偷拍第一页|日韩av夜夜嗨中文字幕|久久蜜综合视频官网|精美人妻一区二区三区

RELATEED CONSULTING
相關(guān)咨詢
選擇下列產(chǎn)品馬上在線溝通
服務(wù)時間：8:30-17:00
你可能遇到了下面的問題
關(guān)閉右側(cè)工具欄

新聞中心

這里有您想知道的互聯(lián)網(wǎng)營銷解決方案
格物致知-記一次Nodejs源碼分析的經(jīng)歷

本文轉(zhuǎn)載自微信公眾號「編程雜技」，作者theanarkh。轉(zhuǎn)載本文請聯(lián)系編程雜技公眾號。 

蒲江縣ssl適用于網(wǎng)站、小程序/APP、API接口等需要進行數(shù)據(jù)傳輸應(yīng)用場景，ssl證書未來市場廣闊！成為創(chuàng)新互聯(lián)建站的ssl證書銷售渠道，可以享受市場價格4-6折優(yōu)惠！如果有意向歡迎電話聯(lián)系或者加微信：028-86922220（備注：SSL證書合作）期待與您的合作！

昨天分析http模塊相關(guān)的代碼時，遇到了一個晦澀的邏輯，看了想，想了看還是沒看懂。百度、谷歌了很多帖子也沒看到合適的答案。突然看到一個題目有點相識的搜索結(jié)果，點進去是Stack Overflow上的帖子，但是已經(jīng)404，最后還是通過快照功能成功看到內(nèi)容。這個帖子[1]和我的疑惑不相關(guān)，但是突然給了我一些靈感。沿著這個靈感去看了代碼，最后下載nodejs源碼，加了一些log，編譯了一夜(太久了，等不及編譯完成，得睡覺了)。上午起來驗證，終于揭開了疑惑。這個問題源于下面這段代碼。

 
 
 
 
    
  
  
  
  
  1. function connectionListenerInternal(server, socket) { 
    2. 
  
  
  
  
  2.   socket.server = server; 
    3. 
  
  
  
  
  3.   // 分配一個http解析器 
    4. 
  
  
  
  
  4.   const parser = parsers.alloc(); 
    5. 
  
  
  
  
  5.   // 解析請求報文 
    6. 
  
  
  
  
  6.   parser.initialize( 
    7. 
  
  
  
  
  7.     HTTPParser.REQUEST, 
    8. 
  
  
  
  
  8.     new HTTPServerAsyncResource('HTTPINCOMINGMESSAGE', socket), 
    9. 
  
  
  
  
  9.     server.maxHeaderSize || 0, 
    10. 
  
  
  
  
  10.     server.insecureHTTPParser === undefined ? 
    11. 
  
  
  
  
  11.       isLenient() : server.insecureHTTPParser, 
    12. 
  
  
  
  
  12.   ); 
    13. 
  
  
  
  
  13.   parser.socket = socket; 
    14. 
  
  
  
  
  14.   // 開始解析頭部的開始時間 
    15. 
  
  
  
  
  15.   parser.parsingHeadersStart = nowDate(); 
    16. 
  
  
  
  
  16.   socket.parser = parser; 
    17. 
  
  
  
  
  17.   const state = { 
    18. 
  
  
  
  
  18.     onData: null, 
    19. 
  
  
  
  
  19.     onEnd: null, 
    20. 
  
  
  
  
  20.     onClose: null, 
    21. 
  
  
  
  
  21.     onDrain: null, 
    22. 
  
  
  
  
  22.     // 同一tcp連接上，請求和響應(yīng)的的隊列 
    23. 
  
  
  
  
  23.     outgoing: [], 
    24. 
  
  
  
  
  24.     incoming: [], 
    25. 
  
  
  
  
  25.     outgoingData: 0, 
    26. 
  
  
  
  
  26.     keepAliveTimeoutSet: false 
    27. 
  
  
  
  
  27.   }; 
    28. 
  
  
  
  
  28.   state.onData = socketOnData.bind(undefined, server, socket, parser, state); 
    29. 
  
  
  
  
  29.   socket.on('data', state.onData); 
    30. 
  
  
  
  
  30.  
    31. 
  
  
  
  
  31.   if (socket._handle && socket._handle.isStreamBase && 
    32. 
  
  
  
  
  32.       !socket._handle._consumed) { 
    33. 
  
  
  
  
  33.     parser._consumed = true; 
    34. 
  
  
  
  
  34.     socket._handle._consumed = true; 
    35. 
  
  
  
  
  35.     parser.consume(socket._handle); 
    36. 
  
  
  
  
  36.   } 
    37. 
  
  
  
  
  37.   parser[kOnExecute] = 
    38. 
  
  
  
  
  38.     onParserExecute.bind(undefined, server, socket, parser, state); 
    39. 
  
  
  
  
  39.  
    40. 
  
  
  
  
  40.   socket._paused = false; 
    41. 
  
  
  
  
  41. } 
    42.

這段代碼看起來很多，這是啟動http服務(wù)器后，有新的tcp連接建立時執(zhí)行的回調(diào)。問題在于tcp上有數(shù)據(jù)到來時，是怎么處理的，上面代碼中nodejs監(jiān)聽了socket的data事件，同時注冊了鉤子kOnExecute。data事件我們都知道是流上有數(shù)據(jù)到來時觸發(fā)的事件。我們看一下socketOnData做了什么事情。

 
 
 
 
    
  
  
  
  
  1. function socketOnData(server, socket, parser, state, d) { 
    2. 
  
  
  
  
  2.   // 交給http解析器處理，返回已經(jīng)解析的字節(jié)數(shù) 
    3. 
  
  
  
  
  3.   const ret = parser.execute(d); 
    4. 
  
  
  
  
  4.   onParserExecuteCommon(server, socket, parser, state, ret, d); 
    5. 
  
  
  
  
  5. } 
    6.

這看起來沒有問題，socket上有數(shù)據(jù)，然后交給http解析器處理。幾乎所有http模塊源碼解析的文章也是這樣分析的，我第一反應(yīng)也覺得這個沒問題，那kOnExecute是做什么的呢?kOnExecute鉤子函數(shù)的值是onParserExecute，這個看起來也是解析tcp上的數(shù)據(jù)的，看起來和onSocketData是一樣的作用，難道tcp上的數(shù)據(jù)有兩個消費者?我們看一下kOnExecute什么時候被回調(diào)的。

 
 
 
 
    
  
  
  
  
  1. void OnStreamRead(ssize_t nread, const uv_buf_t& buf) override { 
    2. 
  
  
  
  
  2.  
    3. 
  
  
  
  
  3.     Local ret = Execute(buf.base, nread); 
    4. 
  
  
  
  
  4.     Local cb = 
    5. 
  
  
  
  
  5.         object()->Get(env()->context(), kOnExecute).ToLocalChecked(); 
    6. 
  
  
  
  
  6.     MakeCallback(cb.As(), 1, &ret); 
    7. 
  
  
  
  
  7.   } 
    8.

在node_http_parser.cc中的OnStreamRead中被回調(diào)，那么OnStreamRead又是什么時候被回調(diào)的呢?OnStreamRead是nodejs中c++層流操作的通用函數(shù)，當(dāng)流有數(shù)據(jù)的時候就會執(zhí)行該回調(diào)。而且OnStreamRead中也會把數(shù)據(jù)交給http解析器解析。這看起來真的有兩個消費者?這就很奇怪，為什么一份數(shù)據(jù)會交給http解析器處理兩次?這時候我的想法就是這兩個地方肯定是互斥的。但是我一直沒有找到是哪里做了處理。最后在connectionListenerInternal的一段代碼中找到了答案。

 
 
 
 
    
  
  
  
  
  1. if (socket._handle && socket._handle.isStreamBase && !socket._handle._consumed) { 
    2. 
  
  
  
  
  2.     parser._consumed = true; 
    3. 
  
  
  
  
  3.     socket._handle._consumed = true; 
    4. 
  
  
  
  
  4.     parser.consume(socket._handle); 
    5. 
  
  
  
  
  5.   } 
    6.

因為tcp流是繼承StreamBase類的，所以if成立(后面會具體分析)。我們看一下consume的實現(xiàn)。

 
 
 
 
    
  
  
  
  
  1. static void Consume(const FunctionCallbackInfo& args) { 
    2. 
  
  
  
  
  2.     Parser* parser; 
    3. 
  
  
  
  
  3.     ASSIGN_OR_RETURN_UNWRAP(&parser, args.Holder()); 
    4. 
  
  
  
  
  4.     CHECK(args[0]->IsObject()); 
    5. 
  
  
  
  
  5.     StreamBase* stream = StreamBase::FromObjject(args[0].As()); 
  
  
  
  
  6.     CHECK_NOT_NULL(stream); 
    7. 
  
  
  
  
  7.     stream->PushStreamListener(parser); 
    8. 
  
  
  
  
  8.   } 
    9. 
 
 
 
 
    http解析器把自己注冊為tcp stream的一個listener。這里涉及到了c++層對流的設(shè)計。我們從頭開始。看一下PushStreamListener做了什么事情。c++層中，流的操作由類StreamResource進行了封裝。
    

     
 
 
 
      
  
  
  
  
    1. class StreamResource { 
      2. 
  
  
  
  
    2.  public: 
      3. 
  
  
  
  
    3.   virtual ~StreamResource(); 
      4. 
  
  
  
  
    4.   virtual int ReadStart() = 0; 
      5. 
  
  
  
  
    5.   virtual int ReadStop() = 0; 
      6. 
  
  
  
  
    6.   virtual int DoShutdown(ShutdownWrap* req_wrap) = 0; 
      7. 
  
  
  
  
    7.   virtual int DoTryWrite(uv_buf_t** bufs, size_t* count); 
      8. 
  
  
  
  
    8.   virtual int DoWrite(WriteWrap* w, 
      9. 
  
  
  
  
    9.                       uv_buf_t* bufs, 
      10. 
  
  
  
  
    10.                       size_t count, 
      11. 
  
  
  
  
    11.                       uv_stream_t* send_handle) = 0; 
      12. 
  
  
  
  
    12.   void PushStreamListener(StreamListener* listener); 
      13. 
  
  
  
  
    13.   void RemoveStreamListener(StreamListener* listener); 
      14. 
  
  
  
  
    14.  
      15. 
  
  
  
  
    15.  protected: 
      16. 
  
  
  
  
    16.   uv_buf_t EmitAlloc(size_t suggested_size); 
      17. 
  
  
  
  
    17.   void EmitRead(ssize_t nread, const uv_buf_t& buf = uv_buf_init(nullptr, 0)); 
      18. 
  
  
  
  
    18.  
      19. 
  
  
  
  
    19.   StreamListener* listener_ = nullptr; 
      20. 
  
  
  
  
    20.   uint64_t bytes_read_ = 0; 
      21. 
  
  
  
  
    21.   uint64_t bytes_written_ = 0; 
      22. 
  
  
  
  
    22.   friend class StreamListener; 
      23. 
  
  
  
  
    23. }; 
      24. 
 
 
 
 
    我們看到StreamResource是一個基類，定義了操作流的公共方法。其中有一個成員是StreamListener類的實例。我們看看StreamListener的實現(xiàn)。
    

     
 
 
 
      
  
  
  
  
    1. class StreamListener { 
      2. 
  
  
  
  
    2.  public: 
      3. 
  
  
  
  
    3.   virtual ~StreamListener(); 
      4. 
  
  
  
  
    4.   virtual uv_buf_t OnStreamAlloc(size_t suggested_size) = 0; 
      5. 
  
  
  
  
    5.   virtual void OnStreamRead(ssize_t nread, 
      6. 
  
  
  
  
    6.                             const uv_buf_t& buf) = 0; 
      7. 
  
  
  
  
    7.   virtual void OnStreamDestroy() {} 
      8. 
  
  
  
  
    8.   inline StreamResource* stream() { return stream_; } 
      9. 
  
  
  
  
    9.  
      10. 
  
  
  
  
    10.  protected: 
      11. 
  
  
  
  
    11.   void PassReadErrorToPreviousListener(ssize_t nread); 
      12. 
  
  
  
  
    12.  
      13. 
  
  
  
  
    13.   StreamResource* stream_ = nullptr; 
      14. 
  
  
  
  
    14.   StreamListener* previous_listener_ = nullptr; 
      15. 
  
  
  
  
    15.   friend class StreamResource; 
      16. 
  
  
  
  
    16. }; 
      17. 
 
 
 
 
    StreamListener是一個負(fù)責(zé)消費流數(shù)據(jù)的類。StreamListener 和StreamResource類的關(guān)系如下。
    

    null我們看到一個流可以注冊多個listener，多個listener形成一個鏈表。接著我們看一下創(chuàng)建一個c++層的tcp對象是怎樣的。下面是TCPWrap的繼承關(guān)系。
    

     
 
 
 
      
  
  
  
  
    1. class TCPWrap : public ConnectionWrap{} 
      2. 
  
  
  
  
    2. class ConnectionWrap : public LibuvStreamWrap{} 
      3. 
  
  
  
  
    3. class LibuvStreamWrap : public HandleWrap, public StreamBase{} 
      4. 
  
  
  
  
    4. class StreamBase : public StreamResource {} 
      5. 
 
 
 
 
    我們看到tcp流是繼承于StreamResource的。新建一個tcp的c++的對象時(tcp_wrap.cc)，會不斷往上調(diào)用父類的構(gòu)造函數(shù)，其中在StreamBase中有一個關(guān)鍵的操作。
    

     
 
 
 
      
  
  
  
  
    1. inline StreamBase::StreamBase(Environment* env) : env_(env) { 
      2. 
  
  
  
  
    2.   PushStreamListener(&default_listener_); 
      3. 
  
  
  
  
    3. } 
      4. 
  
  
  
  
    4.  
      5. 
  
  
  
  
    5. EmitToJSStreamListener default_listener_; 
      6. 
 
 
 
 
    StreamBase會默認(rèn)給流注冊一個listener。我們看下EmitToJSStreamListener 具體的定義。
    

     
 
 
 
      
  
  
  
  
    1. class ReportWritesToJSStreamListener : public StreamListener { 
      2. 
  
  
  
  
    2.  public: 
      3. 
  
  
  
  
    3.   void OnStreamAfterWrite(WriteWrap* w, int status) override; 
      4. 
  
  
  
  
    4.   void OnStreamAfterShutdown(ShutdownWrap* w, int status) override; 
      5. 
  
  
  
  
    5.  
      6. 
  
  
  
  
    6.  private: 
      7. 
  
  
  
  
    7.   void OnStreamAfterReqFinished(StreamReq* req_wrap, int status); 
      8. 
  
  
  
  
    8. }; 
      9. 
  
  
  
  
    9.  
      10. 
  
  
  
  
    10. class EmitToJSStreamListener : public ReportWritesToJSStreamListener { 
      11. 
  
  
  
  
    11.  public: 
      12. 
  
  
  
  
    12.   uv_buf_t OnStreamAlloc(size_t suggested_size) override; 
      13. 
  
  
  
  
    13.   void OnStreamRead(ssize_t nread, const uv_buf_t& buf) override; 
      14. 
  
  
  
  
    14. }; 
      15. 
 
 
 
 
    EmitToJSStreamListener繼承StreamListener ，定義了分配內(nèi)存和讀取接收數(shù)據(jù)的函數(shù)。接著我們看一下PushStreamListener做了什么事情。
    

     
 
 
 
      
  
  
  
  
    1. inline void StreamResource::PushStreamListener(StreamListener* listener) { 
      2. 
  
  
  
  
    2.   // 頭插法  
      3. 
  
  
  
  
    3.   listener->previous_listener_ = listener_; 
      4. 
  
  
  
  
    4.   listener->stream_ = this; 
      5. 
  
  
  
  
    5.   listener_ = listener; 
      6. 
  
  
  
  
    6. } 
      7. 
 
 
 
 
    PushStreamListener就是構(gòu)造出上圖的結(jié)構(gòu)。對應(yīng)到創(chuàng)建一個c++層的tcp對象中，如下圖。
    

    然后我們看一下對于流來說，讀取數(shù)據(jù)的整個鏈路。首先是js層調(diào)用readStart
    

     
 
 
 
      
  
  
  
  
    1. function tryReadStart(socket) { 
      2. 
  
  
  
  
    2.   socket._handle.reading = true; 
      3. 
  
  
  
  
    3.   const err = socket._handle.readStart(); 
      4. 
  
  
  
  
    4.   if (err) 
      5. 
  
  
  
  
    5.     socket.destroy(errnoException(err, 'read')); 
      6. 
  
  
  
  
    6. } 
      7. 
  
  
  
  
    7.  
      8. 
  
  
  
  
    8. // 注冊等待讀事件 
      9. 
  
  
  
  
    9. Socket.prototype._read = function(n) { 
      10. 
  
  
  
  
    10.   tryReadStart(this); 
      11. 
  
  
  
  
    11. }; 
      12. 
 
 
 
 
    我們看看readStart
    

     
 
 
 
      
  
  
  
  
    1. int LibuvStreamWrap::ReadStart() { 
      2. 
  
  
  
  
    2.   return uv_read_start(stream(), [](uv_handle_t* handle, 
      3. 
  
  
  
  
    3.                                     size_t suggested_size, 
      4. 
  
  
  
  
    4.                                     uv_buf_t* buf) { 
      5. 
  
  
  
  
    5.     static_cast(handle->data)->OnUvAlloc(suggested_size, buf); 
      6. 
  
  
  
  
    6.   }, [](uv_stream_t* stream, ssize_t nread, const uv_buf_t* buf) { 
      7. 
  
  
  
  
    7.     static_cast(stream->data)->OnUvRead(nread, buf); 
      8. 
  
  
  
  
    8.   }); 
      9. 
  
  
  
  
    9. } 
      10. 
 
 
 
 
    ReadStart調(diào)用libuv的uv_read_start注冊等待可讀事件，并且注冊了兩個回調(diào)函數(shù)OnUvAlloc和OnUvRead。
    

     
 
 
 
      
  
  
  
  
    1. void LibuvStreamWrap::OnUvRead(ssize_t nread, const uv_buf_t* buf) { 
      2. 
  
  
  
  
    2.    EmitRead(nread, *buf); 
      3. 
  
  
  
  
    3. } 
      4. 
  
  
  
  
    4.  
      5. 
  
  
  
  
    5. inline void StreamResource::EmitRead(ssize_t nread, const uv_buf_t& buf) { 
      6. 
  
  
  
  
    6.   // bytes_read_表示已讀的字節(jié)數(shù) 
      7. 
  
  
  
  
    7.   if (nread > 0) 
      8. 
  
  
  
  
    8.     bytes_read_ += static_cast(nread); 
      9. 
  
  
  
  
    9.   listener_->OnStreamRead(nread, buf); 
      10. 
  
  
  
  
    10. } 
      11. 
 
 
 
 
    通過層層調(diào)用最后會調(diào)用listener_的OnStreamRead。我們看看tcp的OnStreamRead
    

     
 
 
 
      
  
  
  
  
    1. void EmitToJSStreamListener::OnStreamRead(ssize_t nread, const uv_buf_t& buf_) { 
      2. 
  
  
  
  
    2.   StreamBase* stream = static_cast(stream_); 
      3. 
  
  
  
  
    3.   Environment* env = stream->stream_env(); 
      4. 
  
  
  
  
    4.   HandleScope handle_scope(env->isolate()); 
      5. 
  
  
  
  
    5.   Context::Scope context_scope(env->context()); 
      6. 
  
  
  
  
    6.   AllocatedBuffer buf(env, buf_); 
      7. 
  
  
  
  
    7.   stream->CallJSOnreadMethod(nread, buf.ToArrayBuffer()); 
      8. 
  
  
  
  
    8. } 
      9. 
 
 
 
 
    繼續(xù)回調(diào)CallJSOnreadMethod
    

     
 
 
 
      
  
  
  
  
    1. MaybeLocal StreamBase::CallJSOnreadMethod(ssize_t nread, 
      2. 
  
  
  
  
    2.                                                  Local ab, 
      3. 
  
  
  
  
    3.                                                  size_t offset, 
      4. 
  
  
  
  
    4.                                                  StreamBaseJSChecks checks) { 
      5. 
  
  
  
  
    5.   Environment* env = env_; 
      6. 
  
  
  
  
    6.   // ... 
      7. 
  
  
  
  
    7.   AsyncWrap* wrap = GetAsyncWrap(); 
      8. 
  
  
  
  
    8.   CHECK_NOT_NULL(wrap); 
      9. 
  
  
  
  
    9.   Local onread = wrap->object()->GetInternalField(kOnReadFunctionField); 
      10. 
  
  
  
  
    10.   CHECK(onread->IsFunction()); 
      11. 
  
  
  
  
    11.   return wrap->MakeCallback(onread.As(), arraysize(argv), argv); 
      12. 
  
  
  
  
    12. } 
      13. 
 
 
 
 
    CallJSOnreadMethod會回調(diào)js層的onread回調(diào)函數(shù)。onread會把數(shù)據(jù)push到流中，然后觸發(fā)data事件。這是tcp里默認(rèn)的數(shù)據(jù)讀取過程。而文章開頭講到的parser.consume打破了這個默認(rèn)行為。stream->PushStreamListener(parser);修改了tcp流的listener鏈，http parser把自己作為數(shù)據(jù)的接收者。所以這時候tcp流上的數(shù)據(jù)是直接由node_http_parser.cc的OnStreamRead消費的。而不是觸發(fā)socket的data事件，最后通過在nodejs源碼中加log，重新編譯驗證的確如文中所述。最后提一個這個過程中還有一個關(guān)鍵的地方是調(diào)用consume函數(shù)的前提是socket._handle.isStreamBase為true。isStreamBase是在StreamBase::AddMethods中定義為true的，而tcp對象創(chuàng)建的過程中，調(diào)用了這個方法，所以tcp的isStreamBase是true，才會執(zhí)行consume，才會執(zhí)行kOnExecute回調(diào)。
    References
    [1] 帖子: http://cache.baiducontent.com/c?m=rZy2XovtTdJJuXWLM-s8wgpaz8NFubewtolyiC19iAKFJrbGdx2EFnArzlAIDisNP70zWWsCPv-4jwMHTGNcLaUsMVr-lvLqYmmHD-w_fUYz6a5K6OQRC9kZmLYN5RXsb34OdINb8xHIJsdyClaEWOtCGKMQ2saYK7ed7OG8v0E1pRKR4K46phl0rCBrw6amXE3QpPo62dMhvu_VASYYqq&p=cb77c64ad49111a05bee9e264d5693&newp=882a9646dc9712a05ab7cc374f0ccc231615d70e3ad3d501298ffe0cc4241a1a1a3aecbf2d29170ed6c27f630bae4856ecf630723d0834f1f689df08d2ecce7e7b&s=cfcd208495d565ef&user=baidu&fm=sc&query=onParserExecute&qid=869f73bc002e44f5&p1=11
                                                    
    
                                                新聞標(biāo)題：格物致知-記一次Nodejs源碼分析的經(jīng)歷                                                
    
                                                標(biāo)題鏈接：http://www.dlmjj.cn/article/dpejchc.html