下面是nodejs創建一個服務器的代碼。接下來我們一起分析這個過程。
var http = require('http');http.createServer(function (request, response) { response.end('Hello World');}).listen(9297);首先我們去到lib/http.js模塊看一下這個函數的代碼。
function createServer(requestListener) { return new Server(requestListener);}只是對_http_server.js做了些封裝。我們繼續往下看。
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);發現_http_server.js也沒有太多邏輯，繼續看lib/net.js下的代碼。
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;}至此http.createServer就執行結束了，我們發現這個過程還沒有涉及到很多邏輯，并且還是停留到js層面。接下來我們繼續分析listen函數的過程。該函數是net模塊提供的。我們只看關鍵的代碼。
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);}到這我們終于看到了tcp連接的內容，每一個服務器新建一個handle并且保存他，該handle是一個TCP對象。然后執行bind和listen函數。接下來我們就看一下TCP類的代碼。TCP是C++提供的類。對應的文件是tcp_wrap.cc。我們看看new TCP的時候發生了什么。
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); }我們看到，new TCP的時候其實是執行libuv的uv_tcp_init函數，初始化一個uv_tcp_t的結構體。首先我們先看一下uv_tcp_t結構體的結構。
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;}我們接著看uv__stream_init做了什么事情。
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) */} 從代碼可以知道，只是對uv_tcp_t結構體做了一些初始化操作。到這，new TCP的邏輯就執行完畢了。接下來就是繼續分類nodejs里調用bind和listen的邏輯。nodejs的bind對應libuv的函數是uv__tcp_bind，listen對應的是uv_tcp_listen。
先看一個bind的核心代碼。
/* 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;}上面的一系列操作主要是新建一個socket文件描述符，設置一些flag，然后把文件描述符保存到IO觀察者中，libuv在poll IO階段會監聽該文件描述符，如果有事件到來，會執行設置的回調函數，該函數是在uv__stream_init里設置的uv__stream_io。最后執行bind函數進行綁定操作。最后我們來分析一下listen函數。首先看下tcp_wrapper.cc的代碼。
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);}代碼中有個很重要的地方就是OnConnection函數，nodejs給listen函數設置了一個回調函數OnConnection，該函數在IO觀察者里保存的文件描述符有連接到來時會被調用。OnConnection函數是在connection_wrap.cc定義的，tcp_wrapper繼承了connection_wrap。下面我們先看一下uv_listen。該函數調用了uv_tcp_listen。該函數的核心代碼如下。
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);
在libuv的poll IO階段，epoll_wait會監聽到到來的連接，然后調用uv__server_io。下面是該函數的核心代碼。
// 繼續注冊事件，等待連接 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);
libuv會摘下一個連接，得到對應的socket。然后執行nodejs層的回調，這時候我們來看一下OnConnection的代碼。
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);OnConnection新建了一個uv_tcp_t結構體。代表這個連接。然后調用uv_accept。
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); ...}最后執行nodejs的回調。
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);}listen函數總體的邏輯就是把socket設置為可監聽，然后注冊事件，等待連接的到來，連接到來的時候，調用accept獲取新建立的連接，tcp_wrapper.cc的回調新建一個uv_tcp_t結構體，代表新的連接，然后設置可讀寫事件，并且設置回調為uv__stream_io，等待數據的到來。最后執行_http_server.js設置的回調connectionListener。至此，服務器啟動并且接收連接的過程就完成了。接下來就是對用戶數據的讀寫。當用戶傳來數據時，處理數據的函數是uv__stream_io。后面繼續解析數據的讀寫。
總結
以上就是這篇文章的全部內容了，希望本文的內容對大家的學習或者工作具有一定的參考學習價值，謝謝大家對VeVb武林網的支持。
新聞熱點
疑難解答
