进程是操作系统里非常重要的概念，也是不容易理解的概念，但是看起来很复杂的进程，其实在操作系统的代码里，也只是一些数据结构和算法，只不过它比一般的数据结构和算法更复杂。进程在操作系统里，是用一个task_struct结构体表示的。因为操作系统是大部分是用C语言实现的，没有对象这个概念。如果我们用JS来理解的话，每个进程就是一个对象，每次新建一个进程，就是新建一个对象。task_struct结构体里保存了一个进程所需要的一些信息，包括执行状态、执行上下文、打开的文件、根目录、工作目录、收到的信号、信号处理函数、代码段、数据段的信息、进程id、执行时间、退出码等等。本章将会介绍Node.js进程模块的原理和实现。
当我们执行node index.js的时候，操作系统就会创建一个Node.js进程，我们的代码就是在这个Node.js进程中执行。从代码角度来说，我们在Node.js中感知进程的方式是通过process对象。本节我们分析一下这个对象。
Node.js启动的时候会执行以下代码创建process对象（env.cc）。
Local<Object> process_object = node::CreateProcessObject(this).FromMaybe(Local<Object>()); set_process_object(process_object); // process对象通过CreateProcessObject创建，然后保存到env对象中。我们看一下CreateProcessObject。 MaybeLocal<Object> CreateProcessObject(Environment* env) { Isolate* isolate = env->isolate(); EscapableHandleScope scope(isolate); Local<Context> context = env->context(); Local<FunctionTemplate> process_template = FunctionTemplate::New(isolate); process_template->SetClassName(env->process_string()); Local<Function> process_ctor; Local<Object> process; // 新建process对象 if (!process_template->GetFunction(context).ToLocal(&process_ctor) || !process_ctor->NewInstance(context).ToLocal(&process)) { return MaybeLocal<Object>(); } // 设置一系列属性，这就是我们平时通过process对象访问的属性 // Node.js的版本 READONLY_PROPERTY(process,"version", FIXED_ONE_BYTE_STRING(env->isolate(), NODE_VERSION)); // 忽略其他属性 return scope.Escape(process); }
这是使用V8创建一个对象的典型例子，并且设置了一些属性。Node.js启动过程中，很多地方都会给process挂载属性。下面我们看我们常用的process.env是怎么挂载的。
Local<String> env_string = FIXED_ONE_BYTE_STRING(isolate_, "env"); Local<Object> env_var_proxy; // 设置process的env属性 if (!CreateEnvVarProxy(context(), isolate_, as_callback_data()) .ToLocal(&env_var_proxy) || process_object()->Set(context(), env_string, env_var_proxy).IsNothing()) { return MaybeLocal<Value>(); }
上面的代码通过CreateEnvVarProxy创建了一个对象，然后保存到env_var_proxy中，最后给process挂载了env属性。它的值是CreateEnvVarProxy创建的对象。
MaybeLocal<Object> CreateEnvVarProxy(Local<Context> context, Isolate* isolate, Local<Object> data) { EscapableHandleScope scope(isolate); Local<ObjectTemplate> env_proxy_template = ObjectTemplate::New(isolate); env_proxy_template->SetHandler(NamedPropertyHandlerConfiguration( EnvGetter, EnvSetter, EnvQuery, EnvDeleter, EnvEnumerator, data, PropertyHandlerFlags::kHasNoSideEffect)); return scope.EscapeMaybe(env_proxy_template->NewInstance(context)); }
CreateEnvVarProxy首先申请一个对象模板，然后设置通过该对象模板创建的对象的访问描述符。我们看一下getter描述符（EnvGetter）的实现，getter描述符和我们在JS里使用的类似。
static void EnvGetter(Local<Name> property, const PropertyCallbackInfo<Value>& info) { Environment* env = Environment::GetCurrent(info); MaybeLocal<String> value_string = env->env_vars()->Get(env->isolate(), property.As<String>()); if (!value_string.IsEmpty()) { info.GetReturnValue().Set(value_string.ToLocalChecked()); } }
我们看到getter是从env->env_vars()中获取数据，那么env->env_vars()又是什么呢？env_vars是一个kv存储系统，其实就是一个map。它只在Node.js初始化的时候设置（创建env对象时）。
set_env_vars(per_process::system_environment);
那么per_process::system_environment又是什么呢？我们继续往下看，
std::shared_ptr<KVStore> system_environment = std::make_shared<RealEnvStore>();
我们看到system_environment是一个RealEnvStore对象。我们看一下RealEnvStore类的实现。
class RealEnvStore final : public KVStore { public: MaybeLocal<String> Get(Isolate* isolate, Local<String> key) const override; void Set(Isolate* isolate, Local<String> key, Local<String> value) override; int32_t Query(Isolate* isolate, Local<String> key) const override; void Delete(Isolate* isolate, Local<String> key) override; Local<Array> Enumerate(Isolate* isolate) const override; };
比较简单，就是增删改查，我们看一下查询Get的实现。
MaybeLocal<String> RealEnvStore::Get(Isolate* isolate, Local<String> property) const { Mutex::ScopedLock lock(per_process::env_var_mutex); node::Utf8Value key(isolate, property); size_t init_sz = 256; MaybeStackBuffer<char, 256> val; int ret = uv_os_getenv(*key, *val, &init_sz); if (ret >= 0) { // Env key value fetch success. MaybeLocal<String> value_string = String::NewFromUtf8(isolate, *val, NewStringType::kNormal, init_sz); return value_string; } return MaybeLocal<String>(); }
我们看到是通过uv_os_getenv获取的数据。uv_os_getenv是对getenv函数的封装，进程的内存布局中，有一部分是用于存储环境变量的，getenv就是从那一块内存中把数据读取出来。我们执行execve的时候可以设置环境变量。具体的我们在子进程章节会看到。至此，我们知道process的env属性对应的值就是进程环境变量的内容。
在Node.js的启动过程中会不断地挂载属性到process。主要在bootstrap/node.js中。不一一列举。
const rawMethods = internalBinding('process_methods'); process.dlopen = rawMethods.dlopen; process.uptime = rawMethods.uptime; process.nextTick = nextTick;
下面是process_methods模块导出的属性，主列出常用的。
env->SetMethod(target, "memoryUsage", MemoryUsage); env->SetMethod(target, "cpuUsage", CPUUsage); env->SetMethod(target, "hrtime", Hrtime); env->SetMethod(target, "dlopen", binding::DLOpen); env->SetMethodNoSideEffect(target, "uptime", Uptime);
我们看到在JS层访问process属性的时候，访问的是对应的C++层的这些方法，大部分也只是对Libuv的封装。另外在Node.js初始化的过程中会执行PatchProcessObject。PatchProcessObject函数会挂载一些额外的属性给process。
// process.argv process->Set(context, FIXED_ONE_BYTE_STRING(isolate, "argv"), ToV8Value(context, env->argv()).ToLocalChecked()).Check(); READONLY_PROPERTY(process, "pid", Integer::New(isolate, uv_os_getpid())); CHECK(process->SetAccessor(context, FIXED_ONE_BYTE_STRING(isolate, "ppid"), GetParentProcessId).FromJust())
在Node.js初始化的过程中，在多个地方都会给process对象挂载属性，这里只列出了一部分，有兴趣的同学可以从bootstrap/node.js的代码开始看都挂载了什么属性。因为Node.js支持多线程，所以针对线程的情况，有一些特殊的处理。
const perThreadSetup = require('internal/process/per_thread'); // rawMethods来自process_methods模块导出的属性 const wrapped = perThreadSetup.wrapProcessMethods(rawMethods); process.hrtime = wrapped.hrtime; process.cpuUsage = wrapped.cpuUsage; process.memoryUsage = wrapped.memoryUsage; process.kill = wrapped.kill; process.exit = wrapped.exit;
大部分函数都是对process_methods模块（node_process_methods.cc）的封装。但是有一个属性我们需要关注一下，就是exit，因为在线程中调用process.exit的时候，只会退出单个线程，而不是整个进程。
function exit(code) { if (code || code === 0) process.exitCode = code; if (!process._exiting) { process._exiting = true; process.emit('exit', process.exitCode || 0); } process.reallyExit(process.exitCode || 0); }
我们继续看reallyExit
static void ReallyExit(const FunctionCallbackInfo<Value>& args) { Environment* env = Environment::GetCurrent(args); RunAtExit(env); int code = args[0]->Int32Value(env->context()).FromMaybe(0); env->Exit(code); }
调用了env的Exit。
void Environment::Exit(int exit_code) { if (is_main_thread()) { stop_sub_worker_contexts(); DisposePlatform(); exit(exit_code); } else { worker_context_->Exit(exit_code); } }
这里我们看到了重点，根据当前是主线程还是子线程会做不同的处理。一个线程会对应一个env，env对象中的worker_context_保存就是线程对象（Worker）。我们先看子线程的逻辑。
void Worker::Exit(int code) { Mutex::ScopedLock lock(mutex_); if (env_ != nullptr) { exit_code_ = code; Stop(env_); } else { stopped_ = true; } } int Stop(Environment* env) { env->ExitEnv(); return 0; } void Environment::ExitEnv() { set_can_call_into_js(false); set_stopping(true); isolate_->TerminateExecution(); // 退出Libuv事件循环 SetImmediateThreadsafe([](Environment* env) { uv_stop(env->event_loop()); }); }
我们看到子线程最后调用uv_stop提出了Libuv事件循环，然后退出。我们再来看主线程的退出逻辑。
if (is_main_thread()) { stop_sub_worker_contexts(); DisposePlatform(); exit(exit_code); }
我们看到最后主进程中调用exit退出进程。但是退出前还有一些处理工作，我们看stop_sub_worker_contexts
void Environment::stop_sub_worker_contexts() { while (!sub_worker_contexts_.empty()) { Worker* w = *sub_worker_contexts_.begin(); remove_sub_worker_context(w); w->Exit(1); w->JoinThread(); } }
sub_worker_contexts保存的是Worker对象列表，每次创建一个线程的时候，就会往里追加一个元素。这里遍历这个列表，然后调用Exit函数，这个刚才我们已经分析过，就是退出Libuv事件循环。主线程接着调JoinThread，JoinThread主要是为了阻塞等待子线程退出，因为子线程在退出的时候，可能会被操作系统挂起（执行时间片到了），这时候主线程被调度执行，但是这时候主线程还不能退出，所以这里使用join阻塞等待子线程退出。Node.js的JoinThread除了对线程join函数的封装。还做了一些额外的事情，比如触发exit事件。
因为Node.js是单进程的，但有很多事情可能不适合在主进程里处理的，所以Node.js提供了子进程模块，我们可以创建子进程做一些额外任务的处理，另外，子进程的好处是，一旦子进程出问题挂掉不会影响主进程。我们首先看一下在用C语言如何创建一个进程。
#include<unistd.h> #include<stdlib.h> int main(int argc,char *argv[]){ pid_t pid = fork(); if (pid < 0) { // 错误 } else if(pid == 0) { // 子进程，可以使用exec*系列函数执行新的程序 } else { // 父进程 } }
fork函数的特点，我们听得最多的可能是执行一次返回两次，我们可能会疑惑，执行一个函数怎么可能返回了两次呢？之前我们讲过，进程是task_struct表示的一个实例，调用 fork的时候，操作系统会新建一个新的task_struct实例出来（变成两个进程），fork返回两次的意思其实是在在两个进程分别返回一次，执行的都是fork后面的一行代码。而操作系统根据当前进程是主进程还是子进程，设置了fork函数的返回值。所以不同的进程，fork返回值不一样，也就是我们代码中if else条件。但是fork只是复制主进程的内容，如果我们想执行另外一个程序，怎么办呢？这时候就需要用到exec*系列函数，该系列函数会覆盖旧进程（task_struct）的部分内容，重新加载新的程序内容。这也是Node.js中创建子进程的底层原理。Node.js虽然提供了很多种创建进程的方式，但是本质上是同步和异步两种方式。
我们首先看一下异步方式创建进程时的关系图如图13-1所示。 图13-1 我们从fork这个函数开始，看一下整个流程。
function fork(modulePath /* , args, options */) { // 一系列参数处理 return spawn(options.execPath, args, options); }
我们接着看spawn
var spawn = exports.spawn = function(/*file, args, options*/) { var opts = normalizeSpawnArguments.apply(null, arguments); var options = opts.options; var child = new ChildProcess(); child.spawn({ file: opts.file, args: opts.args, cwd: options.cwd, windowsHide: !!options.windowsHide, windowsVerbatimArguments: !!options.windowsVerbatimArguments, detached: !!options.detached, envPairs: opts.envPairs, stdio: options.stdio, uid: options.uid, gid: options.gid }); return child; };
我们看到spawn函数只是对ChildProcess的封装。然后调用它的spawn函数。我们看看ChildProcess。
function ChildProcess() { // C++层定义 this._handle = new Process(); } ChildProcess.prototype.spawn = function(options) { // 创建进程 const err = this._handle.spawn(options); }
ChildProcess是对C++层的封装，不过Process在C++层也没有太多逻辑，进行参数的处理然后调用Libuv的uv_spawn。我们通过uv_spawn来到了C语言层。我们看看uv_spawn的整体流程。
int uv_spawn(uv_loop_t* loop, uv_process_t* process, const uv_process_options_t* options) { uv__handle_init(loop, (uv_handle_t*)process, UV_PROCESS); QUEUE_INIT(&process->queue); // 处理进程间通信 for (i = 0; i < options->stdio_count; i++) { err = uv__process_init_stdio(options->stdio + i, pipes[i]); if (err) goto error; } /* 创建一个管道用于创建进程期间的父进程子通信， 设置UV__O_CLOEXEC标记，子进程执行execvp 的时候管道的一端会被关闭 */ err = uv__make_pipe(signal_pipe, 0); // 注册子进程退出信号的处理函数 uv_signal_start(&loop->child_watcher, uv__chld, SIGCHLD); uv_rwlock_wrlock(&loop->cloexec_lock); // 创建子进程 pid = fork(); // 子进程 if (pid == 0) { uv__process_child_init(options, stdio_count, pipes, signal_pipe[1]); abort(); } // 父进程 uv_rwlock_wrunlock(&loop->cloexec_lock); // 关闭管道写端，等待子进程写 uv__close(signal_pipe[1]); process->status = 0; exec_errorno = 0; // 判断子进程是否执行成功 do r = read(signal_pipe[0],&exec_errorno,sizeof(exec_errorno)); while (r == -1 && errno == EINTR); // 忽略处理r的逻辑 // 保存通信的文件描述符到对应的数据结构 for (i = 0; i < options->stdio_count; i++) { uv__process_open_stream(options->stdio + i, pipes[i]); } // 插入Libuv事件循环的结构体 if (exec_errorno == 0) { QUEUE_INSERT_TAIL(&loop->process_handles, &process->queue); uv__handle_start(process); } process->pid = pid; process->exit_cb = options->exit_cb; return exec_errorno; }
uv_spawn的逻辑大致分为下面几个 1 处理进程间通信 2 注册子进程退出处理函数 3 创建子进程 4 插入Libuv事件循环的process_handles对象，保存状态码和回调等。 我们分析2,3，进程间通信我们单独分析。 1 处理子进程退出 主进程在创建子进程之前，会注册SIGCHLD信号。对应的处理函数是uvchld。当进程退出的时候。Node.js主进程会收到SIGCHLD信号。然后执行uvchld。该函数遍历Libuv进程队列中的节点，通过waitpid判断该节点对应的进程是否已经退出后，从而处理已退出的节点，然后移出Libuv队列，最后执行已退出进程的回调。
static void uv__chld(uv_signal_t* handle, int signum) { uv_process_t* process; uv_loop_t* loop; int exit_status; int term_signal; int status; pid_t pid; QUEUE pending; QUEUE* q; QUEUE* h; // 保存进程（已退出的状态）的队列 QUEUE_INIT(&pending); loop = handle->loop; h = &loop->process_handles; q = QUEUE_HEAD(h); // 收集已退出的进程 while (q != h) { process = QUEUE_DATA(q, uv_process_t, queue); q = QUEUE_NEXT(q); do /* WNOHANG非阻塞等待子进程退出，其实就是看子进程是否退出了， 没有的话就直接返回，而不是阻塞 */ pid = waitpid(process->pid, &status, WNOHANG); while (pid == -1 && errno == EINTR); if (pid == 0) continue; /* 进程退出了，保存退出状态，移出队列， 插入peding队列，等待处理 */ process->status = status; QUEUE_REMOVE(&process->queue); QUEUE_INSERT_TAIL(&pending, &process->queue); } h = &pending; q = QUEUE_HEAD(h); // 是否有退出的进程 while (q != h) { process = QUEUE_DATA(q, uv_process_t, queue); q = QUEUE_NEXT(q); QUEUE_REMOVE(&process->queue); QUEUE_INIT(&process->queue); uv__handle_stop(process); if (process->exit_cb == NULL) continue; exit_status = 0; // 获取退出信息，执行上传回调 if (WIFEXITED(process->status)) exit_status = WEXITSTATUS(process->status); // 是否因为信号而退出 term_signal = 0; if (WIFSIGNALED(process->status)) term_signal = WTERMSIG(process->status); process->exit_cb(process, exit_status, term_signal); } }
当主进程下的子进程退出时，父进程主要负责收集子进程退出状态和原因等信息，然后执行上层回调。
2 创建子进程（uvprocess_child_init） 主进程首先使用uvmake_pipe申请一个匿名管道用于主进程和子进程通信，匿名管道是进程间通信中比较简单的一种，它只用于有继承关系的进程，因为匿名，非继承关系的进程无法找到这个管道，也就无法完成通信，而有继承关系的进程，是通过fork出来的，父子进程可以获得得到管道。进一步来说，子进程可以使用继承于父进程的资源，管道通信的原理如图13-2所示。 图13-2 主进程和子进程通过共享file和inode结构体，实现对同一块内存的读写。主进程fork创建子进程后，会通过read阻塞等待子进程的消息。我们看一下子进程的逻辑。
static void uv__process_child_init(const uv_process_options_t* options, int stdio_count, int (*pipes)[2], int error_fd) { sigset_t set; int close_fd; int use_fd; int err; int fd; int n; // 省略处理文件描述符等参数逻辑 // 处理环境变量 if (options->env != NULL) { environ = options->env; } // 处理信号 for (n = 1; n < 32; n += 1) { // 这两个信号触发时，默认行为是进程退出且不能阻止的 if (n == SIGKILL || n == SIGSTOP) continue; /* Can't be changed. */ // 设置为默认处理方式 if (SIG_ERR != signal(n, SIG_DFL)) continue; // 出错则通知主进程 uv__write_int(error_fd, UV__ERR(errno)); _exit(127); } // 加载新的执行文件 execvp(options->file, options->args); // 加载成功则不会走到这，走到这说明加载执行文件失败 uv__write_int(error_fd, UV__ERR(errno)); _exit(127); }
子进程的逻辑主要是处理文件描述符、信号、设置环境变量等。然后加载新的执行文件。因为主进程和子进程通信的管道对应的文件描述符设置了cloexec标记。所以当子进程加载新的执行文件时，就会关闭用于和主进程通信的管道文件描述符，从而导致主进程读取管道读端的时候返回0，这样主进程就知道子进程成功执行了。
同步方式创建的进程，主进程会等待子进程退出后才能继续执行。接下来看看如何以同步的方式创建进程。JS层入口函数是spawnSync。spawnSync调用C++模块spawn_sync的spawn函数创建进程，我们看一下对应的C++模块spawn_sync导出的属性。
void SyncProcessRunner::Initialize(Local<Object> target, Local<Value> unused, Local<Context> context, void* priv) { Environment* env = Environment::GetCurrent(context); env->SetMethod(target, "spawn", Spawn); }
该模块值导出了一个属性spawn，当我们调用spawn的时候，执行的是C++的Spawn。
void SyncProcessRunner::Spawn(const FunctionCallbackInfo<Value>& args) { Environment* env = Environment::GetCurrent(args); env->PrintSyncTrace(); SyncProcessRunner p(env); Local<Value> result; if (!p.Run(args[0]).ToLocal(&result)) return; args.GetReturnValue().Set(result); }
Spawn中主要是新建了一个SyncProcessRunner对象并且执行Run方法。我们看一下SyncProcessRunner的Run做了什么。
MaybeLocal<Object> SyncProcessRunner::Run(Local<Value> options) { EscapableHandleScope scope(env()->isolate()); Maybe<bool> r = TryInitializeAndRunLoop(options); Local<Object> result = BuildResultObject(); return scope.Escape(result); }
执行了TryInitializeAndRunLoop。
Maybe<bool> SyncProcessRunner::TryInitializeAndRunLoop(Local<Value> options) { int r; lifecycle_ = kInitialized; // 新建一个事件循环 uv_loop_ = new uv_loop_t; if (!ParseOptions(options).To(&r)) return Nothing<bool>(); if (r < 0) { SetError(r); return Just(false); } // 设置子进程执行的时间 if (timeout_ > 0) { r = uv_timer_init(uv_loop_, &uv_timer_); uv_unref(reinterpret_cast<uv_handle_t*>(&uv_timer_)); uv_timer_.data = this; kill_timer_initialized_ = true; // 开启一个定时器，超时执行KillTimerCallback r = uv_timer_start(&uv_timer_, KillTimerCallback, timeout_, 0); } // 子进程退出时处理函数 uv_process_options_.exit_cb = ExitCallback; // 传进去新的loop而不是主进程本身的loop r = uv_spawn(uv_loop_, &uv_process_, &uv_process_options_); uv_process_.data = this; for (const auto& pipe : stdio_pipes_) { if (pipe != nullptr) { r = pipe->Start(); if (r < 0) { SetPipeError(r); return Just(false); } } } // 开启一个新的事件循环 r = uv_run(uv_loop_, UV_RUN_DEFAULT); return Just(true); }
从上面的代码中，我们可以了解到Node.js是如何实现同步创建进程的。同步创建进程时，Node.js重新开启了一个事件循环，然后新建一个子进程，并且把表示子进程结构体的handle插入到新创建的事件循环中，接着Libuv一直处于事件循环中，因为一直有一个uv_process_t（handle），所以新创建的uv_run会一直在执行，所以这时候，Node.js主进程会”阻塞”在该uv_run。直到子进程退出，主进程收到信号后，删除新创建的事件循环中的uv_process_t。然后执行回调ExitCallback。接着事件循环退出，再次回到Node.js原来的事件循环。如图所示13-3。 图13-3 这就是同步的本质和原因。我们分几步分析一下以上代码
因为同步方式创建子进程会导致Node.js主进程阻塞，为了避免子进程有问题，从而影响主进程的执行，Node.js支持可配置子进程的最大执行时间。我们看到，Node.js开启了一个定时器，并设置了回调KillTimerCallback。
void SyncProcessRunner::KillTimerCallback(uv_timer_t* handle) { SyncProcessRunner* self = reinterpret_cast<SyncProcessRunner*>(handle->data); self->OnKillTimerTimeout(); } void SyncProcessRunner::OnKillTimerTimeout() { SetError(UV_ETIMEDOUT); Kill(); } void SyncProcessRunner::Kill() { if (killed_) return; killed_ = true; if (exit_status_ < 0) { // kill_signal_为用户自定义发送的杀死进程的信号 int r = uv_process_kill(&uv_process_, kill_signal_); // 不支持用户传的信号 if (r < 0 && r != UV_ESRCH) { SetError(r); // 回退使用SIGKILL信号杀死进程 r = uv_process_kill(&uv_process_, SIGKILL); CHECK(r >= 0 || r == UV_ESRCH); } } // Close all stdio pipes. CloseStdioPipes(); // 清除定时器 CloseKillTimer(); }
当执行时间到达设置的阈值，Node.js主进程会给子进程发送一个信号，默认是杀死子进程。
退出处理主要是记录子进程退出时的错误码和被哪个信号杀死的（如果有的话）。
void SyncProcessRunner::ExitCallback(uv_process_t* handle, int64_t exit_status, int term_signal) { SyncProcessRunner* self = reinterpret_cast<SyncProcessRunner*>(handle->data); uv_close(reinterpret_cast<uv_handle_t*>(handle), nullptr); self->OnExit(exit_status, term_signal); } void SyncProcessRunner::OnExit(int64_t exit_status, int term_signal) { if (exit_status < 0) return SetError(static_cast<int>(exit_status)); exit_status_ = exit_status; term_signal_ = term_signal; }
进程间通信是多进程系统中非常重要的功能，否则进程就像孤岛一样，不能交流信息。因为进程间的内存是隔离的，如果进程间想通信，就需要一个公共的地方，让多个进程都可以访问，完成信息的传递。在Linux中，同主机的进程间通信方式有很多，但是基本都是使用独立于进程的额外内存作为信息承载的地方，然后在通过某种方式让多个进程都可以访问到这块公共内存，比如管道、共享内存、Unix域、消息队列等等。不过还有另外一种进程间通信的方式，是不属于以上情况的，那就是信号。信号作为一种简单的进程间通信方式，操作系统提供了接口让进程可以直接修改另一个进程的数据（PCB），以此达到通信目的。本节介绍Node.js中进程间通信的原理和实现。
我们从fork函数开始分析Node.js中进程间通信的逻辑。
function fork(modulePath) { // 忽略options参数处理 if (typeof options.stdio === 'string') { options.stdio = stdioStringToArray(options.stdio, 'ipc'); } else if (!ArrayIsArray(options.stdio)) { // silent为true则是管道形式和主进程通信，否则是继承 options.stdio = stdioStringToArray( options.silent ? 'pipe' : 'inherit', 'ipc'); } else if (!options.stdio.includes('ipc')) { // 必须要IPC，支持进程间通信 throw new ERR_CHILD_PROCESS_IPC_REQUIRED('options.stdio'); } return spawn(options.execPath, args, options); }
我们看一下stdioStringToArray的处理。
function stdioStringToArray(stdio, channel) { const options = []; switch (stdio) { case 'ignore': case 'pipe': options.push(stdio, stdio, stdio); break; case 'inherit': options.push(0, 1, 2); break; default: throw new ERR_INVALID_OPT_VALUE('stdio', stdio); } if (channel) options.push(channel); return options; }
stdioStringToArray会返回一个数组，比如['pipe', 'pipe', 'pipe', 'ipc']或[0, 1, 2, 'ipc']，ipc代表需要创建一个进程间通信的通道，并且支持文件描述传递。我们接着看spawn。
ChildProcess.prototype.spawn = function(options) { let i = 0; // 预处理进程间通信的数据结构 stdio = getValidStdio(stdio, false); const ipc = stdio.ipc; // IPC文件描述符 const ipcFd = stdio.ipcFd; stdio = options.stdio = stdio.stdio; // 通过环境变量告诉子进程IPC文件描述符和数据处理模式 if (ipc !== undefined) { options.envPairs.push(`NODE_CHANNEL_FD=${ipcFd}`); options.envPairs.push(`NODE_CHANNEL_SERIALIZATION_MODE=${serialization}`); } // 创建子进程 const err = this._handle.spawn(options); this.pid = this._handle.pid; // 处理IPC通信 if (ipc !== undefined) setupChannel(this, ipc, serialization); return err; }
Spawn中会执行getValidStdio预处理进程间通信的数据结构。我们只关注ipc的。
function getValidStdio(stdio, sync) { let ipc; let ipcFd; stdio = stdio.reduce((acc, stdio, i) => { if (stdio === 'ipc') { ipc = new Pipe(PipeConstants.IPC); ipcFd = i; acc.push({ type: 'pipe', handle: ipc, ipc: true }); } else { // 其它类型的处理 } return acc; }, []); return { stdio, ipc, ipcFd }; }
我们看到这里会new Pipe(PipeConstants.IPC);创建一个Unix域用于进程间通信，但是这里只是定义了一个C++对象，还没有可用的文件描述符。我们接着往下看C++层的spawn中关于进程间通信的处理。C++层首先处理参数，
static void ParseStdioOptions(Environment* env, Local<Object> js_options, uv_process_options_t* options) { Local<Context> context = env->context(); Local<String> stdio_key = env->stdio_string(); // 拿到JS层stdio的值 Local<Array> stdios = js_options->Get(context, stdio_key).ToLocalChecked().As<Array>(); uint32_t len = stdios->Length(); options->stdio = new uv_stdio_container_t[len]; options->stdio_count = len; // 遍历stdio，stdio是一个对象数组 for (uint32_t i = 0; i < len; i++) { Local<Object> stdio = stdios->Get(context, i).ToLocalChecked().As<Object>(); // 拿到stdio的类型 Local<Value> type = stdio->Get(context, env->type_string()).ToLocalChecked(); // 创建IPC通道 if (type->StrictEquals(env->pipe_string())) { options->stdio[i].flags = static_cast<uv_stdio_flags>( UV_CREATE_PIPE | UV_READABLE_PIPE | UV_WRITABLE_PIPE); // 拿到对应的stream options->stdio[i].data.stream = StreamForWrap(env, stdio); } } }
这里会把StreamForWrap的结果保存到stream中，我们看看StreamForWrap的逻辑
static uv_stream_t* StreamForWrap(Environment* env, Local<Object> stdio) { Local<String> handle_key = env->handle_string(); /* 获取对象中的key为handle的值，即刚才JS层的 new Pipe(SOCKET.IPC); */ Local<Object> handle = stdio->Get(env->context(), handle_key).ToLocalChecked().As<Object>(); // 获取JS层使用对象所对应的C++对象中的stream uv_stream_t* stream = LibuvStreamWrap::From(env, handle)->stream(); CHECK_NOT_NULL(stream); return stream; } // 从JS层使用的object中获取关联的C++对象 ibuvStreamWrap* LibuvStreamWrap::From(Environment* env, Local<Object> object) { return Unwrap<LibuvStreamWrap>(object); }
以上代码获取了IPC对应的stream结构体。在Libuv中会把文件描述符保存到stream中。我们接着看C++层调用Libuv的uv_spawn。
int uv_spawn(uv_loop_t* loop, uv_process_t* process, const uv_process_options_t* options) { int pipes_storage[8][2]; int (*pipes)[2]; int stdio_count; // 初始化进程间通信的数据结构 stdio_count = options->stdio_count; if (stdio_count < 3) stdio_count = 3; for (i = 0; i < stdio_count; i++) { pipes[i][0] = -1; pipes[i][1] = -1; } // 创建进程间通信的文件描述符 for (i = 0; i < options->stdio_count; i++) { err = uv__process_init_stdio(options->stdio + i, pipes[i]); if (err) goto error; } // 设置进程间通信文件描述符到对应的数据结构 for (i = 0; i < options->stdio_count; i++) { uv__process_open_stream(options->stdio + i, pipes[i]); } }
Libuv中会创建用于进程间通信的文件描述符，然后设置到对应的数据结构中。
static int uv__process_open_stream(uv_stdio_container_t* container, int pipefds[2]) { int flags; int err; if (!(container->flags & UV_CREATE_PIPE) || pipefds[0] < 0) return 0; err = uv__close(pipefds[1]); if (err != 0) abort(); pipefds[1] = -1; uv__nonblock(pipefds[0], 1); flags = 0; if (container->flags & UV_WRITABLE_PIPE) flags |= UV_HANDLE_READABLE; if (container->flags & UV_READABLE_PIPE) flags |= UV_HANDLE_WRITABLE; return uv__stream_open(container->data.stream, pipefds[0], flags); }
执行完uv__process_open_stream，用于IPC的文件描述符就保存到new Pipe(SOCKET.IPC)中了。有了IPC通道的文件描述符，进程还需要进一步处理。我们看到JS层执行完spawn后，主进程通过setupChannel对进程间通信进行了进一步处理。我们看一下主进程setupChannel中关于进程间通信的处理。
1 读端
function setupChannel(target, channel, serializationMode) { // channel是new Pipe(PipeConstants.IPC); const control = new Control(channel); target.channel = control; // … channel.pendingHandle = null; // 注册处理数据的函数 channel.onread = function(arrayBuffer) { // 收到的文件描述符 const recvHandle = channel.pendingHandle; channel.pendingHandle = null; if (arrayBuffer) { const nread = streamBaseState[kReadBytesOrError]; const offset = streamBaseState[kArrayBufferOffset]; const pool = new Uint8Array(arrayBuffer, offset, nread); if (recvHandle) pendingHandle = recvHandle; // 解析收到的消息 for (const message of parseChannelMessages(channel, pool)) { // 是否是内部通信事件 if (isInternal(message)) { // 收到handle if (message.cmd === 'NODE_HANDLE') { handleMessage(message, pendingHandle, true); pendingHandle = null; } else { handleMessage(message, undefined, true); } } else { handleMessage(message, undefined, false); } } } }; function handleMessage(message, handle, internal) { const eventName = (internal ? 'internalMessage' : 'message'); process.nextTick(emit, eventName, message, handle); } // 开启读 channel.readStart(); return control; }
onread处理完后会触发internalMessage或message事件，message是用户使用的。 2写端
target._send = function(message, handle, options, callback) { let obj; const req = new WriteWrap(); // 发送给对端 const err = writeChannelMessage(channel, req, message,handle); return channel.writeQueueSize < (65536 * 2); }
我们看看writeChannelMessage
writeChannelMessage(channel, req, message, handle) { const ser = new ChildProcessSerializer(); ser.writeHeader(); ser.writeValue(message); const serializedMessage = ser.releaseBuffer(); const sizeBuffer = Buffer.allocUnsafe(4); sizeBuffer.writeUInt32BE(serializedMessage.length); // channel是封装了Unix域的对象 return channel.writeBuffer(req, Buffer.concat([ sizeBuffer, serializedMessage ]), handle); },
channel.writeBuffer通过刚才创建的IPC通道完成数据的发送，并且支持发送文件描述符。
接着我们看看子进程的逻辑，Node.js在创建子进程的时候，主进程会通过环境变量NODE_CHANNEL_FD告诉子进程Unix域通信对应的文件描述符。在执行子进程的时候，会处理这个文件描述符。具体实现在setupChildProcessIpcChannel函数中。
function setupChildProcessIpcChannel() { // 主进程通过环境变量设置该值 if (process.env.NODE_CHANNEL_FD) { const fd = parseInt(process.env.NODE_CHANNEL_FD, 10); delete process.env.NODE_CHANNEL_FD; require('child_process')._forkChild(fd, serializationMode); } }
接着执行_forkChild函数。
function _forkChild(fd, serializationMode) { const p = new Pipe(PipeConstants.IPC); p.open(fd); const control = setupChannel(process, p, serializationMode); }
该函数创建一个Pipe对象，然后把主进程传过来的fd保存到该Pipe对象。对该Pipe对象的读写，就是地对fd进行读写。最后执行setupChannel。setupChannel主要是完成了Unix域通信的封装，包括处理接收的消息、发送消息、处理文件描述符传递等，刚才已经分析过，不再具体分析。最后通过在process对象中挂载函数和监听事件，使得子进程具有和主进程通信的能力。所有的通信都是基于主进程通过环境变量NODE_CHANNEL_FD传递过来的fd进行的。
前面我们已经介绍过传递文件描述符的原理，下面我们看看Node.js是如何处理文件描述符传递的。
我们看进程间通信的发送函数send的实现
process.send = function(message, handle, options, callback) { return this._send(message, handle, options, callback); }; target._send = function(message, handle, options, callback) { // Support legacy function signature if (typeof options === 'boolean') { options = { swallowErrors: options }; } let obj; // 发送文件描述符，handle是文件描述符的封装 if (handle) { message = { cmd: 'NODE_HANDLE', type: null, msg: message }; // handle的类型 if (handle instanceof net.Socket) { message.type = 'net.Socket'; } else if (handle instanceof net.Server) { message.type = 'net.Server'; } else if (handle instanceof TCP || handle instanceof Pipe) { message.type = 'net.Native'; } else if (handle instanceof dgram.Socket) { message.type = 'dgram.Socket'; } else if (handle instanceof UDP) { message.type = 'dgram.Native'; } else { throw new ERR_INVALID_HANDLE_TYPE(); } // 根据类型转换对象 obj = handleConversion[message.type]; // 把JS层使用的对象转成C++层对象 handle=handleConversion[message.type].send.call(target, message, handle, options); } // 发送 const req = new WriteWrap(); // 发送给对端 const err = writeChannelMessage(channel, req, message, handle); }
Node.js在发送一个封装了文件描述符的对象之前，首先会把JS层使用的对象转成C++层使用的对象。如TCP
send(message, server, options) { return server._handle; }
我们接着看writeChannelMessage。
// channel是new Pipe(PipeConstants.IPC); writeChannelMessage(channel, req, message, handle) { const string = JSONStringify(message) + '\n'; return channel.writeUtf8String(req, string, handle); }
我们看一下writeUtf8String
template <enum encoding enc> int StreamBase::WriteString(const FunctionCallbackInfo<Value>& args) { Environment* env = Environment::GetCurrent(args); // new WriteWrap() Local<Object> req_wrap_obj = args[0].As<Object>(); Local<String> string = args[1].As<String>(); Local<Object> send_handle_obj; // 需要发送文件描述符，C++层对象 if (args[2]->IsObject()) send_handle_obj = args[2].As<Object>(); uv_stream_t* send_handle = nullptr; // 是Unix域并且支持传递文件描述符 if (IsIPCPipe() && !send_handle_obj.IsEmpty()) { HandleWrap* wrap; /* send_handle_obj是由C++层创建在JS层使用的对象， 解包出真正在C++层使用的对象 */ ASSIGN_OR_RETURN_UNWRAP(&wrap, send_handle_obj, UV_EINVAL); // 拿到Libuv层的handle结构体 send_handle = reinterpret_cast<uv_stream_t*>(wrap->GetHandle()); /* Reference LibuvStreamWrap instance to prevent it from being garbage，collected before`AfterWrite` is called. */ req_wrap_obj->Set(env->context(), env->handle_string(), send_handle_obj).Check(); } Write(&buf, 1, send_handle, req_wrap_obj); }
Write会调用Libuv的uvwrite，uvwrite会把Libuv层的handle中的fd取出来，使用sendmsg传递到其它进程。整个发送的过程本质是从JS层到Libuv层层层揭开要发送的对象，最后拿到一个文件描述符，然后通过操作系统提供的API把文件描述符传递给另一个进程，如图13-4所示。 图13-4
分析完发送，我们再看一下接收的逻辑。前面我们分析过，当文件描述符收到数据时，会把文件文件描述符封装成对应的对象。
void LibuvStreamWrap::OnUvRead(ssize_t nread, const uv_buf_t* buf) { HandleScope scope(env()->isolate()); Context::Scope context_scope(env()->context()); uv_handle_type type = UV_UNKNOWN_HANDLE; // 是否支持传递文件描述符并且有待处理的文件描述符，则判断文件描述符类型 if (is_named_pipe_ipc() && uv_pipe_pending_count(reinterpret_cast<uv_pipe_t*>(stream())) > 0) { type = uv_pipe_pending_type(reinterpret_cast<uv_pipe_t*>(stream())); } // 读取成功 if (nread > 0) { MaybeLocal<Object> pending_obj; // 根据类型创建一个新的C++对象表示客户端，并且从服务器中摘下一个fd保存到客户端 if (type == UV_TCP) { pending_obj = AcceptHandle<TCPWrap>(env(), this); } else if (type == UV_NAMED_PIPE) { pending_obj = AcceptHandle<PipeWrap>(env(), this); } else if (type == UV_UDP) { pending_obj = AcceptHandle<UDPWrap>(env(), this); } else { CHECK_EQ(type, UV_UNKNOWN_HANDLE); } // 保存到JS层使用的对象中，键是pendingHandle if (!pending_obj.IsEmpty()) { object() ->Set(env()->context(), env()->pending_handle_string(), pending_obj.ToLocalChecked()) .Check(); } } EmitRead(nread, *buf); }
接着我们看看JS层的处理。
channel.onread = function(arrayBuffer) { // 收到的文件描述符 const recvHandle = channel.pendingHandle; channel.pendingHandle = null; if (arrayBuffer) { const nread = streamBaseState[kReadBytesOrError]; const offset = streamBaseState[kArrayBufferOffset]; const pool = new Uint8Array(arrayBuffer, offset, nread); if (recvHandle) pendingHandle = recvHandle; // 解析收到的消息 for (const message of parseChannelMessages(channel, pool)) { // 是否是内部通信事件 if (isInternal(message)) { if (message.cmd === 'NODE_HANDLE') { handleMessage(message, pendingHandle, true); pendingHandle = null; } else { handleMessage(message, undefined, true); } } else { handleMessage(message, undefined, false); } } } };
这里会触发内部事件internalMessage
target.on('internalMessage', function(message, handle) { // 是否收到了handle if (message.cmd !== 'NODE_HANDLE') return; // 成功收到，发送ACK target._send({ cmd: 'NODE_HANDLE_ACK' }, null, true); const obj = handleConversion[message.type]; /* C++对象转成JS层使用的对象。转完之后再根据里层的字段 message.msg进一步处理，或者触发message事件传给用户 */ obj.got.call(this, message, handle, (handle) => { handleMessage(message.msg, handle, isInternal(message.msg)); }); })
我们看到这里会把C++层的对象转成JS层使用的对象。如TCP
got(message, handle, emit) { const server = new net.Server(); server.listen(handle, () => { emit(server); }); }
这就是文件描述符传递在Node.js中的处理流程，传递文件描述符是一个非常有用的能力，比如一个进程可以把一个TCP连接所对应的文件描述符直接发送给另一个进程处理。这也是cluser模块的原理。后续我们会看到。在Node.js中，整体的处理流程就是，发送的时候把一个JS层使用的对象一层层地剥开，变成C++对象，然后再变成fd，最后通过底层API传递给另一个进程。接收的时候就是把一个fd一层层地包裹，变成一个JS层使用的对象。
Copyright© 2013-2020
All Rights Reserved 京ICP备2023019179号-8