18-HTTP

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;  
    };