node.js的http.createServer过程深入解析

var http = require('http');http.createServer(function (request, response) {  response.end('Hello World');}).listen(9297);

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

function Server(requestListener) { if (!(this instanceof Server)) return new Server(requestListener); net.Server.call(this, { allowHalfOpen: true }); // 收到http请求时执行的回调 if (requestListener) {  this.on('request', requestListener); } this.httpAllowHalfOpen = false; // 建立tcp连接的回调 this.on('connection', connectionListener); this.timeout = 2 * 60 * 1000; this.keepAliveTimeout = 5000; this._pendingResponseData = 0; this.maxHeadersCount = null;}util.inherits(Server, net.Server);

function Server(options, connectionListener) { if (!(this instanceof Server))  return new Server(options, connectionListener); EventEmitter.call(this); // connectionListener在http.js处理过了 if (typeof options === 'function') {  connectionListener = options;  options = {};  this.on('connection', connectionListener); } else if (options == null || typeof options === 'object') {  options = options || {};  if (typeof connectionListener === 'function') {   this.on('connection', connectionListener);  } } else {  throw new errors.TypeError('ERR_INVALID_ARG_TYPE',                'options',                'Object',                options); } this._connections = 0; ...... this[async_id_symbol] = -1; this._handle = null; this._usingWorkers = false; this._workers = []; this._unref = false; this.allowHalfOpen = options.allowHalfOpen || false; this.pauseOnConnect = !!options.pauseOnConnect;}

Server.prototype.listen = function(...args) { // 处理入参，根据文档我们知道listen可以接收好几个参数，我们这里是只传了端口号9297 var normalized = normalizeArgs(args); // normalized = [{port: 9297}, null]; var options = normalized[0]; var cb = normalized[1]; // 第一次listen的时候会创建，如果非空说明已经listen过 if (this._handle) {  throw new errors.Error('ERR_SERVER_ALREADY_LISTEN'); } ...... listenInCluster(this, null, options.port | 0, 4,           backlog, undefined, options.exclusive);}function listenInCluster() {  ...  server._listen2(address, port, addressType, backlog, fd);}_listen2 = setupListenHandle = function() {  ......  this._handle = createServerHandle(...);  this._handle.listen(backlog || 511);}function createServerHandle() {  handle = new TCP(TCPConstants.SERVER);  handle.bind(address, port);}

void TCPWrap::New(const FunctionCallbackInfo<Value>& args) { // This constructor should not be exposed to public javascript. // Therefore we assert that we are not trying to call this as a // normal function. CHECK(args.IsConstructCall()); CHECK(args[0]->IsInt32()); Environment* env = Environment::GetCurrent(args); int type_value = args[0].As<Int32>()->Value(); TCPWrap::SocketType type = static_cast<TCPWrap::SocketType>(type_value); ProviderType provider; switch (type) {  case SOCKET:   provider = PROVIDER_TCPWRAP;   break;  case SERVER:   provider = PROVIDER_TCPSERVERWRAP;   break;  default:   UNREACHABLE(); } new TCPWrap(env, args.This(), provider);}TCPWrap::TCPWrap(Environment* env, Local<Object> object, ProviderType provider)  : ConnectionWrap(env, object, provider) { int r = uv_tcp_init(env->event_loop(), &handle_); CHECK_EQ(r, 0); }

uv_tcp_tuv_tcp_t// 初始化一个tcp流的结构体int uv_tcp_init(uv_loop_t* loop, uv_tcp_t* tcp) { // 未指定未指定协议 return uv_tcp_init_ex(loop, tcp, AF_UNSPEC);}int uv_tcp_init_ex(uv_loop_t* loop, uv_tcp_t* tcp, unsigned int flags) { int domain; /* Use the lower 8 bits for the domain */ // 低八位是domain domain = flags & 0xFF; if (domain != AF_INET && domain != AF_INET6 && domain != AF_UNSPEC)  return UV_EINVAL; // 除了第八位的其他位是flags if (flags & ~0xFF)  return UV_EINVAL; uv__stream_init(loop, (uv_stream_t*)tcp, UV_TCP); /* If anything fails beyond this point we need to remove the handle from  * the handle queue, since it was added by uv__handle_init in uv_stream_init.  */ if (domain != AF_UNSPEC) {  int err = maybe_new_socket(tcp, domain, 0);  if (err) {   // 出错则把该handle移除loop队列   QUEUE_REMOVE(&tcp->handle_queue);   return err;  } } return 0;}

void uv__stream_init(uv_loop_t* loop,           uv_stream_t* stream,           uv_handle_type type) { int err; uv__handle_init(loop, (uv_handle_t*)stream, type); stream->read_cb = NULL; stream->alloc_cb = NULL; stream->close_cb = NULL; stream->connection_cb = NULL; stream->connect_req = NULL; stream->shutdown_req = NULL; stream->accepted_fd = -1; stream->queued_fds = NULL; stream->delayed_error = 0; QUEUE_INIT(&stream->write_queue); QUEUE_INIT(&stream->write_completed_queue); stream->write_queue_size = 0; if (loop->emfile_fd == -1) {  err = uv__open_cloexec("/dev/null", O_RDONLY);  if (err < 0)    /* In the rare case that "/dev/null" isn't mounted open "/"     * instead.     */    err = uv__open_cloexec("/", O_RDONLY);  if (err >= 0)   loop->emfile_fd = err; }#if defined(__APPLE__) stream->select = NULL;#endif /* defined(__APPLE_) */ // 初始化io观察者 uv__io_init(&stream->io_watcher, uv__stream_io, -1);}void uv__io_init(uv__io_t* w, uv__io_cb cb, int fd) { assert(cb != NULL); assert(fd >= -1); // 初始化队列，回调，需要监听的fd QUEUE_INIT(&w->pending_queue); QUEUE_INIT(&w->watcher_queue); w->cb = cb; w->fd = fd; w->events = 0; w->pevents = 0;#if defined(UV_HAVE_KQUEUE) w->rcount = 0; w->wcount = 0;#endif /* defined(UV_HAVE_KQUEUE) */}

/* Cannot set IPv6-only mode on non-IPv6 socket. */ if ((flags & UV_TCP_IPV6ONLY) && addr->sa_family != AF_INET6)  return UV_EINVAL; // 获取一个socket并且设置某些标记 err = maybe_new_socket(tcp, addr->sa_family, 0); if (err)  return err; on = 1; // 设置在端口可重用 if (setsockopt(tcp->io_watcher.fd, SOL_SOCKET, SO_REUSEADDR, &on, sizeof(on)))  return UV__ERR(errno); bind(tcp->io_watcher.fd, addr, addrlen) && errno != EADDRINUSEstatic int maybe_new_socket(uv_tcp_t* handle, int domain, unsigned long flags) { struct sockaddr_storage saddr; socklen_t slen; if (domain == AF_UNSPEC) {  handle->flags |= flags;  return 0; } return new_socket(handle, domain, flags);}static int new_socket(uv_tcp_t* handle, int domain, unsigned long flags) { struct sockaddr_storage saddr; socklen_t slen; int sockfd; int err; // 获取一个socket err = uv__socket(domain, SOCK_STREAM, 0); if (err < 0)  return err; sockfd = err; // 设置选项和保存socket的文件描述符到io观察者中 err = uv__stream_open((uv_stream_t*) handle, sockfd, flags); if (err) {  uv__close(sockfd);  return err; } ... return 0;}int uv__stream_open(uv_stream_t* stream, int fd, int flags) { if (!(stream->io_watcher.fd == -1 || stream->io_watcher.fd == fd))  return UV_EBUSY; assert(fd >= 0); stream->flags |= flags; if (stream->type == UV_TCP) {  if ((stream->flags & UV_HANDLE_TCP_NODELAY) && uv__tcp_nodelay(fd, 1))   return UV__ERR(errno);  /* TODO Use delay the user passed in. */  if ((stream->flags & UV_HANDLE_TCP_KEEPALIVE) &&    uv__tcp_keepalive(fd, 1, 60)) {   return UV__ERR(errno);  } } ... // 保存socket对应的文件描述符到io观察者中，libuv会在io poll阶段监听该文件描述符 stream->io_watcher.fd = fd; return 0;}

void TCPWrap::Listen(const FunctionCallbackInfo<Value>& args) { TCPWrap* wrap; ASSIGN_OR_RETURN_UNWRAP(&wrap,             args.Holder(),             args.GetReturnValue().Set(UV_EBADF)); int backlog = args[0]->Int32Value(); int err = uv_listen(reinterpret_cast<uv_stream_t*>(&wrap->handle_),           backlog,           OnConnection); args.GetReturnValue().Set(err);}

 if (listen(tcp->io_watcher.fd, backlog))  return UV__ERR(errno); // cb即OnConnection tcp->connection_cb = cb; tcp->flags |= UV_HANDLE_BOUND; // 有连接到来时的libuv层回调，覆盖了uv_stream_init时设置的值 tcp->io_watcher.cb = uv__server_io; // 注册事件 uv__io_start(tcp->loop, &tcp->io_watcher, POLLIN);

// 继续注册事件，等待连接 uv__io_start(stream->loop, &stream->io_watcher, POLLIN); err = uv__accept(uv__stream_fd(stream)); // 保存连接对应的socket stream->accepted_fd = err; // 执行nodejs层回调 stream->connection_cb(stream, 0);

OnConnection(uv_stream_t* handle,int status)  if (status == 0) {    // 新建一个uv_tcp_t结构体    Local<Object> client_obj = WrapType::Instantiate(env, wrap_data, WrapType::SOCKET);    WrapType* wrap;    ASSIGN_OR_RETURN_UNWRAP(&wrap, client_obj);    uv_stream_t* client_handle = reinterpret_cast<uv_stream_t*>(&wrap->handle_);    // uv_accept返回0表示成功    if (uv_accept(handle, client_handle))     return;    argv[1] = client_obj; } // 执行上层的回调，该回调是net.js设置的onconnection wrap_data->MakeCallback(env->onconnection_string(), arraysize(argv), argv);

int uv_accept(uv_stream_t* server, uv_stream_t* client) {  ...  // 新建的uv_tcp_t结构体关联accept_fd，注册读写事件  uv__stream_open(client, server->accepted_fd, UV_HANDLE_READABLE | UV_HANDLE_WRITABLE);  ...}

function onconnection(err, clientHandle) { var handle = this; var self = handle.owner; if (err) {  self.emit('error', errnoException(err, 'accept'));  return; } if (self.maxConnections && self._connections >= self.maxConnections) {  clientHandle.close();  return; } var socket = new Socket({  handle: clientHandle,  allowHalfOpen: self.allowHalfOpen,  pauseOnCreate: self.pauseOnConnect }); socket.readable = socket.writable = true; self._connections++; socket.server = self; socket._server = self; DTRACE_NET_SERVER_CONNECTION(socket); LTTNG_NET_SERVER_CONNECTION(socket); COUNTER_NET_SERVER_CONNECTION(socket); // 触发_http_server.js里设置的connectionListener回调 self.emit('connection', socket);}