本文为来自 字节跳动-国际化电商-S 项目团队 成员的文章，已授权 ELab 发布。
一个 TCP 连接的案例TCP 服务端
const.NET = require('net');
const server = new net.Server();
server.listen(9999, '127.0.0.1', () => {
console.log(`server is listening on ${server.address().address}:${server.address().port}`);
});
server.on('connection', (socket) => {
server.getConnections((err, connections) => {
console.log('current clients is: ', connections);
});
socket.on('data', (data) => {
console.log(`received data: ${data.toString()}`);
});
});
const net = require('net');
const client = new net.Socket();
client.connect({
port: 9999,
address: '127.0.0.1',
});
client.on('connect', () => {
console.log('connect success');
client.write(`Hello Server!, I'm ${Math.round(Math.random() * 100)}`);
});
疑问❓
NodeJS 代码是如何跑起来的
TCP 连接在 NodeJS 中是如何保持一直监听而进程不中断的
NodeJS 是如何处理并发连接的，当遇到阻塞型调用时如何不阻塞主线程的
NodeJS 源码分为三层：JS、C++ 以及 C。
JS 层提供面向用户的调用底层能力的接口，即各种 NodeJS 原生模块，如 net、http、fs、DNS 以及 path 等
C++ 层主要通过 V8 为 JS 层提供与底层交互的能力，起到类似桥梁的作用，通过 V8 不仅实现 JS 的解释执行，还扩展的 JS 的能力边界
C 层主要包括 Libuv 这一跨平台的异步 IO 库以及其他第三方 C 库
image.png
RegisterBuiltinModules 函数的作用是注册一系列 C++ 模块，通过宏定义展开，最终变成如下逻辑：
void RegisterBuiltinModules() {
_register_async_wrap();
_register_buffer();
_register_fs();
_register_url();
// ...
}
通过注册函数，将各个 C++ 模块维护在 modlist_internal 这一链表中，后续在原生 JS 模块中调用 C++ 模块时就可以根据模块名找到对应的模块。
Environment 在 NodeJS 中是一个运行时的环境对象，很多全局变量都托管在该类上，创建完 environment 后，就将其和 Context 进行绑定，后续 V8 可通过 context 获取 env 对象。
下面简单介绍一下 V8 的 isolate 、 context、scope、handle 等对象。
isolate 是一个独立隔离实例的环境，同一时刻只能被一个线程进入；
context 可以理解为执行上下文对象，可以导入不同的环境变量和函数；
Scope 指的是作用域，可看成是句柄的容器，一个作用域里面可以有很多个句柄；
HandleScope 是用来管理 Handle 的，而 Context::Scope 仅仅用来管理 Context 对象。
Handle 是 V8 引用对象的技术手段，Handle 分为 Local 和 Persistent 两种。Local 是局部的，它同时被 HandleScope 进行管理。 persistent，类似于全局的，不受 HandleScope 的管理，其作用域可以延伸到不同的函数。
RunBootstrApping 主要调用了 BootstrapInternalLoaders 和 BootstrapNode 函数。
BootstrapInternalLoaders 用于编译执行 /lib/internal/bootstrap/loader.js，它的具体逻辑是为了NodeJS 能在JS层 通过 binding 函数加载C++模块，以便在原生 JS 模块中调用 C++ 模块。
BootstrapNode 用于初始化执行上下文，暴露 global 对象在全局上下文中，编译执行 /lib/internal/bootstrap/node，从而设置一些全局变量或方法到 global 或者 process
// lib/internal/bootstrap/node.js
// proces 挂载一系列属性方法
{
process.dlopen = rawMethods.dlopen;
process.uptime = rawMethods.uptime;
// TODO(joyeecheung): either remove them or make them public
process._getActiveRequests = rawMethods._getActiveRequests;
process._getActiveHandles = rawMethods._getActiveHandles;
// TODO(joyeecheung): remove these
process.reallyExit = rawMethods.reallyExit;
process._kill = rawMethods._kill;
const wrapped = perThreadSetup.wrapProcessMethods(rawMethods);
process._rawDebug = wrapped._rawDebug;
process.hrtime = wrapped.hrtime;
process.hrtime.bigint = wrapped.hrtimeBigInt;
process.cpuUsage = wrapped.cpuUsage;
process.resourceUsage = wrapped.resourceUsage;
process.memoryUsage = wrapped.memoryUsage;
process.kill = wrapped.kill;
process.exit = wrapped.exit;
process.openStdin = function() {
process.stdin.resume();
return process.stdin;
};
}
// global 挂载一系列属性和方法
if (!config.noBrowserGlobals) {
// Override global console from the one provided by the VM
// to the one implemented by Node.js
// https://console.spec.whatwg.org/#console-namespace
exposeNamespace(global, 'console', createGlobalConsole(global.console));
const { URL, URLSearchParams } = require('internal/url');
// https://url.spec.whatwg.org/#url
exposeInterface(global, 'URL', URL);
// https://url.spec.whatwg.org/#urlsearchparams
exposeInterface(global, 'URLSearchParams', URLSearchParams);
const {
TextEncoder, TextDecoder
} = require('internal/encoding');
// https://encoding.spec.whatwg.org/#textencoder
exposeInterface(global, 'TextEncoder', TextEncoder);
// https://encoding.spec.whatwg.org/#textdecoder
exposeInterface(global, 'TextDecoder', TextDecoder);
// https://html.spec.whatwg.org/multipage/webappapis.html#windoworworkerglobalscope
const timers = require('timers');
defineOperation(global, 'clearInterval', timers.clearInterval);
defineOperation(global, 'clearTimeout', timers.clearTimeout);
defineOperation(global, 'setInterval', timers.setInterval);
defineOperation(global, 'setTimeout', timers.setTimeout);
defineOperation(global, 'queueMicrotask', queueMicrotask);
// Non-standard extensions:
defineOperation(global, 'clearImmediate', timers.clearImmediate);
defineOperation(global, 'setImmediate', timers.setImmediate);
}
// ...
这里对事件循环的部分阶段做一些初始化的操作，创建一个默认的 event_loop 结构体用于管理后续各个阶段产生的任务
void Environment::InitializeLibuv(bool start_profiler_idle_notifier) {
HandleScope handle_scope(isolate());
Context::Scope context_scope(context());
CHECK_EQ(0, uv_timer_init(event_loop(), timer_handle()));
uv_unref(reinterpret_cast<uv_handle_t*>(timer_handle()));
uv_check_init(event_loop(), immediate_check_handle());
uv_unref(reinterpret_cast<uv_handle_t*>(immediate_check_handle()));
uv_idle_init(event_loop(), immediate_idle_handle());
uv_check_start(immediate_check_handle(), CheckImmediate);
uv_prepare_init(event_loop(), &idle_prepare_handle_);
uv_check_init(event_loop(), &idle_check_handle_);
uv_async_init(
event_loop(),
&task_queues_async_,
[](uv_async_t* async) {
Environment* env = ContAInerOf(
&Environment::task_queues_async_, async);
env->CleanupFinalizationGroups();
env->RunAndClearNativeImmediates();
});
uv_unref(reinterpret_cast<uv_handle_t*>(&idle_prepare_handle_));
uv_unref(reinterpret_cast<uv_handle_t*>(&idle_check_handle_));
uv_unref(reinterpret_cast<uv_handle_t*>(&task_queues_async_));
// ...
}
StartExecution 用于加载用户 JS 代码并执行
// src/node.cc
MaybeLocal<Value> StartExecution(Environment* env, StartExecutionCallback cb) {
// ...
if (!first_argv.empty() && first_argv != "-") {
return StartExecution(env, "internal/main/run_main_module");
}
// ...
}
// lib/internal/main/run_main_module.js
require('internal/modules/cjs/loader').Module.runMain(process.argv[1]);
执行完用户 JS 代码，用户代码就会往 Libuv 中注册一些任务，然后进入整个事件循环，直到没有待处理的任务，Libuv 则会退出事件循环，进而退出 NodeJS 进程。
// src/node_main_instance.cc
do {
uv_run(env->event_loop(), UV_RUN_DEFAULT);
per_process::v8_platform.DrainVMTasks(isolate_);
more = uv_loop_alive(env->event_loop());
if (more && !env->is_stopping()) continue;
if (!uv_loop_alive(env->event_loop())) {
EmitBeforeExit(env.get());
}
// Emit `beforeExit` if the loop became alive either after emitting
// event, or after running some callbacks.
more = uv_loop_alive(env->event_loop());
} while (more == true && !env->is_stopping());
// src/node_main.cc
int main(int argc, char* argv[]) {
return node::Start(argc, argv);
}
// src/node.cc
namespace node {
int Start(int argc, char** argv) {
InitializationResult result = InitializeOncePerProcess(argc, argv);
// ...
NodeMainInstance main_instance(¶ms,
uv_default_loop(),
per_process::v8_platform.Platform(),
result.args,
result.exec_args,
indexes);
result.exit_code = main_instance.Run();
}
InitializationResult InitializeOncePerProcess(int argc, char** argv) {
// ...
{
result.exit_code =
InitializeNodeWithArgs(&(result.args), &(result.exec_args), &errors);
//...
}
V8::Initialize();
return result;
}
int InitializeNodeWithArgs(std::vector<std::string>* argv,
std::vector<std::string>* exec_argv,
std::vector<std::string>* errors) {
// ...
// Register built-in modules
binding::RegisterBuiltinModules();
// ...
}
MaybeLocal<Value> Environment::RunBootstrapping() {
EscapableHandleScope scope(isolate_);
//...
if (BootstrapInternalLoaders().IsEmpty()) {
return MaybeLocal<Value>();
}
Local<Value> result;
if (!BootstrapNode().ToLocal(&result)) {
return MaybeLocal<Value>();
}
//...
return scope.Escape(result);
}
}
// src/node_main_instance.cc
namespace node {
int NodeMainInstance::Run() {
// ...
DeleteFnPtr<Environment, FreeEnvironment> env =
CreateMainEnvironment(&exit_code);
if (exit_code == 0) {
LoadEnvironment(env.get());
// ...
{
// ...
do {
uv_run(env->event_loop(), UV_RUN_DEFAULT);
per_process::v8_platform.DrainVMTasks(isolate_);
more = uv_loop_alive(env->event_loop());
if (more && !env->is_stopping()) continue;
if (!uv_loop_alive(env->event_loop())) {
EmitBeforeExit(env.get());
}
// Emit `beforeExit` if the loop became alive either after emitting
// event, or after running some callbacks.
more = uv_loop_alive(env->event_loop());
} while (more == true && !env->is_stopping());
}
}
// ...
}
NodeMainInstance::CreateMainEnvironment(int* exit_code) {
// ...
context = NewContext(isolate_);
Context::Scope context_scope(context);
DeleteFnPtr<Environment, FreeEnvironment> env { CreateEnvironment(
isolate_data_.get(),
context,
args_,
exec_args_,
EnvironmentFlags::kDefaultFlags) };
return env;
}
}
// src/environment.cc
namespace node {
void LoadEnvironment(Environment* env) {
USE(LoadEnvironment(env,
StartExecutionCallback{},
{}));
}
MaybeLocal<Value> LoadEnvironment(
Environment* env,
StartExecutionCallback cb,
std::unique_ptr<InspectorParentHandle> removeme) {
env->InitializeLibuv(per_process::v8_is_profiling);
env->InitializeDiagnostics();
return StartExecution(env, cb);
}
Environment* CreateEnvironment(IsolateData* isolate_data,
Local<Context> context,
int argc,
const char* const* argv,
int exec_argc,
const char* const* exec_argv) {
return CreateEnvironment(
isolate_data, context,
std::vector<std::string>(argv, argv + argc),
std::vector<std::string>(exec_argv, exec_argv + exec_argc));
}
Environment* CreateEnvironment(
IsolateData* isolate_data,
Local<Context> context,
const std::vector<std::string>& args,
const std::vector<std::string>& exec_args,
EnvironmentFlags::Flags flags,
ThreadId thread_id,
std::unique_ptr<InspectorParentHandle> inspector_parent_handle) {
Isolate* isolate = context->GetIsolate();
HandleScope handle_scope(isolate);
Context::Scope context_scope(context);
// TODO(addaleax): This is a much better place for parsing per-Environment
// options than the global parse call.
Environment* env = new Environment(
isolate_data,
context,
args,
exec_args,
flags,
thread_id);
// ...
if (env->RunBootstrapping().IsEmpty()) {
FreeEnvironment(env);
return nullptr;
}
return env;
}
}
Libuv 是 NodeJS 的核心组件，是一个跨平台的处理异步 I/O 请求的 C 库，从架构来看，它把各类请求主要分为两大类：网络 I/O 相关请求，以及文件 I/O、DNS Ops 以及 User code 组成的请求。
对于网络 I/O 相关请求，根据 OS 平台的不同，分别采用了 linux 的 epoll、OSX 和 BSD 类 OS 的 kqueue、SunOS 的 event ports 以及 windows 的 IOCP 等 I/O 读写机制。
对于 File I/O 为代表的请求，则使用线程池实现异步请求处理，具有更好的跨平台特性。
在 Node 应用启动后，就会进入 Libuv 事件循环中，每一轮循环 Libuv 都会处理维护在各个阶段的任务队列的回调节点，在回调节点中可能会产生新的任务，任务可能在当前循环或是下个循环继续被处理。
以下是 Libuv 的执行流程图：
下面简述一下各个阶段代表的含义：
// src/unix/core.c
int uv_run(uv_loop_t* loop, uv_run_mode mode) {
int timeout;
int r;
int can_sleep;
r = uv__loop_alive(loop);
if (!r)
uv__update_time(loop);
while (r != 0 && loop->stop_flag == 0) {
uv__update_time(loop);
uv__run_timers(loop);
can_sleep =
QUEUE_EMPTY(&loop->pending_queue) && QUEUE_EMPTY(&loop->idle_handles);
uv__run_pending(loop);
uv__run_idle(loop);
uv__run_prepare(loop);
timeout = 0;
if ((mode == UV_RUN_ONCE && can_sleep) || mode == UV_RUN_DEFAULT)
timeout = uv__backend_timeout(loop);
uv__metrics_inc_loop_count(loop);
// 动态设置 epoll_wait 的超时时间
uv__io_poll(loop, timeout);
for (r = 0; r < 8 && !QUEUE_EMPTY(&loop->pending_queue); r++)
uv__run_pending(loop);
uv__metrics_update_idle_time(loop);
uv__run_check(loop);
uv__run_closing_handles(loop);
if (mode == UV_RUN_ONCE) {
uv__update_time(loop);
uv__run_timers(loop);
}
r = uv__loop_alive(loop);
if (mode == UV_RUN_ONCE || mode == UV_RUN_NOWAIT)
break;
}
/* The if statement lets gcc compile it to a conditional store. Avoids
* dirtying a cache line.
*/
if (loop->stop_flag != 0)
loop->stop_flag = 0;
return r;
}
了解了 Libuv 的事件循环流程，接下来结合 JS 代码具体看看 NodeJS 是如何进行任务调度的。
image.png
目前，主要有五种主要类型的队列被 Libuv 的事件循环所处理：
除了以上五种主要的任务列表，还有额外两种不属于 libuv 而是作为 NodeJS 一部分的任务队列：
nextTicks 队列和其他微任务队列会在事件循环每一阶段穿插调用，nextTicks 优先级会比其他微任务队列更高。
// timer -> pending -> idle -> prepare -> poll io -> check -> close
// timer phase
setTimeout(() => {
Promise.resolve().then(() => {
console.log('promise resolve in timeout');
process.nextTick(() => {
console.log('tick task in timeout promise');
});
})
process.nextTick(() => {
console.log('tick task in timeout');
process.nextTick(() => {
console.log("tick task in timeout->tick");
})
});
console.log('timer task');
}, 0);
// check phase
setImmediate(() => {
process.nextTick(() => {
console.log('imeediate->tick task')
});
console.log('immediate task');
});
Promise.resolve().then(() => {
console.log('promise resolve');
});
process.nextTick(() => {
console.log("tick task");
});
console.log('run main thread');
// result
run main thread
tick task
promise resolve
timer task
tick task in timeout
tick task in timeout->tick
promise resolve in timeout
tick task in timeout promise
immediate task
imeediate->tick task
现在解读一下以上的执行流程：
NodeJS 在经过一系列初始化工作后，开始执行用户 JS 代码，解释执行过程中，分别把 setTimeout、setImmediate、Promise、nextTick 函数的回调插入 timer、immediate、microtask 和 nexttick 队列。
nextTicks = [() => {
console.log("tick task");
}];
microtasks = [() => {
console.log('promise resolve');
}];
分别打印 "tick task" 以及 "promise resolve"
nextTicks = [() => {
console.log('tick task in timeout');
process.nextTick(() => {
console.log("tick task in timeout->tick");
})
}];
microtasks = [() => {
console.log('promise resolve in timeout');
process.nextTick(() => {
console.log('tick task in timeout promise');
});
}];
打印主线程任务中的 "timer task"，再进入下一阶段，发现 nextTicks 和 microtasks 队列为非空，执行微任务。由于 nextTicks 优先级更高，先打印 "tick task in timeout"，然后又往 nextTicks 插入 () => {console.log("tick task in timeout->tick")} ，继续执行 nextTicks 任务打印 "tick task in timeout->tick"。
此时 nextTicks 队列已空，执行 miacrotasks 队列，打印 "promise resolve in timeout"，此时又往 nextTicks 插入任务 () => {console.log('tick task in timeout promise')}，继续执行 nextTick 任务，打印 "tick task in timeout promise"。
进入 check 阶段（Immediate），为 nextTicks 添加 () => {console.log('imeediate->tick task') }，主线程打印 "immediate task"，进入下一阶段前先执行 nextTicks 任务，打印 'imeediate->tick task'。
setImmediate(() => console.log('this is set immediate 1'));
setImmediate(() => console.log('this is set immediate 2'));
setImmediate(() => console.log('this is set immediate 3'));
setTimeout(() => console.log('this is set timeout 1'), 0);
setTimeout(() => {
console.log('this is set timeout 2');
process.nextTick(() => console.log('this is process.nextTick added inside setTimeout'));
}, 0);
setTimeout(() => console.log('this is set timeout 3'), 0);
setTimeout(() => console.log('this is set timeout 4'), 0);
setTimeout(() => console.log('this is set timeout 5'), 0);
process.nextTick(() => console.log('this is process.nextTick 1'));
process.nextTick(() => {
process.nextTick(console.log.bind(console, 'this is the inner next tick inside next tick'));
});
process.nextTick(() => console.log('this is process.nextTick 2'));
process.nextTick(() => console.log('this is process.nextTick 3'));
process.nextTick(() => console.log('this is process.nextTick 4'));
得益于 Libuv 这一跨平台的高性能异步 I/O 库，使得 NodeJS 在处理 I/O 密集型任务上十分彰显优势。下面结合不同的 I/O 模型，对比分析一下 NodeJS 目前工程实践所采用的 I/O 模型的优越性。
首先理清一下阻塞和非阻塞、异步和同步的概念：
在应用程序通过 I/O 函数申请读写数据时，如果在数据就绪前进程一直在等待的，就是阻塞 I/O，即发起 I/O 请求时是阻塞的
数据从内核缓冲区到到用户内存复制过程中，需要用户进程等待，就是同步 I/O，即实际的 I/O 读写是同步的
图片来源：https://www.51cto.com/article/693213.html
在网络编程中，当调用 recvfrom 获取客户端数据时，首先会阻塞进程，等待数据通过网卡到内核缓冲区；当数据就绪后，再同步等待指代数据从内核缓冲区拷贝到用户空间，此时用户进程再进行数据处理。
同步阻塞 I/O 模型是最简单的 I/O 模型，好处是使用简单，通常在 fd 较少、数据就绪很快的场景，缺点是如果内核数据一直没准备好，则用户进程将会一直阻塞无法执行后续任务。
以网络编程为例，默认情况下socket 是 blocking 的，即函数 accept , recvfrom 等，需等待函数执行结束之后才能够返回(此时操作系统切换到其他进程执行)。accpet 等待到有 client 连接请求并接受成功之后，recvfrom 需要读取完client 发送的数据之后才能够返回
// 创建套接字
int serv_sock = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
// IP 和端口配置 ...
// ...
// 绑定 IP 和端口
bind(serv_sock, ...);
// 监听来自监听型套接字的请求
listen(serv_sock, ...);
int clnt_sock;
while (1) {
// 创建新的通信型套接字用于接收来自客户端的请求，此时会阻塞程序执行，直到有请求到来
clnt_sock = accept(serv_sock, ...);
// 接收客户端的数据，同步阻塞 I/O，等待数据就绪
recvfrom(clnt_sock, ...);
// 处理数据
handle(data);
}
图片来源：https://www.51cto.com/article/693213.html
同步非阻塞 I/O 的特点是当用户进程发起网络读请求时，如果内核缓冲区还没接收到客户端数据，会立即返回 EWOULDBLOCK 错误码，而不会阻塞用户进程，用户进程可结合轮询调用方式继续发起 recvfrom 调用，直到数据就绪，然后同步等待数据从内核缓冲区复制到用户空间，然后用户进程进行数据处理。
同步非阻塞 I/O 的优势在于当发起 I/O 请求时不会阻塞用户进程，一定程度上提升了程序的性能，但是为了及时获取数据的就绪状态，需要频繁轮询，这样也会消耗不小的 CPU 资源。
以网络编程为例，可设置 socket 为 non-blocking 模式，使用 socket()创建的 socket 默认是阻塞的；可使用函数 fcntl 可设置创建的 socket 为非阻塞的，这样使用原本 blocking 的各种函数（accept、recvfrom），可以立即获得返回结果。通过判断返回的errno了解状态：
这样就实现同步非阻塞 I/O 请求：
// 创建套接字
int serv_sock = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
// 设置 socket 为非阻塞 I/O
int flags = fcntl(serv_sock, F_GETFL, 0);
fcntl(serv_sock, F_SETFL, flags | O_NONBLOCK);
// IP 和端口配置 ...
// ...
// 绑定 IP 和端口
bind(serv_sock, ...);
// 监听来自监听型套接字的请求
listen(serv_sock, ...);
// 创建新的通信型套接字用于接收来自客户端的请求
int clnt_sock;
while (1) {
// 在non-blocking模式下，如果返回值为-1，且 errno == EAGAIN 或errno == EWOULDBLOCK 表示no connections 没有新连接请求；
clnt_sock = accept(serv_sock, ...);
if (clnt_sock == -1 && errno == EAGAIN) {
fprintf(stderr, "no client connectionsn");
continue;
} else if (clnt_sock == -1) {
perror("accept failed");
}
// 接收客户端的数据，同步非阻塞 I/O，在non-blocking模式下，如果返回值为-1，且 errno == EAGAIN表示没有可接受的数据或正在接受尚未完成；
while (1) {
int ret = recvfrom(clnt_sock, ...);
if (ret == -1 && errno == EAGAIN) {
fprintf(stderr, "no data readyn");
continue;
} else if (ret == -1) {
perror("read failed");
}
// 处理数据
handle(data);
}
}
图片来源：https://www.51cto.com/article/693213.html
上述两种 I/O 模型均是面向单个客户端连接的，同一时间只能处理一个 client 请求，虽然可以通过多进程/多线程的方法解决，但是多进程/多线程需要考虑额外的资源消耗以及同步互斥的相关问题。
为了高效解决多个 fd 的状态监听，I/O 多路复用技术应运而生。
I/O 多路复用的核心思想是可以同时监听多个不同的 fd（网络环境下即是网络套接字），当套接字中的任何一个数据就绪了，就可以通知用户进程，此时用户进程再发起 recvfrom 请求去读取数据。
以网络编程为例，可通过维护一个需要监听的所有 socket 的 fd 列表，然后调用 select/epoll 等监听函数，如果 fd 列表中所有 socket 都没有数据就绪，则 select/epoll 会阻塞，直到有一个 socket 接收到数据，然后唤醒进程。
int serv_sock = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
// 设置 socket 为非阻塞 I/O
int flags = fcntl(serv_sock, F_GETFL, 0);
fcntl(serv_sock, F_SETFL, flags | O_NONBLOCK);
// IP 和端口配置 ...
// ...
// 绑定 IP 和端口
bind(serv_sock, ...);
// 监听来自监听型套接字的请求
listen(serv_sock, ...);
// 存放需要监听的 socket 列表
fd_set readfds;
// 添加 需要监听的 socket 到 readfds
FD_SET(serv_sock, readfds);
// 创建新的通信型套接字用于接收来自客户端的请求
int clnt_sock;
// 调用 select 返回的结果值
int res;
// 预计可接受的最大连接数量，select 最大支持 1024 个 fd
int maxfd = 1000;
while (1) {
// 调用 select 阻塞监听 fd 列表，直到有一个 socket 接收到请求，唤醒进程
res = select(maxfd + 1, &readfds, ...);
if (res == -1) {
perror("select failed");
exit(EXIT_FAILURE);
} else if (res == 0) {
fprintf(stderr, "no socket ready for readn");
}
// 遍历每个 socket，如果是 serv_sock 则 accept，否则进行读操作
for (int i = 0; i <= maxfd; i++) {
// 是否 socket 是否在 监听的 fd 列表中
if (!FD_ISSET(i, &readfds)) {
continue;
}
if (i == serv_sock) {
// 当前请求是 server sock，则建立 accept 连接
clnt_sock = accpet(serv_sock, ...);
// 将新建立的客户端连接添加进行 readfds 监听列表中
FD_SET(clnt_sock, &readfds);
} else {
// 当请求是客户端的 socket，接收客户端的数据，此时数据已经就绪，将数据从内核空间复制到用户空间
int ret = recvfrom(i, ...);
if (ret == -1 && errno == EAGAIN) {
fprintf(stderr, "no data readyn");
continue;
} else if (ret == -1) {
perror("read failed");
}
// 处理数据
handle(data);
}
}
}
上面是使用 select 函数实现的 I/O 多路复用，实际在 Libuv 采用的是 epoll 函数，epoll 函数是为了解决 select 的以下缺点而诞生的：
图片来源：https://www.51cto.com/article/693213.html
epoll 的优势在于：
int serv_sock = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
// 设置 socket 为非阻塞 I/O
int flags = fcntl(serv_sock, F_GETFL, 0);
fcntl(serv_sock, F_SETFL, flags | O_NONBLOCK);
// IP 和端口配置 ...
// ...
// 绑定 IP 和端口
bind(serv_sock, ...);
// 监听来自监听型套接字的请求
listen(serv_sock, ...);
// 创建 epoll 对象
int MAX_EVENTS = 5; // 告诉内核可能需要监听的 fd 数量，如果使用时大于该数，则内核会申请动态申请工多空间
int epoll_fd = epoll_create(MAX_EVENTS);
if (epoll_fd == -1) {
printf("epoll_create error!n");
return -1;
}
// 注册 serv_sock 所监听的事件
struct epoll_event ev;
struct epoll_event events[MAX_EVENTS];
ev.data.fd = serv_sock; // 设置该事件的 fd 为 serv_sock
ev.events = EPOLLIN; // 设置监听 serv_sock 的可读事件
// 添加 serv_sock 到 epoll 的可读事件监听队列中
int ret = epoll_ctl(epoll_fd, EPOLL_CTL_ADD, serv_sock, &ev);
if (ret == -1) {
printf("epoll_ctl error!n");
return -1;
}
int connfd = 0; // 与客户端连接成功后的通信型 fd
while (1) {
// int epoll_wait(int epfd, struct epoll_event *events,int maxevents, int timeout);
// 等待 epoll_fd 中的事件，如果 serv_sock 有可读事件发生，则函数返回就绪后的 fd 数量
// 最后一个 timeout 参数可用来控制 epoll_wait 的等待事件发生的时间，-1 为阻塞等待，0 为非阻塞立即返回
int nfds = epoll_wait(epoll_fd, events, MAX_EVENTS, -1);
for (int i = 0; i < nfds; i++) {
// 客户端发起请求
if (events[i].data.fd == serv_sock) {
connfd = accept(serv_sock, ...);
if (connfd == -1) {
printf("accept error!n");
}
ev.data.fd = connfd; // 设置该事件的 fd 为当前的 connfd
ev.events = EPOLLIN; // 设置当前的 connfd 的可读事件
// 添加当前 connfd 到 epoll 的可读事件监听队列中
if (epoll_ctl(epoll_fd, EPOLL_CTL_ADD, connfd, &ev) == -1) {
printf("epoll_ctl add error!n");
return -1;
}
} else {
// 某个来自客户端的请求的数据已就绪
int ret = recvfrom(i, ...);
if (ret == -1 && errno == EAGAIN) {
fprintf(stderr, "no data readyn");
continue;
} else if (ret == -1) {
perror("read failed");
}
// 处理数据
handle(data);
}
}
}
以上就是基于 epoll 机制的事件驱动型的 I/O 多路复用模型，服务器通过注册文件描述符及其对应监听的事件到 epoll（epoll_ctl），epoll 开始阻塞监听事件直到有某个 fd 的监听事件触发（epoll_wait），然后就遍历就绪事件，根据 fd 类型的不同执行不同的任务。
int serv_sock = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
// IP 和端口配置 ...
// ...
// 绑定 IP 和端口
bind(serv_sock, ...);
// 监听来自监听型套接字的请求
listen(serv_sock, ...);
int clnt_sock;
while (1) {
// 创建新的通信型套接字用于接收来自客户端的请求，此时会阻塞程序执行，直到有请求到来
clnt_sock = accept(serv_sock, ...);
// 接收客户端的数据，同步阻塞 I/O，等待数据就绪
recvfrom(clnt_sock, ...);
// 处理数据
handle(data);
}
单进程单线程·串行处理请求是最简单的服务器架构，先从经过三次握手，然后从连接队列中获取客户端连接节点（accept 返回的套接字），然后从客户端的套接字获取数据进行处理，接下来再进行下个连接节点处理。
在并发连接数较大的情况下，并且采用的是阻塞式 I/O 模型，那么处理客户端连接的效率就会非常低。
单进程串行处理请求因为阻塞 I/O 导致连接队列中的节点被阻塞导致处理效率低下，通过把请求分给多个进程处理从而提升效率，人多力量大。在多进程/多线程架构下，如果一个请求发送阻塞 I/O，那么操作系统会挂起该进程，接着调度其他进程，实现并发处理能力的提高。
但这种架构模式下的性能瓶颈在于系统的进程数、线程数是有限的，开辟进程和线程的开销也是需要考虑的问题，系统资源消耗高。
int serv_sock = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
// IP 和端口配置 ...
// ...
// 绑定 IP 和端口
bind(serv_sock, ...);
// 监听来自监听型套接字的请求
listen(serv_sock, ...);
int clnt_sock;
int pid;
while (1) {
// 创建新的通信型套接字用于接收来自客户端的请求，此时会阻塞程序执行，直到有请求到来
clnt_sock = accept(serv_sock, ...);
pid = fork();
if (pid < 0) {
perror("fork failed");
return -1;
}
if (fork() > 0) {
// 父进程
continue;
} else {
// 子进程
// 接收客户端的数据，同步阻塞 I/O，等待数据就绪
recvfrom(clnt_sock, ...);
// 处理数据
handle(data);
}
}
除了通过多进程/多线程方式去应对并发量大的场景，基于 I/O 多路复用模型的单进程单线程·事件驱动架构也是较好的解决方案，同时由于是单线程，所以不会因开辟大量进场/线程所带来的资源开销以及同步互斥的问题。
单线程不适合执行 CPU 密集型任务，因为如果任务一直占用 CPU 时间，则后续任务无法执行，因此针对大量 CPU 计算、引起进程阻塞的任务，可引入线程池技术去解决。
目前 NodeJS 就是采用这种设计架构，所有 JS 代码跑在主线程中（单线程），基于 I/O 多路复用的模型去实现事件驱动的多读写请求的管理，配合线程池，将 CPU 密集型任务从主线程分离出来，以保证主线程的高效响应。
注册 C++ 系列的模块和 V8 的初始化操作
创建 environment 对象用于存放一些全局的公共变量
初始化模块加载器，以便在用户 JS 代码层调用原生 JS 模块以及原生 JS 模块调用 C++ 模块能够成功加载
初始化执行上下文，暴露 global 在全局上下文中，并设置一些全局变量和方法在 global 或 process 对象
初始化 libuv，创建一个默认的 event_loop 结构体用于管理后续各个阶段产生的任务
紧接着 NodeJS 执行用户 JS 代码，用户 JS 代码执行一些初始化的逻辑以及往事件循环注册任务，然后进程就进入事件循环的阶段。
整个事件循环分为 7 个阶段，timer 处理定时器任务，pending 处理 poll io 阶段的成功或错误回调，idle、prepare、check 是自定义阶段，poll io 主要处理网络 I/O，文件 I/O等任务，close 处理关闭的回调任务，同时在各个事件阶段还会穿插微任务队列。
以开篇的 TCP 服务为例，当创建 TCP 服务器调用原生 JS 的 net 模块的 server.listen 方法后， net 模块就会引用 C++ 的 TCP 模块实例化一个 TCP 服务器，内部调用了 Libuv 的 uv_tcp_init 方法，该方法封装了 C 中用于创建套接字的 socket 函数；接着就是调用 C++ 的 TCP 模块的 Bind 方法，该方法封装了 Libuv 的 uv_ip_addr 以及 uv_tcp_bind，分别用于设置 TCP 的 IP 地址和端口信息以及调用 C 中的 bind 方法用于绑定地址信息。
然后 net 模块注册 onconnect 回调函数，该函数将在客户端请求到来后，在 Libuv 的 poll io 阶段执行，onconnect 函数调用了 C++ 的 ConnectionWrap::OnConnection 方法，内部调用了 Libuv 的 uv_accpet 去接收来自客户端的连接。最后调用 TCP 实例的 listen 方法使得服务器进入被动监听状态，listen 使用了 C++ 的 TCPWrap::Listen 方法，该方法是对 uv_listen 的封装，最终调用的 C 的 listen 方法。
当客户端请求通过网卡传递过来，对应的监听型 socket 发生状态变更，事件循环模块根据命中之前设置的可读事件，将 onconnection 回调插入 poll io 阶段的任务队列，当新一轮的事件循环到达 poll io 时执行回调，调用 accept 方法创建与客户端的通信型 socket，此时进入进程阻塞，经过三次握手后，建立与客户端的连接，将用户 JS 的回调插入 poll io 的任务队列，在新一轮的事件循环中进行数据的处理。
image.png
TCP 服务器在启动之后，就往 NodeJS 的事件循环系统插入 listen 的监听任务，该任务会一直阻塞监听（不超过 timeout）来自客户端的请求，当发生请求后，建立连接然后进行数据处理后，再会进入监听请求的阻塞状态，新一轮的事件循环发现 poll io 队列还有任务所以不会退出事件循环，从而驱动进程一直运行。
NodeJS 采用的是单线程+事件驱动的服务端架构，首先对于事件循环以外的代码会在初始化时执行完，然后进程就进入事件循环，针对网络 I/O NodeJS 底层采用的是 I/O 多路复用模型，通过监听就绪的连接做到从容应对大并发连接。对于网络数据而言，当调用阻塞的 recvfrom 处理来自的网络的数据，此时数据已经就绪，所以数据处理起来很快，如果是大文件，则需要业务代码自行开辟线程去处理；对于文件 I/O，NodeJS 底层采用线程池的机制，在主线程外开辟工作线程去处理本地大文件，在处理完后通过事件通知机制告诉上层 JS 代码。
京公网安备 11010802035086号