Node.js 源码剖析

 主页   资讯   文章   代码   电子书 

HTTP模块实现了HTTP服务器和客户端的功能,是Node.js的核心模块,也是我们使用得最多的模块。本章我们来分析HTTP模块,从中我们可以学习到一个HTTP服务器和客户端是怎么实现的,以及HTTP协议本身的一些原理和优化。

18.1 HTTP解析器

HTTP解析器是HTTP模块的核心,不管是作为服务器处理请求还是客户端处理响应都需要使用HTTP解析器解析HTTP协议。新版Node.js使用了新的HTTP解析器llhttp。根据官方说明llhttp比旧版的http_parser在性能上有了非常大的提高。本节我们分析分析llhttp的基础原理和使用。HTTP解析器是一个非常复杂的状态机,在解析数据的过程中,会不断触发钩子函数。下面是llhttp支持的钩子函数。如果用户定义了对应的钩子,在解析的过程中就会被回调。

    // 开始解析HTTP协议
    int llhttp__on_message_begin(llhttp_t* s, const char* p, const char* endp) {  
      int err;  
      CALLBACK_MAYBE(s, on_message_begin, s);  
      return err;  
    }  

    // 解析出请求url时的回调,最后拿到一个url
    int llhttp__on_url(llhttp_t* s, const char* p, const char* endp) {  
      int err;  
      CALLBACK_MAYBE(s, on_url, s, p, endp - p);  
      return err;  
    }  

    // 解析出HTTP响应状态的回调
    int llhttp__on_status(llhttp_t* s, const char* p, const char* endp) {  
      int err;  
      CALLBACK_MAYBE(s, on_status, s, p, endp - p);  
      return err;  
    }  

    // 解析出头部键时的回调
    int llhttp__on_header_field(llhttp_t* s, const char* p, const char* endp) {  
      int err;  
      CALLBACK_MAYBE(s, on_header_field, s, p, endp - p);  
      return err;  
    }  

    // 解析出头部值时的回调
    int llhttp__on_header_value(llhttp_t* s, const char* p, const char* endp) {  
      int err;  
      CALLBACK_MAYBE(s, on_header_value, s, p, endp - p);  
      return err;  
    }  

    // 解析HTTP头完成时的回调  
    int llhttp__on_headers_complete(llhttp_t* s, const char* p, const char* endp) {  
      int err;  
      CALLBACK_MAYBE(s, on_headers_complete, s);  
      return err;  
    }  

    // 解析完body的回调 
    int llhttp__on_message_complete(llhttp_t* s, const char* p, const char* endp) {  
      int err;  
      CALLBACK_MAYBE(s, on_message_complete, s);  
      return err;  
    }  

    // 解析body时的回调
    int llhttp__on_body(llhttp_t* s, const char* p, const char* endp) {  
      int err;  
      CALLBACK_MAYBE(s, on_body, s, p, endp - p);  
      return err;  
    }  

     // 解析到一个chunk结构头时的回调 
    int llhttp__on_chunk_header(llhttp_t* s, const char* p, const char* endp) {  
      int err;  
      CALLBACK_MAYBE(s, on_chunk_header, s);  
      return err;  
    }  

    // 解析完一个chunk时的回调  
    int llhttp__on_chunk_complete(llhttp_t* s, const char* p, const char* endp) {  
      int err;  
      CALLBACK_MAYBE(s, on_chunk_complete, s);  
      return err;  
    }  

Node.js在node_http_parser.cc中对llhttp进行了封装。该模块导出了一个HTTPParser。

    Local<FunctionTemplate> t=env->NewFunctionTemplate(Parser::New); 
    t->InstanceTemplate()->SetInternalFieldCount(1);  
    t->SetClassName(FIXED_ONE_BYTE_STRING(env->isolate(), 
                      "HTTPParser"));  
    target->Set(env->context(),  
      FIXED_ONE_BYTE_STRING(env->isolate(), "HTTPParser"),   
      t->GetFunction(env->context()).ToLocalChecked()).Check();  

在Node.js中我们通过以下方式使用HTTPParser。

      const parser = new HTTPParser();  

      cleanParser(parser);  
      parser.onIncoming = null;  
      parser[kOnHeaders] = parserOnHeaders;  
      parser[kOnHeadersComplete] = parserOnHeadersComplete;  
      parser[kOnBody] = parserOnBody;  
      parser[kOnMessageComplete] = parserOnMessageComplete; 
      // 初始化HTTP解析器处理的报文类型,这里是响应报文
      parser.initialize(HTTPParser.RESPONSE,
         new HTTPClientAsyncResource('HTTPINCOMINGMESSAGE', req),
         req.maxHeaderSize || 0,
         req.insecureHTTPParser === undefined ?
         isLenient() : req.insecureHTTPParser); 
      // 收到数据后传给解析器处理
      const ret = parser.execute(data);
    }  

我们看一下initialize和execute的代码。Initialize函数用于初始化llhttp。

    static void Initialize(const FunctionCallbackInfo<Value>& args) {
       Environment* env = Environment::GetCurrent(args);  
       bool lenient = args[3]->IsTrue();  

       uint64_t max_http_header_size = 0;  
       // 头部的最大大小  
       if (args.Length() > 2) {  
         max_http_header_size = args[2].As<Number>()->Value();  
       }  
       // 没有设置则取Node.js的默认值  
       if (max_http_header_size == 0) {  
         max_http_header_size=env->options()->max_http_header_size;
       }  
       // 解析的报文类型  
       llhttp_type_t type =  
           static_cast<llhttp_type_t>(args[0].As<Int32>()->Value());

       CHECK(type == HTTP_REQUEST || type == HTTP_RESPONSE);  
       Parser* parser;  
       ASSIGN_OR_RETURN_UNWRAP(&parser, args.Holder());  
       parser->Init(type, max_http_header_size, lenient);  
     }  

Initialize做了一些预处理后调用Init。

    void Init(llhttp_type_t type, uint64_t max_http_header_size, bool lenient) {  
       // 初始化llhttp  
       llhttp_init(&parser_, type, &settings);  
       llhttp_set_lenient(&parser_, lenient);  
       header_nread_ = 0;  
       url_.Reset();  
       status_message_.Reset();  
       num_fields_ = 0;  
       num_values_ = 0;  
      have_flushed_ = false;  
      got_exception_ = false;  
      max_http_header_size_ = max_http_header_size;  
    }  

Init做了一些字段的初始化,最重要的是调用了llhttp_init对llhttp进行了初始化,另外kOn开头的属性是钩子函数,由node_http_parser.cc中的回调,而node_http_parser.cc也会定义钩子函数,由llhttp回调,我们看一下node_http_parser.cc钩子函数的定义和实现。

    const llhttp_settings_t Parser::settings = {  
      Proxy<Call, &Parser::on_message_begin>::Raw,  
      Proxy<DataCall, &Parser::on_url>::Raw,  
      Proxy<DataCall, &Parser::on_status>::Raw,  
      Proxy<DataCall, &Parser::on_header_field>::Raw,  
      Proxy<DataCall, &Parser::on_header_value>::Raw,  
      Proxy<Call, &Parser::on_headers_complete>::Raw,  
      Proxy<DataCall, &Parser::on_body>::Raw,  
      Proxy<Call, &Parser::on_message_complete>::Raw,  
     Proxy<Call, &Parser::on_chunk_header>::Raw,  
     Proxy<Call, &Parser::on_chunk_complete>::Raw,  
    };  

1 开始解析报文的回调

      // 开始解析报文,一个TCP连接可能会有多个报文  
      int on_message_begin() {  
        num_fields_ = num_values_ = 0;  
        url_.Reset();  
        status_message_.Reset();  
        return 0;  
      }  

2 解析url时的回调

    int on_url(const char* at, size_t length) {  
        int rv = TrackHeader(length);  
        if (rv != 0) {  
          return rv;  
        }  

        url_.Update(at, length);  
        return 0;  
      }  

3解析HTTP响应时的回调

    int on_status(const char* at, size_t length) {  
       int rv = TrackHeader(length);  
       if (rv != 0) {  
         return rv;  
       }  

       status_message_.Update(at, length);  
       return 0;  
     }  

4解析到HTTP头的键时回调

    int on_header_field(const char* at, size_t length) {  
        int rv = TrackHeader(length);  
        if (rv != 0) {  
          return rv;  
        }  
        // 相等说明键对值的解析是一一对应的  
        if (num_fields_ == num_values_) {  
          // start of new field name  
          // 键的数加一  
          num_fields_++;  
          // 超过阈值则先回调js消费掉  
          if (num_fields_ == kMaxHeaderFieldsCount) {  
            // ran out of space - flush to javascript land  
            Flush();  
            // 重新开始  
            num_fields_ = 1;  
            num_values_ = 0;  
          }  
          // 初始化  
          fields_[num_fields_ - 1].Reset();  
        }  

        // 保存键  
        fields_[num_fields_ - 1].Update(at, length);  

        return 0;  
    }  

当解析的头部个数达到阈值时,Node.js会先通过Flush函数回调JS层保存当前的一些数据。

    void Flush() {  
        HandleScope scope(env()->isolate());  

        Local<Object> obj = object();  
        // JS层的钩子  
        Local<Value> cb = obj->Get(env()->context(), kOnHeaders).ToLocalChecked();  
        if (!cb->IsFunction())  
          return;  

        Local<Value> argv[2] = {  
          CreateHeaders(),  
          url_.ToString(env())  
        };  

        MaybeLocal<Value> r = MakeCallback(cb.As<Function>(),  
                                           arraysize(argv),  
                                           argv);  
        url_.Reset();  
        have_flushed_ = true;  
      } 

    Local<Array> CreateHeaders() {  
       // HTTP头的个数乘以2,因为一个头由键和值组成  
       Local<Value> headers_v[kMaxHeaderFieldsCount * 2];  
       // 保存键和值到HTTP头  
       for (size_t i = 0; i < num_values_; ++i) {  
         headers_v[i * 2] = fields_[i].ToString(env());  
         headers_v[i * 2 + 1] = values_[i].ToString(env());  
       }  

       return Array::New(env()->isolate(), headers_v, num_values_ * 2);  
     }  

Flush会调用JS层的kOnHeaders钩子函数。

5解析到HTTP头的值时回调

    int on_header_value(const char* at, size_t length) {  
       int rv = TrackHeader(length);  
       if (rv != 0) {  
         return rv;  
       }  
       /* 
         值的个数不等于键的个数说明正解析到键对应的值,即一一对应。 
         否则说明一个键存在多个值,则不更新值的个数,多个值累加到一个slot 
       */  
       if (num_values_ != num_fields_) {  
         // start of new header value  
         num_values_++;  
         values_[num_values_ - 1].Reset();  
       }  

       CHECK_LT(num_values_, arraysize(values_));  
       CHECK_EQ(num_values_, num_fields_);  

       values_[num_values_ - 1].Update(at, length);  

       return 0;  
     }  

6解析完HTTP头后的回调

    int on_headers_complete() {  
        header_nread_ = 0;   
        enum on_headers_complete_arg_index {  
           A_VERSION_MAJOR = 0,  
           A_VERSION_MINOR,  
           A_HEADERS,  
           A_METHOD,  
           A_URL,  
           A_STATUS_CODE,  
          A_STATUS_MESSAGE,  
          A_UPGRADE,  
          A_SHOULD_KEEP_ALIVE,  
          A_MAX  
        };  

        Local<Value> argv[A_MAX];  
        Local<Object> obj = object();  
        Local<Value> cb = obj->Get(env()->context(),  
                                   kOnHeadersComplete).ToLocalChecked();  

        Local<Value> undefined = Undefined(env()->isolate());  
        for (size_t i = 0; i < arraysize(argv); i++)  
          argv[i] = undefined;  
        // 之前flush过,则继续flush到JS层,否则返回全部头给js  
        if (have_flushed_) {  
          // Slow case, flush remaining headers.  
          Flush();  
        } else {  
          // Fast case, pass headers and URL to JS land.  
          argv[A_HEADERS] = CreateHeaders();  
          if (parser_.type == HTTP_REQUEST)  
            argv[A_URL] = url_.ToString(env());  
        }  

        num_fields_ = 0;  
        num_values_ = 0;  

        // METHOD  
        if (parser_.type == HTTP_REQUEST) {  
          argv[A_METHOD] =  
              Uint32::NewFromUnsigned(env()->isolate(), parser_.method);  
        }  

        // STATUS  
        if (parser_.type == HTTP_RESPONSE) {  
          argv[A_STATUS_CODE] =  
              Integer::New(env()->isolate(), parser_.status_code);  
          argv[A_STATUS_MESSAGE] = status_message_.ToString(env());  
        }  

        // VERSION  
        argv[A_VERSION_MAJOR] = Integer::New(env()->isolate(), parser_.http_major);  
        argv[A_VERSION_MINOR] = Integer::New(env()->isolate(), parser_.http_minor);  

        bool should_keep_alive;  
        // 是否定义了keepalive头  
        should_keep_alive = llhttp_should_keep_alive(&parser_);  

        argv[A_SHOULD_KEEP_ALIVE] =  
            Boolean::New(env()->isolate(), should_keep_alive);  
        // 是否是升级协议  
        argv[A_UPGRADE] = Boolean::New(env()->isolate(), parser_.upgrade);  

        MaybeLocal<Value> head_response;  
        {  
          InternalCallbackScope callback_scope(  
              this, InternalCallbackScope::kSkipTaskQueues);  
          head_response = cb.As<Function>()->Call(  
              env()->context(), object(), arraysize(argv), argv);  
        }  

        int64_t val;  

        if (head_response.IsEmpty() || !head_response.ToLocalChecked()  
                                            ->IntegerValue(env()->context())  
                                            .To(&val)) {  
          got_exception_ = true;  
          return -1;  
        }  

        return val;  
      }  

on_headers_complete会执行JS层的kOnHeadersComplete钩子。

7 解析body时的回调

    int on_body(const char* at, size_t length) {  
       EscapableHandleScope scope(env()->isolate());  

       Local<Object> obj = object();  
       Local<Value> cb = obj->Get(env()->context(), kOnBody).ToLocalChecked();  

       // We came from consumed stream  
       if (current_buffer_.IsEmpty()) {  
         // Make sure Buffer will be in parent HandleScope  
         current_buffer_ = scope.Escape(Buffer::Copy(  
             env()->isolate(),  
             current_buffer_data_,  
             current_buffer_len_).ToLocalChecked());  
       }  

       Local<Value> argv[3] = {  
         // 当前解析中的数据  
         current_buffer_,  
         // body开始的位置  
         Integer::NewFromUnsigned(env()->isolate(), at - current_buffer_data_),  
         // body当前长度  
         Integer::NewFromUnsigned(env()->isolate(), length)  
       };  

       MaybeLocal<Value> r = MakeCallback(cb.As<Function>(),  
                                          arraysize(argv),  
                                          argv);   

       return 0;  
     }  

Node.js中并不是每次解析HTTP报文的时候就新建一个HTTP解析器,Node.js使用FreeList数据结构对HTTP解析器实例进行了管理。

    class FreeList {  
      constructor(name, max, ctor) {  
        this.name = name;  
        // 构造函数  
        this.ctor = ctor;  
        // 节点的最大值  
        this.max = max;  
        // 实例列表  
        this.list = [];  
      }  
      // 分配一个实例  
      alloc() {  
        // 有空闲的则直接返回,否则新建一个  
        return this.list.length > 0 ?  
          this.list.pop() :  
          ReflectApply(this.ctor, this, arguments);  
      }  
      // 释放实例  
      free(obj) {  
        // 小于阈值则放到空闲列表,否则释放(调用方负责释放)  
        if (this.list.length < this.max) {  
          this.list.push(obj);  
          return true;  
        }  
        return false;  
      }  
    }  

我们看一下在Node.js中对FreeList的使用。。

    const parsers = new FreeList('parsers', 1000, function parsersCb() {  
      const parser = new HTTPParser();  
      // 初始化字段  
      cleanParser(parser);  
      // 设置钩子  
      parser.onIncoming = null;  
      parser[kOnHeaders] = parserOnHeaders;  
      parser[kOnHeadersComplete] = parserOnHeadersComplete;  
      parser[kOnBody] = parserOnBody;  
      parser[kOnMessageComplete] = parserOnMessageComplete;  

      return parser;  
    });  

HTTP解析器的使用

    var HTTPParser = process.binding('http_parser').HTTPParser;  
    var parser = new HTTPParser(HTTPParser.REQUEST);  

    const kOnHeaders = HTTPParser.kOnHeaders;  
    const kOnHeadersComplete = HTTPParser.kOnHeadersComplete;  
    const kOnBody = HTTPParser.kOnBody;  
    const kOnMessageComplete = HTTPParser.kOnMessageComplete;  
    const kOnExecute = HTTPParser.kOnExecute;  

    parser[kOnHeaders] = function(headers, url) {  
        console.log('kOnHeaders', headers.length, url);  
    }  
    parser[kOnHeadersComplete] = function(versionMajor, versionMinor, headers, method,  
             url, statusCode, statusMessage, upgrade, shouldKeepAlive) {  
        console.log('kOnHeadersComplete', headers);  
    }  

    parser[kOnBody] = function(b, start, len) {  
        console.log('kOnBody', b.slice(start).toString('utf-8'));  
    }  
    parser[kOnMessageComplete] = function() {  
        console.log('kOnMessageComplete');  
    }  
    parser[kOnExecute] = function() {  
        console.log('kOnExecute');  
    }  

    parser.execute(Buffer.from(  
        'GET / HTTP/1.1\r\n' +  
        'Host: http://localhost\r\n\r\n'   
    ));  

以上代码的输出

    kOnHeadersComplete [ 'Host', 'http://localhost' ]  
    kOnMessageComplete  

我们看到只执行了kOnHeadersComplete和 kOnMessageComplete。那其它几个回调什么时候会执行呢?我们接着看。我们把输入改一下。

    parser.execute(Buffer.from(  
        'GET / HTTP/1.1\r\n' +  
        'Host: http://localhost\r\n' +  
        'content-length: 1\r\n\r\n'+  
        '1'  
    ));  

上面代码的输出

    kOnHeadersComplete [ 'Host', 'http://localhost', 'content-length', '1' ]  
    kOnBody 1  
    kOnMessageComplete  

我们看到多了一个回调kOnBody,因为我们加了一个HTTP头content-length指示有body,所以HTTP解析器解析到body的时候就会回调kOnBody。那kOnHeaders什么时候会执行呢?我们继续修改代码。

    parser.execute(Buffer.from(  
        'GET / HTTP/1.1\r\n' +  
        'Host: http://localhost\r\n' +  
        'a: b\r\n'+  
         // 很多'a: b\r\n'+
        'content-length: 1\r\n\r\n'+  
        '1'  
    ));  

以上代码的输出

    kOnHeaders 62 /  
    kOnHeaders 22  
    kOnHeadersComplete undefined  
    kOnBody 1  
    kOnMessageComplete  

我们看到kOnHeaders被执行了,并且执行了两次。因为如果HTTP头的个数达到阈值,在解析HTTP头部的过程中,就先flush到JS层(如果多次达到阈值,则回调多次),并且在解析完所有HTTP头后,会在kOnHeadersComplet回调之前再次回调kOnHeaders(如果还有的话)。最后我们看一下kOnExecute如何触发。

    var HTTPParser = process.binding('http_parser').HTTPParser;  
    var parser = new HTTPParser(HTTPParser.REQUEST);  
    var net = require('net');  

    const kOnHeaders = HTTPParser.kOnHeaders;  
    const kOnHeadersComplete = HTTPParser.kOnHeadersComplete;  
    const kOnBody = HTTPParser.kOnBody;  
    const kOnMessageComplete = HTTPParser.kOnMessageComplete;  
    const kOnExecute = HTTPParser.kOnExecute;  

    parser[kOnHeaders] = function(headers, url) {  
        console.log('kOnHeaders', headers.length, url);  
    }  
    parser[kOnHeadersComplete] = function(versionMajor, versionMinor, headers, method,  
             url, statusCode, statusMessage, upgrade, shouldKeepAlive) {  
        console.log('kOnHeadersComplete', headers);  
    }  

    parser[kOnBody] = function(b, start, len) {  
        console.log('kOnBody', b.slice(start).toString('utf-8'));  
    }  
    parser[kOnMessageComplete] = function() {  
        console.log('kOnMessageComplete');  
    }  
    parser[kOnExecute] = function(a,b) {  
        console.log('kOnExecute,解析的字节数:',a);  
    }  
    // 启动一个服务器  
    net.createServer((socket) => {  
      parser.consume(socket._handle);  
    }).listen(80);  

    // 启动一个客户端  
    setTimeout(() => {  
      var socket = net.connect({port: 80});  
      socket.end('GET / HTTP/1.1\r\n' +  
        'Host: http://localhost\r\n' +  
        'content-length: 1\r\n\r\n'+  
        '1');  
    }, 1000);  

我们需要调用parser.consume方法并且传入一个isStreamBase的流(stream_base.cc定义),才会触发kOnExecute。因为kOnExecute是在StreamBase流可读时触发的。

18.2 HTTP客户端

我们首先看一下使用Node.js作为客户端的例子。

    const data = querystring.stringify({  
      'msg': 'hi'  
    });  

    const options = {  
      hostname: 'your domain',  
      path: '/',  
      method: 'POST',  
      headers: {  
        'Content-Type': 'application/x-www-form-urlencoded',  
        'Content-Length': Buffer.byteLength(data)  
      }  
    };  

    const req = http.request(options, (res) => {  
      res.setEncoding('utf8');  
      res.on('data', (chunk) => {  
        console.log(`${chunk}`);  
      });  
      res.on('end', () => {  
        console.log('end');  
      });  
    });  

    req.on('error', (e) => {  
      console.error(`${e.message}`);  
    });  
    // 发送请求的数据  
    req.write(data);  
    // 设置请求结束  
    req.end();  

我们看一下http.request的实现。

    function request(url, options, cb) {  
      return new ClientRequest(url, options, cb);  
    }  

HTTP客户端通过_http_client.js的ClientRequest实现,ClientRequest的代码非常多,我们只分析核心的流程。我们看初始化一个请求的逻辑。

    function ClientRequest(input, options, cb) {  
      // 继承OutgoingMessage  
      OutgoingMessage.call(this);  
      // 是否使用agent  
      let agent = options.agent;   
      // 忽略agent的处理,具体参考_http_agent.js,主要用于复用TCP连接  
      this.agent = agent;  
      // 建立连接的超时时间  
      if (options.timeout !== undefined)  
        this.timeout = getTimerDuration(options.timeout, 'timeout');  
      // HTTP头个数的阈值  
      const maxHeaderSize = options.maxHeaderSize;  
      this.maxHeaderSize = maxHeaderSize;  
      // 监听响应事件  
      if (cb) {  
        this.once('response', cb);  
      }  
      // 忽略设置http协议的请求行或请求头的逻辑
      // 建立TCP连接后的回调  
      const oncreate = (err, socket) => {  
        if (called)  
          return;  
        called = true;  
        if (err) {  
          process.nextTick(() => this.emit('error', err));  
          return;  
        }  
        // 建立连接成功,执行回调  
        this.onSocket(socket);  
        // 连接成功后发送数据  
        this._deferToConnect(null, null, () => this._flush());  
      };  

      // 使用agent时,socket由agent提供,否则自己创建socket  
      if (this.agent) {  
        this.agent.addRequest(this, options);  
      } else {  
        // 不使用agent则每次创建一个socket,默认使用net模块的接口
        if (typeof options.createConnection === 'function') {  
          const newSocket = options.createConnection(options, 
                                                          oncreate);  
          if (newSocket && !called) {  
            called = true;  
            this.onSocket(newSocket);  
          } else {  
            return;  
          }  
        } else {  
          this.onSocket(net.createConnection(options));  
        }  
      }  
      // 连接成功后发送待缓存的数据  
      this._deferToConnect(null, null, () => this._flush());  
    }  

获取一个ClientRequest实例后,不管是通过agent还是自己创建一个TCP连接,在连接成功后都会执行onSocket。

    // socket可用时的回调  
    ClientRequest.prototype.onSocket = function onSocket(socket) {  
      process.nextTick(onSocketNT, this, socket);  
    };  

    function onSocketNT(req, socket) {  
      // 申请socket过程中,请求已经终止  
      if (req.aborted) {
        // 不使用agent,直接销毁socekt  
        if (!req.agent) {  
          socket.destroy();  
        } else {  
          // 使用agent触发free事件,由agent处理socekt  
          req.emit('close');  
          socket.emit('free');  
        }  
      } else {  
        // 处理socket  
        tickOnSocket(req, socket);  
      }  
    }  

我们继续看tickOnSocket

    // 初始化HTTP解析器和注册data事件等,等待响应  
    function tickOnSocket(req, socket) {  
      // 分配一个HTTP解析器  
      const parser = parsers.alloc();  
      req.socket = socket;  
      // 初始化,处理响应报文  
      parser.initialize(HTTPParser.RESPONSE,  
             new HTTPClientAsyncResource('HTTPINCOMINGMESSAGE', req),          req.maxHeaderSize || 0,  
             req.insecureHTTPParser === undefined ?  
            isLenient() : req.insecureHTTPParser);  
      parser.socket = socket;  
      parser.outgoing = req;  
      req.parser = parser;  

      socket.parser = parser;  
      // socket正处理的请求  
      socket._httpMessage = req;  

      // Propagate headers limit from request object to parser  
      if (typeof req.maxHeadersCount === 'number') {  
        parser.maxHeaderPairs = req.maxHeadersCount << 1;  
      }  
      // 解析完HTTP头部的回调  
      parser.onIncoming = parserOnIncomingClient;  
      socket.removeListener('error', freeSocketErrorListener);  
      socket.on('error', socketErrorListener);  
      socket.on('data', socketOnData);  
      socket.on('end', socketOnEnd);  
      socket.on('close', socketCloseListener);  
      socket.on('drain', ondrain);  

      if (  
        req.timeout !== undefined ||  
        (req.agent && req.agent.options && 
         req.agent.options.timeout)  
      ) {  
        // 处理超时时间  
        listenSocketTimeout(req);  
      }  
      req.emit('socket', socket);  
    }  

拿到一个socket后,就开始监听socket上http报文的到来。并且申请一个HTTP解析器准备解析http报文,我们主要分析超时时间的处理和data事件的处理逻辑。
1 超时时间的处理

    function listenSocketTimeout(req) {  
      // 设置过了则返回  
      if (req.timeoutCb) {  
        return;  
      }  
      // 超时回调  
      req.timeoutCb = emitRequestTimeout;  
      // Delegate socket timeout event.  
      // 设置socket的超时时间,即socket上一定时间后没有响应则触发超时  
      if (req.socket) {  
        req.socket.once('timeout', emitRequestTimeout);  
      } else {  
        req.on('socket', (socket) => {  
          socket.once('timeout', emitRequestTimeout);  
        });  
      }  
    }  

    function emitRequestTimeout() {  
      const req = this._httpMessage;  
      if (req) {  
        req.emit('timeout');  
      }  
    }  

2 处理响应数据

    function socketOnData(d) {  
      const socket = this;  
      const req = this._httpMessage;  
      const parser = this.parser;  
      // 交给HTTP解析器处理  
      const ret = parser.execute(d);  
      // ...  
    }  

当Node.js收到响应报文时,会把数据交给HTTP解析器处理。http解析在解析的过程中会不断触发钩子函数。我们看一下JS层各个钩子函数的逻辑。
1 解析头部过程中执行的回调

    function parserOnHeaders(headers, url) {  
      // 保存头和url  
      if (this.maxHeaderPairs <= 0 ||  
          this._headers.length < this.maxHeaderPairs) {  
        this._headers = this._headers.concat(headers);  
      }  
      this._url += url;
    }  

2 解析完头部的回调

    function parserOnHeadersComplete(versionMajor, 
                                        versionMinor, 
                                        headers, 
                                        method,  
                                     url, 
                                        statusCode, 
                                        statusMessage, 
                                        upgrade,  
                                     shouldKeepAlive) {  
      const parser = this;  
      const { socket } = parser;  
      // 剩下的HTTP头  
      if (headers === undefined) {  
        headers = parser._headers;  
        parser._headers = [];  
      }  

      if (url === undefined) {  
        url = parser._url;  
        parser._url = '';  
      }  

      // Parser is also used by http client  
      // IncomingMessage  
      const ParserIncomingMessage=(socket && 
                                      socket.server &&  
                                   socket.server[kIncomingMessage]
                                      ) ||                                 
                                      IncomingMessage;  
      // 新建一个IncomingMessage对象  
      const incoming = parser.incoming = new ParserIncomingMessage(socket);  
      incoming.httpVersionMajor = versionMajor;  
      incoming.httpVersionMinor = versionMinor;  
      incoming.httpVersion = `${versionMajor}.${versionMinor}`;  
      incoming.url = url;  
      incoming.upgrade = upgrade;  

      let n = headers.length;  
      // If parser.maxHeaderPairs <= 0 assume that there's no limit.
      if (parser.maxHeaderPairs > 0)  
        n = MathMin(n, parser.maxHeaderPairs);  
      // 更新到保存HTTP头的对象   
      incoming._addHeaderLines(headers, n);  
      // 请求方法或响应行信息  
      if (typeof method === 'number') {  
        // server only  
        incoming.method = methods[method];  
      } else {  
        // client only  
        incoming.statusCode = statusCode;  
        incoming.statusMessage = statusMessage;  
      }  
      // 执行回调  
      return parser.onIncoming(incoming, shouldKeepAlive);  
    }  

我们看到解析完头部后会执行另一个回调onIncoming,并传入IncomingMessage实例,这就是我们平时使用的res。在前面分析过,onIncoming设置的值是parserOnIncomingClient。

    function parserOnIncomingClient(res, shouldKeepAlive) {  
      const socket = this.socket;  
      // 请求对象  
      const req = socket._httpMessage;  
      // 服务器发送了多个响应  
      if (req.res) {  
        socket.destroy();  
        return 0;    
      }  
      req.res = res;  

      if (statusIsInformational(res.statusCode)) {  
        req.res = null;   
        // 请求时设置了expect头,则响应码为100,可以继续发送数据  
        if (res.statusCode === 100) {  
          req.emit('continue');  
        }  
        return 1;   
      }  

      req.res = res;  
      res.req = req;  

      // 等待响应结束,响应结束后会清除定时器  
      res.on('end', responseOnEnd);  
      // 请求终止了或触发response事件,返回false说明没有监听response事件,则丢弃数据  
      if (req.aborted || !req.emit('response', res))  
        res._dump();  

    }  

从源码中我们看出在解析完HTTP响应头时,就执行了http.request设置的回调函数。例如下面代码中的回调。

    http. request('domain', { agent }, (res) => {  
        // 解析body
        res.on('data', (data) => {  
          //   
        });
         // 解析body结束,响应结束
         res.on('end', (data) => {  
          //   
        });  
    });  
    // ...

在回调里我们可以把res作为一个流使用,在解析完HTTP头后,HTTP解析器会继续解析HTTP body。我们看一下HTTP解析器在解析body过程中执行的回调。

    function parserOnBody(b, start, len) {  
      const stream = this.incoming;  
      if (len > 0 && !stream._dumped) {  
        const slice = b.slice(start, start + len);  
        // 把数据push到流中,流会触发data事件  
        const ret = stream.push(slice);  
        // 数据过载,暂停接收  
        if (!ret)  
          readStop(this.socket);  
      }  
    }  

最后我们再看一下解析完body时HTTP解析器执行的回调。

    function parserOnMessageComplete() {  
      const parser = this;  
      const stream = parser.incoming;  

      if (stream !== null) {  
        // body解析完了  
        stream.complete = true;  
        // 在body后可能有trailer头,保存下来  
        const headers = parser._headers;  
        if (headers.length) {  
          stream._addHeaderLines(headers, headers.length);  
          parser._headers = [];  
          parser._url = '';  
        }  
        // 流结束  
        stream.push(null);  
      }  

      // 读取下一个响应,如果有的话  
      readStart(parser.socket);  
    }  

我们看到在解析body过程中会不断往流中push数据,从而不断触发res的data事件,最后解析body结束后,通过push(null)通知流结束,从而触发res.end事件。我们沿着onSocket函数分析完处理响应后我们再来分析请求的过程。执行完http.request后我们会得到一个标记请求的实例。然后执行它的write方法发送数据。

    OutgoingMessage.prototype.write = function write(chunk, encoding, callback) {  
      const ret = write_(this, chunk, encoding, callback, false);  
      // 返回false说明需要等待drain事件  
      if (!ret)  
        this[kNeedDrain] = true;  
      return ret;  
    };  

    function write_(msg, chunk, encoding, callback, fromEnd) {  

      // 还没有设置this._header字段,则把请求行和HTTP头拼接到this._header字段  
      if (!msg._header) {  
        msg._implicitHeader();  
      }  

      let ret;  
      // chunk模式则需要额外加一下字段,否则直接发送  
      if (msg.chunkedEncoding && chunk.length !== 0) {  
        let len;  
        if (typeof chunk === 'string')  
          len = Buffer.byteLength(chunk, encoding);  
        else  
          len = chunk.length;  
        /* 
          chunk模式时,http报文的格式如下 
          chunk长度 回车换行 
          数据 回车换行 
        */  
        msg._send(len.toString(16), 'latin1', null);  
        msg._send(crlf_buf, null, null);  
        msg._send(chunk, encoding, null);  
        ret = msg._send(crlf_buf, null, callback);  
      } else {  
        ret = msg._send(chunk, encoding, callback);  
      }  

      return ret;  
    }  

我们接着看_send函数

    OutgoingMessage.prototype._send = function _send(data, encoding, callback) {  
      // 头部还没有发送  
      if (!this._headerSent) {  
        // 是字符串则追加到头部,this._header保存了HTTP请求行和HTTP头  
        if (typeof data === 'string' &&  
            (encoding === 'utf8' || 
             encoding === 'latin1' || 
             !encoding)) {  
          data = this._header + data;  
        } else {  
          // 否则缓存起来  
          const header = this._header;  
          // HTTP头需要放到最前面  
          if (this.outputData.length === 0) {  
            this.outputData = [{  
              data: header,  
              encoding: 'latin1',  
              callback: null  
            }];  
          } else {  
            this.outputData.unshift({  
              data: header,  
              encoding: 'latin1',  
              callback: null  
            });  
          }  
          // 更新缓存大小  
          this.outputSize += header.length;  
          this._onPendingData(header.length);  
        }  
        // 已经在排队等待发送了,不能修改  
        this._headerSent = true;  
      }  
      return this._writeRaw(data, encoding, callback);  
    };  

我们继续看_writeRaw

    OutgoingMessage.prototype._writeRaw = function _writeRaw(data, encoding, callback) {  

      // 可写的时候直接发送  
      if (conn && conn._httpMessage === this && conn.writable) {  
        // There might be pending data in the this.output buffer.  
        // 如果有缓存的数据则先发送缓存的数据  
        if (this.outputData.length) {  
          this._flushOutput(conn);  
        }  
        // 接着发送当前需要发送的  
        return conn.write(data, encoding, callback);  
      }  
      // 否先缓存  
      this.outputData.push({ data, encoding, callback });  
      this.outputSize += data.length;  
      this._onPendingData(data.length);  
      return this.outputSize < HIGH_WATER_MARK;  
    }  

    OutgoingMessage.prototype._flushOutput = function _flushOutput(socket) {  
      // 之前设置了加塞,则操作socket先积攒数据  
      while (this[kCorked]) {  
        this[kCorked]--;  
        socket.cork();  
      }  

      const outputLength = this.outputData.length;  
      if (outputLength <= 0)  
        return undefined;  

      const outputData = this.outputData;  
      socket.cork();  
      // 把缓存的数据写到socket  
      let ret;  
      for (let i = 0; i < outputLength; i++) {  
        const { data, encoding, callback } = outputData[i];  
        ret = socket.write(data, encoding, callback);  
      }  
      socket.uncork();  

      this.outputData = [];  
      this._onPendingData(-this.outputSize);  
      this.outputSize = 0;  

      return ret;  
    };  

写完数据后,我们还需要执行end函数标记HTTP请求的结束。

    OutgoingMessage.prototype.end = function end(chunk, encoding, callback) {  
      // 还没结束  
      // 加塞  
      if (this.socket) {  
        this.socket.cork();  
      }  

      // 流结束后回调  
      if (typeof callback === 'function')  
        this.once('finish', callback);  
      // 数据写入底层后的回调  
      const finish = onFinish.bind(undefined, this);  
      // chunk模式后面需要发送一个0\r\n结束标记,否则不需要结束标记  
      if (this._hasBody && this.chunkedEncoding) {  
        this._send('0\r\n' + 
                    this._trailer + '\r\n', 'latin1', finish);  
      } else {  
        this._send('', 'latin1', finish);  
      }  
      // uncork解除塞子,发送数据  
      if (this.socket) {  
        // Fully uncork connection on end().  
        this.socket._writableState.corked = 1;  
        this.socket.uncork();  
      }  
      this[kCorked] = 0;  
      // 标记执行了end  
      this.finished = true;  
      // 数据发完了  
      if (this.outputData.length === 0 &&  
          this.socket &&  
          this.socket._httpMessage === this) {  
        this._finish();  
      }  

      return this;  
    };  

18.3 HTTP服务器

本节我们来分析使用Node.js作为服务器的例子。

    const http = require('http');  
    http.createServer((req, res) => {  
      res.write('hello');  
      res.end();  
    })  
    .listen(3000);  

接着我们沿着createServer分析Node.js作为服务器的原理。

    function createServer(opts, requestListener) {  
      return new Server(opts, requestListener);  
    }  

我们看Server的实现

    function Server(options, requestListener) {  
      // 可以自定义表示请求的对象和响应的对象  
      this[kIncomingMessage] = options.IncomingMessage || IncomingMessage;  
      this[kServerResponse] = options.ServerResponse || ServerResponse;  
      // HTTP头个数的阈值  
      const maxHeaderSize = options.maxHeaderSize;  
      this.maxHeaderSize = maxHeaderSize;  
      // 允许半关闭  
      net.Server.call(this, { allowHalfOpen: true });  
      // 有请求时的回调  
      if (requestListener) {  
        this.on('request', requestListener);  
      }  
      // 服务器socket读端关闭时是否允许继续处理队列里的响应(tcp上有多个请求,管道化)  
      this.httpAllowHalfOpen = false;  
      // 有连接时的回调,由net模块触发  
      this.on('connection', connectionListener);  
      // 服务器下所有请求和响应的超时时间  
      this.timeout = 0;  
      // 同一个TCP连接上,两个请求之前最多间隔的时间   
      this.keepAliveTimeout = 5000;  
      this.maxHeadersCount = null;  
      // 解析头部的超时时间,防止ddos  
      this.headersTimeout = 60 * 1000; // 60 seconds  
    }  

接着调用listen函数,因为HTTP Server继承于net.Server,net.Server的listen函数前面我们已经分析过,就不再分析。当有请求到来时,会触发connection事件。从而执行connectionListener。

    function connectionListener(socket) {  
      defaultTriggerAsyncIdScope(  
        getOrSetAsyncId(socket), connectionListenerInternal, this, socket  
      );  
    }  

    // socket表示新连接  
    function connectionListenerInternal(server, socket) {  
      // socket所属server  
      socket.server = server;  
      // 设置连接的超时时间,超时处理函数为socketOnTimeout  
      if (server.timeout && typeof socket.setTimeout === 'function')     socket.setTimeout(server.timeout);  
      socket.on('timeout', socketOnTimeout);  
      // 分配一个HTTP解析器  
      const parser = parsers.alloc();  
      // 解析请求报文  
      parser.initialize(  
        HTTPParser.REQUEST,  
        new HTTPServerAsyncResource('HTTPINCOMINGMESSAGE', socket), 
        server.maxHeaderSize || 0,  
        server.insecureHTTPParser === undefined ?  
          isLenient() : server.insecureHTTPParser,  
      );  
      parser.socket = socket;  
      // 记录开始解析头部的开始时间  
      parser.parsingHeadersStart = nowDate();  
      socket.parser = parser;  
      if (typeof server.maxHeadersCount === 'number') {  
        parser.maxHeaderPairs = server.maxHeadersCount << 1;  
      }  

      const state = {  
        onData: null,  
        onEnd: null,  
        onClose: null,  
        onDrain: null,  
        // 同一TCP连接上,请求和响应的的队列,线头阻塞的原理  
        outgoing: [],  
        incoming: [],  
        // 待发送的字节数,如果超过阈值,则先暂停接收请求的数据  
        outgoingData: 0,  
        /*
          是否重新设置了timeout,用于响应一个请求时,
          标记是否重新设置超时时间的标记  
        */
        keepAliveTimeoutSet: false  
      };  
      // 监听tcp上的数据,开始解析http报文  
      state.onData = socketOnData.bind(undefined, 
                                          server, 
                                          socket, 
                                          parser, 
                                          state);  
      state.onEnd = socketOnEnd.bind(undefined,
                                        server, 
                                        socket, 
                                        parser, 
                                        state);  
      state.onClose = socketOnClose.bind(undefined, socket, state);  
      state.onDrain = socketOnDrain.bind(undefined, socket, state);  
      socket.on('data', state.onData);  
      socket.on('error', socketOnError);  
      socket.on('end', state.onEnd);  
      socket.on('close', state.onClose);  
      socket.on('drain', state.onDrain);  
      // 解析HTTP头部完成后执行的回调  
      parser.onIncoming = parserOnIncoming.bind(undefined, 
                                                    server, 
                                                    socket, 
                                                    state);  
      socket.on('resume', onSocketResume);  
      socket.on('pause', onSocketPause);  

      /*
        如果handle是继承StreamBase的流则执行consume消费http
        请求报文,而不是上面的onData,tcp模块的isStreamBase为true 
      */
      if (socket._handle && socket._handle.isStreamBase &&  
          !socket._handle._consumed) {  
        parser._consumed = true;  
        socket._handle._consumed = true;  
        parser.consume(socket._handle);  
      }  
      parser[kOnExecute] =  
        onParserExecute.bind(undefined, 
                               server, 
                               socket, 
                               parser, 
                               state);  

      socket._paused = false;  
    }  

执行完connectionListener后就开始等待tcp上数据的到来,即HTTP请求报文。上面代码中Node.js监听了socket的data事件,同时注册了钩子kOnExecute。data事件我们都知道是流上有数据到来时触发的事件。我们看一下socketOnData做了什么事情。

    function socketOnData(server, socket, parser, state, d) {  
      // 交给HTTP解析器处理,返回已经解析的字节数  
      const ret = parser.execute(d);  
      onParserExecuteCommon(server, socket, parser, state, ret, d);  
    }  

socketOnData的处理逻辑是当socket上有数据,然后交给HTTP解析器处理。这看起来没什么问题,那么kOnExecute是做什么的呢?kOnExecute钩子函数的值是onParserExecute,这个看起来也是解析tcp上的数据的,看起来和onSocketData是一样的作用,难道tcp上的数据有两个消费者?我们看一下kOnExecute什么时候被回调的。

    void OnStreamRead(ssize_t nread, const uv_buf_t& buf) override {  

        Local<Value> ret = Execute(buf.base, nread);  
        Local<Value> cb =  
            object()->Get(env()->context(), kOnExecute).ToLocalChecked();  
        MakeCallback(cb.As<Function>(), 1, &ret);  
      }  

OnStreamRead是node_http_parser.cc实现的函数,所以kOnExecute在node_http_parser.cc中的OnStreamRead中被回调,那么OnStreamRead又是什么时候被回调的呢?在C++层章节我们分析过,OnStreamRead是Node.js中C++层流操作的通用函数,当流有数据的时候就会执行该回调。而且OnStreamRead中也会把数据交给HTTP解析器解析。这看起来真的有两个消费者?这就很奇怪,为什么一份数据会交给HTTP解析器处理两次?

    if (socket._handle && socket._handle.isStreamBase && !socket._handle._consumed) {  
      parser._consumed = true;  
      socket._handle._consumed = true;  
      parser.consume(socket._handle);  
    }  

因为TCP流是继承StreamBase类的,所以if成立。我们看一下consume的实现。

    static void Consume(const FunctionCallbackInfo<Value>& args) {  
      Parser* parser;  
      ASSIGN_OR_RETURN_UNWRAP(&parser, args.Holder());  
      CHECK(args[0]->IsObject());  
      StreamBase* stream = StreamBase::FromObjject(args[0].As<Object>());  
      CHECK_NOT_NULL(stream);  
      stream->PushStreamListener(parser);  
    }  

HTTP解析器把自己注册为TCP stream的一个listener。这会使得TCP流上的数据由node_http_parser.cc的OnStreamRead直接消费,而不是触发onData事件。在OnStreamRead中会源源不断地把数据交给HTTP解析器处理,在解析的过程中,会不断触发对应的钩子函数,直到解析完HTTP头部后执行parserOnIncoming。

    function parserOnIncoming(server, socket, state, req, keepAlive) {  
      // 需要重置定时器  
      resetSocketTimeout(server, socket, state);  
      // 设置了keepAlive则响应后需要重置一些状态  
      if (server.keepAliveTimeout > 0) {  
        req.on('end', resetHeadersTimeoutOnReqEnd);  
      }  

      // 标记头部解析完毕  
      socket.parser.parsingHeadersStart = 0;  

      // 请求入队(待处理的请求队列)  
      state.incoming.push(req);  

      if (!socket._paused) {  
        const ws = socket._writableState;  
        // 待发送的数据过多,先暂停接收请求数据  
        if (ws.needDrain || 
            state.outgoingData >= socket.writableHighWaterMark) {  
          socket._paused = true;  
          socket.pause();  
        }  
      }  
      // 新建一个表示响应的对象  
      const res = new server[kServerResponse](req);  
      // 设置数据写入待发送队列时触发的回调,见OutgoingMessage  
      res._onPendingData = updateOutgoingData.bind(undefined, 
                                                        socket, 
                                                        state);  
      // 根据请求的HTTP头设置是否支持keepalive(管道化)  
      res.shouldKeepAlive = keepAlive;  
      /*
         socket当前已经在处理其它请求的响应,则先排队,
         否则挂载响应对象到socket,作为当前处理的响应  
      */
      if (socket._httpMessage) {  
        state.outgoing.push(res);  
      } else {  
        res.assignSocket(socket);  
      }  

      // 响应处理完毕后,需要做一些处理  
      res.on('finish',  
             resOnFinish.bind(undefined, 
                                req, 
                                res, 
                                socket, 
                                state, 
                                server));  
      // 有expect请求头,并且是http1.1  
      if (req.headers.expect !== undefined &&  
          (req.httpVersionMajor === 1 && 
           req.httpVersionMinor === 1)
         ) {  
        // Expect头的值是否是100-continue  
        if (continueExpression.test(req.headers.expect)) {  
          res._expect_continue = true;  
          /*
            监听了checkContinue事件则触发,
            否则直接返回允许继续请求并触发request事件
           */  
          if (server.listenerCount('checkContinue') > 0) {  
            server.emit('checkContinue', req, res);  
          } else {  
            res.writeContinue();  
            server.emit('request', req, res);  
          }  
        } else if (server.listenerCount('checkExpectation') > 0) {  
          /*
            值异常,监听了checkExpectation事件,
            则触发,否则返回417拒绝请求 
           */ 
          server.emit('checkExpectation', req, res);  
        } else {  
          res.writeHead(417);  
          res.end();  
        }  
      } else {  
        // 触发request事件说明有请求到来  
        server.emit('request', req, res);  
      }  
      return 0;  // No special treatment.  
    }  

我们看到这里会触发request事件通知用户有新请求到来,用户就可以处理请求了。我们看到Node.js解析头部的时候就会执行上层回调,通知有新请求到来,并传入request和response作为参数,分别对应的是表示请求和响应的对象。另外Node.js本身是不会解析body部分的,我们可以通过以下方式获取body的数据。

    const server = http.createServer((request, response) => {  
      request.on('data', (chunk) => {  
       // 处理body  
      });  
      request.on('end', () => {  
       // body结束  
      });  
    })  

18.3.1 HTTP管道化的原理和实现

HTTP1.0的时候,不支持管道化,客户端发送一个请求的时候,首先建立TCP连接,然后服务器返回一个响应,最后断开TCP连接,这种是最简单的实现方式,但是每次发送请求都需要走三次握手显然会带来一定的时间损耗,所以HTTP1.1的时候,支持了管道化。管道化的意思就是可以在一个TCP连接上发送多个请求,这样服务器就可以同时处理多个请求,但是由于HTTP1.1的限制,多个请求的响应需要按序返回。因为在HTTP1.1中,没有标记请求和响应的对应关系。所以HTTP客户端会假设第一个返回的响应是对应第一个请求的。如果乱序返回,就会导致问题,如图18-2所示。

图18-2
而在HTTP 2.0中,每个请求会分配一个id,响应中也会返回对应的id,这样就算乱序返回,HTTP客户端也可以知道响应所对应的请求。在HTTP 1.1这种情况下,HTTP服务器的实现就会变得复杂,服务器可以以串行的方式处理请求,当前面请求的响应返回到客户端后,再继续处理下一个请求,这种实现方式是相对简单的,但是很明显,这种方式相对来说还是比较低效的,另一种实现方式是并行处理请求,串行返回,这样可以让请求得到尽快的处理,比如两个请求都访问数据库,那并行处理两个请求就会比串行快得多,但是这种实现方式相对比较复杂,Node.js就是属于这种方式,下面我们来看一下Node.js中是如何实现的。前面分析过,Node.js在解析完HTTP头部的时候会执行parserOnIncoming。

    function parserOnIncoming(server, socket, state, req, keepAlive) {  
      // 标记头部解析完毕  
      socket.parser.parsingHeadersStart = 0;  
      // 请求入队  
      state.incoming.push(req);  
      // 新建一个表示响应的对象,一般是ServerResponse  
      const res = new server[kServerResponse](req);  
      /*
        socket当前已经在处理其它请求的响应,则先排队,
       否则挂载响应对象到socket,作为当前处理的响应
      */  
      if (socket._httpMessage) {  
        state.outgoing.push(res);  
      } else {  
        res.assignSocket(socket); // socket._httpMessage = res;  
      }  
      // 响应处理完毕后,需要做一些处理  
      res.on('finish', resOnFinish.bind(undefined, 
                                            req, 
                                            res, 
                                            socket, 
                                            state, 
                                            server));  
      // 触发request事件说明有请求到来  
      server.emit('request', req, res);  
      return 0;  
    }  

当Node.js解析HTTP请求头完成后,就会创建一个ServerResponse对象表示响应。然后判断当前是否有正在处理的响应,如果有则排队等待处理,否则把新建的ServerResponse对象作为当前需要处理的响应。最后触发request事件通知用户层。用户就可以进行请求的处理了。我们看到Node.js维护了两个队列,分别是请求和响应队列,如图18-3所示。

图18-3
当前处理的请求在请求队列的队首,该请求对应的响应会挂载到socket的_httpMessage属性上。但是我们看到Node.js会触发request事件通知用户有新请求到来,所有在管道化的情况下,Node.js会并行处理多个请求(如果是cpu密集型的请求则实际上还是会变成串行,这和Node.js的单线程相关)。那Node.js是如何控制响应的顺序的呢?我们知道每次触发request事件的时候,我们都会执行一个函数。比如下面的代码。

     http.createServer((req, res) => {  
      // 一些网络IO  
      res.writeHead(200, { 'Content-Type': 'text/plain' });  
      res.end('okay');  
    });  

我们看到每个请求的处理是独立的。假设每个请求都去操作数据库,如果请求2比请求1先完成数据库的操作,从而请求2先执行res.write和res.end。那岂不是请求2先返回?我们看一下ServerResponse和OutgoingMessage的实现,揭开迷雾。ServerResponse是OutgoingMessage的子类。write函数是在OutgoingMessage中实现的,write的调用链路很长,我们不层层分析,直接看最后的节点。

    function _writeRaw(data, encoding, callback) {  
      const conn = this.socket;  
      // socket对应的响应是自己并且可写  
      if (conn && conn._httpMessage === this && conn.writable) {  
        // 如果有缓存的数据则先发送缓存的数据  
        if (this.outputData.length) {  
          this._flushOutput(conn);  
        }  
        // 接着发送当前需要发送的  
        return conn.write(data, encoding, callback);  
      }  
      // socket当前处理的响应对象不是自己,则先缓存数据。  
      this.outputData.push({ data, encoding, callback });  
      this.outputSize += data.length;  
      this._onPendingData(data.length);  
      return this.outputSize < HIGH_WATER_MARK;  
    }  

我们看到我们调用res.write的时候,Node.js会首先判断,res是不是属于当前处理中响应,如果是才会真正发送数据,否则会先把数据缓存起来。分析到这里,相信大家已经差不多明白Node.js是如何控制响应按序返回的。最后我们看一下这些缓存的数据什么时候会被发送出去。前面代码已经贴过,当一个响应结束的时候,Node.js会做一些处理。

    res.on('finish', resOnFinish.bind(undefined, 
                                         req, 
                                         res, 
                                         socket, 
                                         state, 
                                         server));  

我们看看resOnFinish

    function resOnFinish(req, res, socket, state, server) {  
      // 删除响应对应的请求  
      state.incoming.shift();  
      clearIncoming(req);  
      // 解除socket上挂载的响应对象  
      res.detachSocket(socket);  
      req.emit('close');  
      process.nextTick(emitCloseNT, res);  
      // 是不是最后一个响应  
      if (res._last) {  
        // 是则销毁socket  
        if (typeof socket.destroySoon === 'function') {  
          socket.destroySoon();  
        } else {  
          socket.end();  
        }  
      } else if (state.outgoing.length === 0) {  
        /*
          没有待处理的响应了,则重新设置超时时间,
          等待请求的到来,一定时间内没有请求则触发timeout事件
        */  
        if (server.keepAliveTimeout && 
             typeof socket.setTimeout === 'function') {  
          socket.setTimeout(server.keepAliveTimeout);  
          state.keepAliveTimeoutSet = true;  
        }  
      } else {  
        // 获取下一个要处理的响应  
        const m = state.outgoing.shift();  
        // 挂载到socket作为当前处理的响应  
        if (m) {  
          m.assignSocket(socket);  
        }  
      }  
    }  

我们看到,Node.js处理完一个响应后,会做一些判断。分别有三种情况,我们分开分析。
1 是否是最后一个响应
什么情况下,会被认为是最后一个响应的?因为响应和请求是一一对应的,最后一个响应就意味着最后一个请求了,那么什么时候被认为是最后一个请求呢?当非管道化的情况下,一个请求一个响应,然后关闭TCP连接,所以非管道化的情况下,tcp上的第一个也是唯一一个请求就是最后一个请求。在管道化的情况下,理论上就没有所谓的最后一个响应。但是实现上会做一些限制。在管道化的情况下,每一个响应可以通过设置HTTP响应头connection来定义是否发送该响应后就断开连接,我们看一下Node.js的实现。

    // 是否显示删除过connection头,是则响应后断开连接,并标记当前响应是最后一个  
     if (this._removedConnection) {  
       this._last = true;  
       this.shouldKeepAlive = false;  
     } else if (!state.connection) {  
       /* 
        没有显示设置了connection头,则取默认行为 
        1 Node.js的shouldKeepAlive默认为true,也可以根据请求报文里
          的connection头定义
       2 设置content-length或使用chunk模式才能区分响应报文编边界,
          才能支持keepalive 
       3 使用了代理,代理是复用TCP连接的,支持keepalive 
       */  
       const shouldSendKeepAlive = this.shouldKeepAlive &&  
           (state.contLen || 
             this.useChunkedEncodingByDefault || 
             this.agent);  
       if (shouldSendKeepAlive) {  
         header += 'Connection: keep-alive\r\n';  
       } else {  
         this._last = true;  
         header += 'Connection: close\r\n';  
       }  
     }  

另外当读端关闭的时候,也被认为是最后一个请求,毕竟不会再发送请求了。我们看一下读端关闭的逻辑。

    function socketOnEnd(server, socket, parser, state) {  
      const ret = parser.finish();  

      if (ret instanceof Error) {  
        socketOnError.call(socket, ret);  
        return;  
      }  
      // 不允许半开关则终止请求的处理,不响应,关闭写端  
      if (!server.httpAllowHalfOpen) {  
        abortIncoming(state.incoming);  
        if (socket.writable) socket.end();  
      } else if (state.outgoing.length) {  
        /*
          允许半开关,并且还有响应需要处理,
          标记响应队列最后一个节点为最后的响应,
          处理完就关闭socket写端
        */  
        state.outgoing[state.outgoing.length - 1]._last = true;  
      } else if (socket._httpMessage) {  
        /*
          没有等待处理的响应了,但是还有正在处理的响应,
          则标记为最后一个响应
         */  
        socket._httpMessage._last = true;  
      } else if (socket.writable) {  
        // 否则关闭socket写端  
        socket.end();  
      }  
    }  

以上就是Node.js中判断是否是最后一个响应的情况,如果一个响应被认为是最后一个响应,那么发送响应后就会关闭连接。
2 响应队列为空
我们继续看一下如果不是最后一个响应的时候,Node.js又是怎么处理的。如果当前的待处理响应队列为空,说明当前处理的响应是目前最后一个需要处理的,但是不是TCP连接上最后一个响应,这时候,Node.js会设置超时时间,如果超时还没有新的请求,则Node.js会关闭连接。
3 响应队列非空
如果当前待处理队列非空,处理完当前请求后会继续处理下一个响应。并从队列中删除该响应。我们看一下Node.js是如何处理下一个响应的。

    // 把响应对象挂载到socket,标记socket当前正在处理的响应  
    ServerResponse.prototype.assignSocket = function assignSocket(socket) {  
      // 挂载到socket上,标记是当前处理的响应  
      socket._httpMessage = this;  
      socket.on('close', onServerResponseClose);  
      this.socket = socket;  
      this.emit('socket', socket);  
      this._flush();  
    };  

我们看到Node.js是通过_httpMessage标记当前处理的响应的,配合响应队列来实现响应的按序返回。标记完后执行_flush发送响应的数据(如果这时候请求已经被处理完成)

    OutgoingMessage.prototype._flush = function _flush() {  
      const socket = this.socket;  
      if (socket && socket.writable) {  
        const ret = this._flushOutput(socket);  
    };  

    OutgoingMessage.prototype._flushOutput = function _flushOutput(socket) {  
      // 之前设置了加塞,则操作socket先积攒数据  
      while (this[kCorked]) {  
        this[kCorked]--;  
        socket.cork();  
      }  

      const outputLength = this.outputData.length;  
      // 没有数据需要发送  
      if (outputLength <= 0)  
        return undefined;  

      const outputData = this.outputData;  
      // 加塞,让数据一起发送出去  
      socket.cork();  
      // 把缓存的数据写到socket  
      let ret;  
      for (let i = 0; i < outputLength; i++) {  
        const { data, encoding, callback } = outputData[i];  
        ret = socket.write(data, encoding, callback);  
      }  
      socket.uncork();  

      this.outputData = [];  
      this._onPendingData(-this.outputSize);  
      this.outputSize = 0;  

      return ret;  
    }  

以上就是Node.js中对于管道化的实现。

18.3.2 HTTP Connect方法的原理和实现

分析HTTP Connect实现之前我们首先看一下为什么需要HTTP Connect方法或者说它出现的背景。Connect方法主要用于代理服务器的请求转发。我们看一下传统HTTP服务器的工作原理,如图18-4所示。

图18-4
1 客户端和代理服务器建立TCP连接
2 客户端发送HTTP请求给代理服务器
3 代理服务器解析HTTP协议,根据配置拿到业务服务器的地址
4 代理服务器和业务服务器建立TCP连接,通过HTTP协议或者其它协议转发请求
5 业务服务器返回数据,代理服务器回复HTTP报文给客户端。

接着我们看一下HTTPS服务器的原理。
1 客户端和服务器建立TCP连接
2 服务器通过TLS报文返回证书信息,并和客户端完成后续的TLS通信。
3 完成TLS通信后,后续发送的HTTP报文会经过TLS层加密解密后再传输。

那么如果我们想实现一个HTTPS的代理服务器怎么做呢?因为客户端只管和直接相连的服务器进行HTTPS的通信,如果我们的业务服务器前面还有代理服务器,那么代理服务器就必须要有证书才能和客户端完成TLS握手,从而进行HTTPS通信。代理服务器和业务服务器使用HTTP或者HTTPS还是其它协议都可以。这样就意味着我们所有的服务的证书都需要放到代理服务器上,这种场景的限制是,代理服务器和业务服务器都由我们自己管理或者公司统一管理。如果我们想加一个代理对业务服务器不感知那怎么办呢(比如写一个代理服务器用于开发调试)?有一种方式就是为我们的代理服务器申请一个证书,这样客户端和代理服务器就可以完成正常的HTTPS通信了。从而也就可以完成代理的功能。另外一种方式就是HTTP Connect方法。HTTP Connect方法的作用是指示服务器帮忙建立一条TCP连接到真正的业务服务器,并且透传后续的数据,这样不申请证书也可以完成代理的功能,如图18-5所示。

图18-5
这时候代理服务器只负责透传两端的数据,不像传统的方式一样解析请求然后再转发。这样客户端和业务服务器就可以自己完成TLS握手和HTTPS通信。代理服务器就像不存在一样。了解了Connect的原理后看一下来自Node.js官方的一个例子。

    const http = require('http');  
    const net = require('net');  
    const { URL } = require('url');  
    // 创建一个HTTP服务器作为代理服务器  
    const proxy = http.createServer((req, res) => {  
      res.writeHead(200, { 'Content-Type': 'text/plain' });  
      res.end('okay');  
    });  
    // 监听connect事件,有http connect请求时触发  
    proxy.on('connect', (req, clientSocket, head) => {  
      // 获取真正要连接的服务器地址并发起连接  
      const { port, hostname } = new URL(`http://${req.url}`);  
      const serverSocket = net.connect(port || 80, hostname, () => {  
        // 连接成功告诉客户端  
        clientSocket.write('HTTP/1.1 200 Connection Established\r\n' +  
                        'Proxy-agent: Node.js-Proxy\r\n' +  
                        '\r\n');  
        // 透传客户端和服务器的数据    
        serverSocket.write(head);              
        serverSocket.pipe(clientSocket);  
        clientSocket.pipe(serverSocket);  
      });  
    });  

    proxy.listen(1337, '127.0.0.1', () => {  

      const options = {  
        port: 1337,  
        // 连接的代理服务器地址  
        host: '127.0.0.1',  
        method: 'CONNECT',  
        // 我们需要真正想访问的服务器地址  
        path: 'www.baidu.com',  
      };  
      // 发起http connect请求  
      const req = http.request(options);  
      req.end();  
      // connect请求成功后触发  
      req.on('connect', (res, socket, head) => {  
        // 发送真正的请求  
        socket.write('GET / HTTP/1.1\r\n' +  
                     'Host: www.baidu.com\r\n' +  
                     'Connection: close\r\n' +  
                     '\r\n');  
        socket.on('data', (chunk) => {  
          console.log(chunk.toString());  
        });  
        socket.on('end', () => {  
          proxy.close();  
        });  
      });  
    });  

官网的这个例子很好地说明了Connect的原理,如图18-6所示。

图18-6
下面我们看一下Node.js中Connect的实现。我们从HTTP Connect请求开始。之前已经分析过,客户端和Node.js服务器建立TCP连接后,Node.js收到数据的时候会交给HTTP解析器处理,

    // 连接上有数据到来  
    function socketOnData(server, socket, parser, state, d) {  
      // 交给HTTP解析器处理,返回已经解析的字节数  
      const ret = parser.execute(d);  
      onParserExecuteCommon(server, socket, parser, state, ret, d);  
    }  

HTTP解析数据的过程中会不断回调Node.js的回调,然后执行onParserExecuteCommon。我们这里只关注当Node.js解析完所有HTTP请求头后执行parserOnHeadersComplete。

    function parserOnHeadersComplete(versionMajor, versionMinor, headers, method,  
                                     url, statusCode, statusMessage, upgrade,  
                                     shouldKeepAlive) {  
      const parser = this;  
      const { socket } = parser;  

      // IncomingMessage  
      const ParserIncomingMessage = (socket && socket.server &&  
                                     socket.server[kIncomingMessage]) ||  
                                     IncomingMessage;  
      // 新建一个IncomingMessage对象  
      const incoming = parser.incoming = new ParserIncomingMessage(socket);  
      incoming.httpVersionMajor = versionMajor;  
      incoming.httpVersionMinor = versionMinor;  
      incoming.httpVersion = `${versionMajor}.${versionMinor}`;  
      incoming.url = url;  
      // 是否是connect请求或者upgrade请求  
      incoming.upgrade = upgrade;  

      // 执行回调  
      return parser.onIncoming(incoming, shouldKeepAlive);  
    }  

我们看到解析完HTTP头后,Node.js会创建一个表示请求的对象IncomingMessage,然后回调onIncoming。

    function parserOnIncoming(server, socket, state, req, keepAlive) {  
      // 请求是否是connect或者upgrade  
      if (req.upgrade) {  
        req.upgrade = req.method === 'CONNECT' ||  
                      server.listenerCount('upgrade') > 0;  
        if (req.upgrade)  
          return 2;  
      }  
     // ...  
    }  

Node.js解析完头部并且执行了响应的钩子函数后,会执行onParserExecuteCommon。

    function onParserExecuteCommon(server, socket, parser, state, ret, d) {  
      if (ret instanceof Error) {  
        prepareError(ret, parser, d);  
        ret.rawPacket = d || parser.getCurrentBuffer();  
        socketOnError.call(socket, ret);  
      } else if (parser.incoming && parser.incoming.upgrade) {  
        // 处理Upgrade或者CONNECT请求  
        const req = parser.incoming;  
        const eventName = req.method === 'CONNECT' ? 
                           'connect' : 'upgrade';  
        // 监听了对应的事件则处理,否则关闭连接  
        if (eventName === 'upgrade' || 
            server.listenerCount(eventName) > 0) {  
          // 还没有解析的数据  
          const bodyHead = d.slice(ret, d.length);  
          socket.readableFlowing = null;  
          server.emit(eventName, req, socket, bodyHead);  
        } else {  
          socket.destroy();  
        }  
      }  
    }  

这时候Node.js会判断请求是不是Connect或者协议升级的upgrade请求,是的话继续判断是否有处理该事件的函数,没有则关闭连接,否则触发对应的事件进行处理。所以这时候Node.js会触发Connect方法。Connect事件的处理逻辑正如我们开始给出的例子中那样。我们首先和真正的服务器建立TCP连接,然后返回响应头给客户端,后续客户就可以和真正的服务器真正进行TLS握手和HTTPS通信了。这就是Node.js中Connect的原理和实现。

不过在代码中我们发现一个好玩的地方。那就是在触发connect事件的时候,Node.js给回调函数传入的参数。

    server.emit('connect', req, socket, bodyHead);  

第一第二个参数没什么特别的,但是第三个参数就有意思了,bodyHead代表的是HTTP Connect请求中除了请求行和HTTP头之外的数据。因为Node.js解析完HTTP头后就不继续处理了。把剩下的数据交给了用户。我们来做一些好玩的事情。

    const http = require('http');  
    const net = require('net');  
    const { URL } = require('url');  

    const proxy = http.createServer((req, res) => {  
      res.writeHead(200, { 'Content-Type': 'text/plain' });  
      res.end('okay');  
    });  
    proxy.on('connect', (req, clientSocket, head) => {  
      const { port, hostname } = new URL(`http://${req.url}`);  
      const serverSocket = net.connect(port || 80, hostname, () => {  
        clientSocket.write('HTTP/1.1 200 Connection Established\r\n' +  
                        'Proxy-agent: Node.js-Proxy\r\n' +  
                        '\r\n');  
        // 把connect请求剩下的数据转发给服务器                 
        serverSocket.write(head);  
        serverSocket.pipe(clientSocket);  
        clientSocket.pipe(serverSocket);  
      });  
    });  

    proxy.listen(1337, '127.0.0.1', () => {  
      const net = require('net');  
      const body = 'GET http://www.baidu.com:80 HTTP/1.1\r\n\r\n';  
      const length = body.length;  
      const socket = net.connect({host: '127.0.0.1', port: 1337});  
      socket.write(`CONNECT www.baidu.com:80 HTTP/1.1\r\n\r\n${body}`);  
      socket.setEncoding('utf-8');  
      socket.on('data', (chunk) => {  
       console.log(chunk)  
      });  
    });  

我们新建一个socket,然后自己构造HTTP Connect报文,并且在HTTP行后面加一个额外的字符串,这个字符串是两一个HTTP请求。当Node.js服务器收到Connect请求后,我们在connect事件的处理函数中,把Connect请求多余的那一部分数据传给真正的服务器。这样就节省了发送一个请求的时间。

18.3.3 超时管理

在解析HTTP协议或者支持长连接的时候,Node.js需要设置一些超时的机制,否则会造成攻击或者资源浪费。下面我们看一下HTTP服务器中涉及到超时的一些逻辑。 1 解析HTTP头部超时
当收到一个HTTP请求报文时,会从HTTP请求行,HTTP头,HTTP body的顺序进行解析,如果用户构造请求,只发送HTTP头的一部分。那么HTTP解析器就会一直在等待后续数据的到来。这会导致DDOS攻击,所以Node.js中设置了解析HTTP头的超时时间,阈值是60秒。如果60秒内没有解析完HTTP头部,则会触发timeout事件。如果用户不处理,则Node.js会自动关闭连接。我们看一下Node.js的实现。Node.js在初始化的时候会设置超时时间。

    this.headersTimeout = 60 * 1000; // 60 seconds  
    //  Node.js在建立TCP连接成功后初始化解析HTTP头的开始时间。
    function connectionListenerInternal(server, socket) {  
      parser.parsingHeadersStart = nowDate();  
    }  

然后在每次收到数据的时候判断HTTP头部是否解析完成,如果没有解析完成并且超时了则会触发timeout事件。

    function onParserExecute(server, socket, parser, state, ret) {  
      socket._unrefTimer();  
      const start = parser.parsingHeadersStart;  
      // start等于0,说明HTTP头已经解析完毕,否则说明正在解析头,然后再判断解析时间是否超时了  
      if (start !== 0 && nowDate() - start > server.headersTimeout) {  
        // 触发timeout,如果没有监听timeout,则默认会销毁socket,即关闭连接  
        const serverTimeout = server.emit('timeout', socket);  

        if (!serverTimeout)  
          socket.destroy();  
        return;  
      }  

      onParserExecuteCommon(server, socket, parser, state, ret, undefined);  
    }  

如果在超时之前解析HTTP头完成,则把parsingHeadersStart置为0表示解析完成。

    function parserOnIncoming(server, socket, state, req, keepAlive) {  
      // 设置了keepAlive则响应后需要重置一些状态  
      if (server.keepAliveTimeout > 0) {  
        req.on('end', resetHeadersTimeoutOnReqEnd);  
      }  

      // 标记头部解析完毕  
      socket.parser.parsingHeadersStart = 0;  
    }  

    function resetHeadersTimeoutOnReqEnd() {  
      if (parser) {  
        parser.parsingHeadersStart = nowDate();  
      }  
    }  

另外如果支持长连接,即一个TCP连接上可以发送多个请求。则在每个响应结束之后,需要重新初始化解析HTTP头的开始时间。当下一个请求数据到来时再次判断解析HTTP头部是否超时。这里是响应结束后就开始计算。而不是下一个请求到来时。 2 支持管道化的情况下,多个请求的时间间隔
Node.js支持在一个TCP连接上发送多个HTTP请求,所以需要设置一个定时器,如果超时都没有新的请求到来,则触发超时事件。这里涉及定时器的设置和重置。

    // 是不是最后一个响应  
      if (res._last) {  
        // 是则销毁socket  
        if (typeof socket.destroySoon === 'function') {  
          socket.destroySoon();  
        } else {  
          socket.end();  
        }  
      } else if (state.outgoing.length === 0) {  
        // 没有待处理的响应了,则重新设置超时时间,等待请求的到来,一定时间内没有请求则触发timeout事件  
        if (server.keepAliveTimeout && typeof socket.setTimeout === 'function') {  
          socket.setTimeout(server.keepAliveTimeout);  
          state.keepAliveTimeoutSet = true;  
        }  
      }  

每次响应结束的时候,Node.js首先会判断当前响应是不是最后一个,例如读端不可读了,说明不会又请求到来了,也不会有响应了,那么就不需要保持这个TCP连接。如果当前响应不是最后一个,则Node.js会根据keepAliveTimeout的值做下一步判断,如果keepAliveTimeout 非空,则设置定时器,如果keepAliveTimeout 时间内都没有新的请求则触发timeout事件。那么如果有新请求到来,则需要重置这个定时器。Node.js在收到新请求的第一个请求包中,重置该定时器。

    function onParserExecuteCommon(server, socket, parser, state, ret, d) {  
      resetSocketTimeout(server, socket, state);  
    }  

    function resetSocketTimeout(server, socket, state) {  
      if (!state.keepAliveTimeoutSet)  
        return;  

      socket.setTimeout(server.timeout || 0);  
      state.keepAliveTimeoutSet = false;  
    }  

onParserExecuteCommon会在每次收到数据时执行,然后Node.js会重置定时器为server.timeout的值。

18.4 Agent

本节我们先分析Agent模块的实现,Agent对TCP连接进行了池化管理。简单的情况下,客户端发送一个HTTP请求之前,首先建立一个TCP连接,收到响应后会立刻关闭TCP连接。但是我们知道TCP的三次握手是比较耗时的。所以如果我们能复用TCP连接,在一个TCP连接上发送多个HTTP请求和接收多个HTTP响应,那么在性能上面就会得到很大的提升。Agent的作用就是复用TCP连接。不过Agent的模式是在一个TCP连接上串行地发送请求和接收响应,不支持HTTP PipeLine模式。下面我们看一下Agent模块的具体实现。看它是如何实现TCP连接复用的。

    function Agent(options) {  
      if (!(this instanceof Agent))  
        return new Agent(options);  
      EventEmitter.call(this);  
      this.defaultPort = 80;  
      this.protocol = 'http:';  
      this.options = { ...options };  
      // path字段表示是本机的进程间通信时使用的路径,比如Unix域路径  
      this.options.path = null;  
      // socket个数达到阈值后,等待空闲socket的请求  
      this.requests = {};  
      // 正在使用的socket  
      this.sockets = {};  
      // 空闲socket  
      this.freeSockets = {};  
      // 空闲socket的存活时间  
      this.keepAliveMsecs = this.options.keepAliveMsecs || 1000;  
      /* 
        用完的socket是否放到空闲队列, 
          开启keepalive才会放到空闲队列, 
          不开启keepalive 
            还有等待socket的请求则复用socket 
            没有等待socket的请求则直接销毁socket 
      */  
      this.keepAlive = this.options.keepAlive || false;  
      // 最大的socket个数,包括正在使用的和空闲的socket  
      this.maxSockets = this.options.maxSockets 
                          || Agent.defaultMaxSockets;  
      // 最大的空闲socket个数  
      this.maxFreeSockets = this.options.maxFreeSockets || 256;  
    }  

Agent维护了几个数据结构,分别是等待socket的请求、正在使用的socket、空闲socket。每一个数据结构是一个对象,对象的key是根据HTTP请求参数计算的。对象的值是一个队列。具体结构如图18-7所示。

图18-7
下面我们看一下Agent模块的具体实现。

18.4.1 key的计算

key的计算是池化管理的核心。正确地设计key的计算规则,才能更好地利用池化带来的好处。

    // 一个请求对应的key  
    Agent.prototype.getName = function getName(options) {  
      let name = options.host || 'localhost'; 
      name += ':';  
      if (options.port)  
        name += options.port;  
      name += ':';  
      if (options.localAddress)  
        name += options.localAddress;  
      if (options.family === 4 || options.family === 6)  
        name += `:${options.family}`;  
      if (options.socketPath)  
        name += `:${options.socketPath}`; 
      return name;  
    };  

我们看到key由host、port、本地地址、地址簇类型、unix路径计算而来。所以不同的请求只有这些因子都一样的情况下才能复用连接。另外我们看到Agent支持Unix域。

18.4.2 创建一个socket

    function createSocket(req, options, cb) {  
      options = { ...options, ...this.options };  
      // 计算key
      const name = this.getName(options);  
      options._agentKey = name;  
      options.encoding = null;  
      let called = false;  
      // 创建socket完毕后执行的回调
      const oncreate = (err, s) => {  
        if (called)  
          return;  
        called = true;  
        if (err)  
          return cb(err);  
        if (!this.sockets[name]) {  
          this.sockets[name] = [];  
        }  
        // 插入正在使用的socket队列  
        this.sockets[name].push(s); 
         // 监听socket的一些事件,用于回收socket 
        installListeners(this, s, options); 
        // 有可用socket,通知调用方 
        cb(null, s);  
      };  
      // 创建一个新的socket,使用net.createConnection  
      const newSocket = this.createConnection(options, oncreate);  
      if (newSocket)  
        oncreate(null, newSocket);  
    }  

    function installListeners(agent, s, options) {  
      /*
        socket触发空闲事件的处理函数,告诉agent该socket空闲了,
        agent会回收该socket到空闲队列  
      */
      function onFree() {  
        agent.emit('free', s, options);  
      }  
      /* 
        监听socket空闲事件,调用方使用完socket后触发,
        通知agent socket用完了 
      */ 
      s.on('free', onFree);  

      function onClose(err) {  
        agent.removeSocket(s, options);  
      }  
      // socket关闭则agent会从socket队列中删除它  
      s.on('close', onClose);  

      function onRemove() {  
        agent.removeSocket(s, options);  
        s.removeListener('close', onClose);  
        s.removeListener('free', onFree);  
        s.removeListener('agentRemove', onRemove);  
      }  
      // agent被移除  
      s.on('agentRemove', onRemove);  

    }  

创建socket的主要逻辑如下
1 调用net模块创建一个socket(TCP或者Unix域),然后插入使用中的socket队列,最后通知调用方socket创建成功。
2 监听socket的close、free事件和agentRemove事件,触发时从队列中删除socket。

18.4.3 删除socket

    // 把socket从正在使用队列或者空闲队列中移出  
    function removeSocket(s, options) {  
      const name = this.getName(options);  
      const sets = [this.sockets];  
      /*
        socket不可写了,则有可能是存在空闲的队列中,
        所以需要遍历空闲队列,因为removeSocket只会在
        使用完socket或者socket关闭的时候被调用,前者只有在
        可写状态时会调用,后者是不可写的
      */
      if (!s.writable)  
        sets.push(this.freeSockets);  
      // 从队列中删除对应的socket  
      for (const sockets of sets) {  
        if (sockets[name]) {  
          const index = sockets[name].indexOf(s);  
          if (index !== -1) {  
            sockets[name].splice(index, 1);  
            // Don't leak  
            if (sockets[name].length === 0)  
              delete sockets[name];  
          }  
        }  
      }  
      /* 
        如果还有在等待socekt的请求,则创建socket去处理它, 
        因为socket数已经减一了,说明socket个数还没有达到阈值
        但是这里应该先判断是否还有空闲的socket,有则可以复用,
        没有则创建新的socket 
      */  
      if (this.requests[name] && this.requests[name].length) {  
        const req = this.requests[name][0];  
        const socketCreationHandler = handleSocketCreation(this, 
                                                                req,            
                                                                false);  
        this.createSocket(req, options, socketCreationHandler);  
      }  
    };  

前面已经分析过,Agent维护了两个socket队列,删除socket就是从这两个队列中找到对应的socket,然后移除它。移除后需要判断一下是否还有等待socket的请求队列,有的话就新建一个socket去处理它。因为移除了一个socket,就说明可以新增一个socket。

18.4.4 设置socket keepalive

当socket被使用完并且被插入空闲队列后,需要重新设置socket的keepalive值。等到超时会自动关闭socket。在一个socket上调用一次setKeepAlive就可以了,这里可能会导致多次调用setKeepAlive,不过也没有影响。

    function keepSocketAlive(socket) {  
      socket.setKeepAlive(true, this.keepAliveMsecs);  
      socket.unref();  
      return true;  
    };  

另外需要设置ref标记,防止该socket阻止事件循环的退出,因为该socket是空闲的,不应该影响事件循环的退出。

18.4.5 复用socket

    function reuseSocket(socket, req) {  
      req.reusedSocket = true;  
      socket.ref();  
    };  

重新使用该socket,需要修改ref标记,阻止事件循环退出,并标记请求使用的是复用socket。

18.4.6 销毁Agent

    function destroy() {  
      for (const set of [this.freeSockets, this.sockets]) {  
        for (const key of ObjectKeys(set)) {  
          for (const setName of set[key]) {  
            setName.destroy();  
          }  
        }  
      }  
    };  

因为Agent本质上是一个socket池,销毁Agent即销毁池里维护的所有socket。

18.4.7 使用连接池

我们看一下如何使用Agent。

    function addRequest(req, options, port, localAddress) {  
      // 参数处理  
      if (typeof options === 'string') {  
        options = {  
          host: options,  
          port,  
          localAddress  
        };  
      }  

      options = { ...options, ...this.options };  
      if (options.socketPath)  
        options.path = options.socketPath;  

      if (!options.servername && options.servername !== '')  
        options.servername = calculateServerName(options, req);  
      // 拿到请求对应的key  
      const name = this.getName(options);  
      // 该key还没有在使用的socekt则初始化数据结构  
      if (!this.sockets[name]) {  
        this.sockets[name] = [];  
      }  
      // 该key对应的空闲socket列表  
      const freeLen = this.freeSockets[name] ? 
                        this.freeSockets[name].length : 0;  
      // 该key对应的所有socket个数  
      const sockLen = freeLen + this.sockets[name].length;  
      // 该key有对应的空闲socekt  
      if (freeLen) {    
        // 获取一个该key对应的空闲socket  
        const socket = this.freeSockets[name].shift();  
        // 取完了删除,防止内存泄漏  
        if (!this.freeSockets[name].length)  
          delete this.freeSockets[name];  
        // 设置ref标记,因为正在使用该socket  
        this.reuseSocket(socket, req);  
        // 设置请求对应的socket  
        setRequestSocket(this, req, socket);  
        // 插入正在使用的socket队列  
        this.sockets[name].push(socket);  
      } else if (sockLen < this.maxSockets) {   
        /* 
          如果该key没有对应的空闲socket并且使用的 
          socket个数还没有得到阈值,则继续创建 
        */  
        this.createSocket(req,
                            options, 
                            handleSocketCreation(this, req, true));  
      } else {  
        // 等待该key下有空闲的socket  
        if (!this.requests[name]) {  
          this.requests[name] = [];  
        }  
        this.requests[name].push(req);  
      }  
    }  

当我们需要发送一个HTTP请求的时候,我们可以通过Agent的addRequest方法把请求托管到Agent中,当有可用的socket时,Agent会通知我们。addRequest的代码很长,主要分为三种情况。 1 有空闲socket,则直接复用,并插入正在使用的socket队列中
我们主要看一下setRequestSocket函数

    function setRequestSocket(agent, req, socket) {  
      // 通知请求socket创建成功  
      req.onSocket(socket);  
      const agentTimeout = agent.options.timeout || 0;  
      if (req.timeout === undefined || req.timeout === agentTimeout) 
      {  
        return;  
      }  
      // 开启一个定时器,过期后触发timeout事件  
      socket.setTimeout(req.timeout);  
      /*
        监听响应事件,响应结束后需要重新设置超时时间,
        开启下一个请求的超时计算,否则会提前过期 
      */ 
      req.once('response', (res) => {  
        res.once('end', () => {  
          if (socket.timeout !== agentTimeout) {  
            socket.setTimeout(agentTimeout);  
          }  
        });  
      });  
    }  

setRequestSocket函数通过req.onSocket(socket)通知调用方有可用socket。然后如果请求设置了超时时间则设置socket的超时时间,即请求的超时时间。最后监听响应结束事件,重新设置超时时间。 2 没有空闲socket,但是使用的socket个数还没有达到阈值,则创建新的socket。
我们主要分析创建socket后的回调handleSocketCreation。

    function handleSocketCreation(agent, request, informRequest) {  
      return function handleSocketCreation_Inner(err, socket) {  
        if (err) {  
          process.nextTick(emitErrorNT, request, err);  
          return;  
        }  
        /* 
         是否需要直接通知请求方,这时候request不是来自等待
          socket的requests队列, 而是来自调用方,见addRequest 
        */  
        if (informRequest)  
          setRequestSocket(agent, request, socket);  
        else  
          /*
            不直接通知,先告诉agent有空闲的socket,
            agent会判断是否有正在等待socket的请求,有则处理  
           */
          socket.emit('free');  
      };  
    }  

3 不满足1,2,则把请求插入等待socket队列。
插入等待socket队列后,当有socket空闲时会触发free事件,我们看一下该事件的处理逻辑。

    // 监听socket空闲事件  
     this.on('free', (socket, options) => {  
       const name = this.getName(options);
       // socket还可写并且还有等待socket的请求,则复用socket  
       if (socket.writable &&  
           this.requests[name] && this.requests[name].length) {  
         // 拿到一个等待socket的请求,然后通知它有socket可用  
         const req = this.requests[name].shift();  
         setRequestSocket(this, req, socket);  
         // 没有等待socket的请求则删除,防止内存泄漏  
         if (this.requests[name].length === 0) {  
           // don't leak  
           delete this.requests[name];  
         }  
       } else {  
         // socket不可用写或者没有等待socket的请求了  
         const req = socket._httpMessage;  
         // socket可写并且请求设置了允许使用复用的socket  
         if (req &&  
             req.shouldKeepAlive &&  
             socket.writable &&  
             this.keepAlive) {  
           let freeSockets = this.freeSockets[name];  
           // 该key下当前的空闲socket个数  
           const freeLen = freeSockets ? freeSockets.length : 0;  
           let count = freeLen;  
           // 正在使用的socket个数  
           if (this.sockets[name])  
             count += this.sockets[name].length;  
           /*
               该key使用的socket个数达到阈值或者空闲socket达到阈值,
               则不复用socket,直接销毁socket  
            */
           if (count > this.maxSockets || 
             freeLen >= this.maxFreeSockets) {  
             socket.destroy();  
           } else if (this.keepSocketAlive(socket)) {   
             /*
                重新设置socket的存活时间,设置失败说明无法重新设置存活时
                间,则说明可能不支持复用  
              */
             freeSockets = freeSockets || [];  
             this.freeSockets[name] = freeSockets;  
             socket[async_id_symbol] = -1;  
             socket._httpMessage = null;  
             // 把socket从正在使用队列中移除  
             this.removeSocket(socket, options);  
             // 插入socket空闲队列  
             freeSockets.push(socket);  
           } else {  
             // 不复用则直接销毁  
             socket.destroy();  
           }  
         } else {  
           socket.destroy();  
         }  
       }  
     });  

当有socket空闲时,分为以下几种情况
1 如果有等待socket的请求,则直接复用socket。
2 如果没有等待socket的请求,允许复用并且socket个数没有达到阈值则插入空闲队列。
3 直接销毁

18.4.8 测试例子

客户端

    const http = require('http');  
    const keepAliveAgent = new http.Agent({ keepAlive: true, maxSockets: 1 });  
    const options = {port: 10000, method: 'GET',  host: '127.0.0.1',}  
    options.agent = keepAliveAgent;  
    http.get(options, () => {});  
    http.get(options, () => {});  
    console.log(options.agent.requests)  

服务器

    let i =0;  
    const net = require('net');  
    net.createServer((socket) => {  
      console.log(++i);  
    }).listen(10000);  

在例子中,首先创建了一个tcp服务器。然后在客户端使用agent。但是maxSocket的值为1,代表最多只能有一个socket,而这时候客户端发送两个请求,所以有一个请求就会在排队。服务器也只收到了一个连接。