使用例子
为了让node应用能够在多核服务器中提高性能，node提供cluster API，用于创建多个工作进程，然后由这些工作进程并行处理请求。
// master.js
const cluster = require('cluster');
const cpusLen = require('os').cpus().length;
const path = require('path');
console.log(`主进程：${process.pid}`);
cluster.setupMaster({
exec: path.resolve(__dirname, './work.js'),
});
for (let i = 0; i < cpusLen; i++) {
cluster.fork();
}
// work.js
const http = require('http');
console.log(`工作进程：${process.pid}`);
http.createServer((req, res) => {
res.end('hello');
}).listen(8080);
上面例子中，使用cluster创建多个工作进程，这些工作进程能够共用8080端口，我们请求localhost:8080，请求任务会交给其中一个工作进程进行处理，该工作进程处理完成后，自行响应请求。
端口占用问题
这里有个问题，前面例子中，出现多个进程监听相同的端口，为什么程序没有报端口占用问题，由于socket套接字监听端口会有一个文件描述符，而每个进程的文件描述符都不相同，无法让多个进程都监听同一个端口，如下：
// master.js
const fork = require('child_process').fork;
const cpusLen = require('os').cpus().length;
const path = require('path');
console.log(`主进程：${process.pid}`);
for (let i = 0; i < cpusLen; i++) {
fork(path.resolve(__dirname, './work.js'));
}
// work.js
const http = require('http');
console.log(`工作进程：${process.pid}`);
http.createServer((req, res) => {
res.end('hello');
}).listen(8080);
当运行master.js文件的时候，会报端口被占用的问题（Error: listen EADDRINUSE: address already in use :::8080）。
我们修改下，只使用主进程监听端口，主进程将请求套接字发放给工作进程，由工作进程来进行业务处理。
// master.js
const fork = require('child_process').fork;
const cpusLen = require('os').cpus().length;
const path = require('path');
const net = require('net');
const server = net.createServer();
console.log(`主进程：${process.pid}`);
const works = [];
let current = 0
for (let i = 0; i < cpusLen; i++) {
works.push(fork(path.resolve(__dirname, './work.js')));
}
server.listen(8080, () => {
if (current > works.length - 1) current = 0
works[current++].send('server', server);
server.close();
});
// work.js
const http = require('http');
const server = http.createServer((req, res) => {
res.end('hello');
});
console.log(`工作进程：${process.pid}`);
process.on('message', (type, tcp) => {
if (type === 'server') {
tcp.on('connection', socket => {
server.emit('connection', socket)
});
}
})
实际上，cluster新建的工作进程并没有真正去监听端口，在工作进程中的net server listen函数会被hack，工作进程调用listen，不会有任何效果。监听端口工作交给了主进程，该端口对应的工作进程会被绑定到主进程中，当请求进来的时候，主进程会将请求的套接字下发给相应的工作进程，工作进程再对请求进行处理。
接下来我们看看cluster API中的实现，看下cluster内部是如何做到下面两个功能：
- 主进程：对传入的端口进行监听
- 工作进程：
- 主进程注册当前工作进程，如果主进程是第一次监听此端口，就新建一个TCP服务器，并将当前工作进程和TCP服务器绑定。
- hack掉工作进程中的listen函数，让该进程不能监听端口
源码解读
本文使用的是node@14.15.4。
// lib/cluster.js
'use strict';
const childOrPrimary = 'NODE_UNIQUE_ID' in process.env ? 'child' : 'primary';
module.exports = require(`internal/cluster/${childOrPrimary}`);
这个是cluster API入口，在引用cluster的时候，程序首先会判断环境变量中是否存在NODE_UNIQUE_ID变量，来确定当前程序是在主进程运行还是工作进程中运行。NODE_UNIQUE_ID实际上就是一个自增的数字，是工作进程的ID，后面会在创建工作进程相关代码中看到，这里就不多做解释了。
通过前面代码我们知道，如果在主进程中引用cluster，程序导出的是internal/cluster/primary.js这文件，因此我们先看看这个文件内部的一些实现。
// internal/cluster/primary.js
// ...
const EventEmitter = require('events');
const cluster = new EventEmitter();
// 下面这三个参数会在node内部功能实现的时候用到，之后我们看net源码的时候会用到这些参数
cluster.isWorker = false; // 是否是工作进程
cluster.isMaster = true; // 是否是主进程
cluster.isPrimary = true; // 是否是主进程
module.exports = cluster;
cluster.setupPrimary = function(options) {
const settings = {
args: ArrayPrototypeSlice(process.argv, 2),
exec: process.argv[1],
execArgv: process.execArgv,
silent: false,
...cluster.settings,
...options
};
cluster.settings = settings;
// ...
}
cluster.setupMaster = cluster.setupPrimary;
cluster.fork = function(env) {
cluster.setupPrimary();
const id = ++ids;
const workerProcess = createWorkerProcess(id, env);
}
const { fork } = require('child_process');
function createWorkerProcess(id, env) {
// 这里的NODE_UNIQUE_ID就是入口文件用来分辨当前进程类型用的
const workerEnv = { ...process.env, ...env, NODE_UNIQUE_ID: `${id}` };
// ...
return fork(cluster.settings.exec, cluster.settings.args, {
env: workerEnv,
// ...
});
}
cluster.fork用来新建一个工作进程，其内部使用child_process中的fork函数，来创建一个进程，创建的新进程默认会运行命令行中执行的入口文件（process.argv[1]），当然我们也可以执行luster.setupPrimary或者cluster.setupMaster并传入exec参数来修改工作进程执行的文件。
我们再来简单看下工作进程引用的internal/cluster/child.js文件：
// internal/cluster/child.js
const EventEmitter = require('events');
const cluster = new EventEmitter();
module.exports = cluster;
// 这里定义的就是一个工作进程，后续会用到这里的参数
cluster.isWorker = true;
cluster.isMaster = false;
cluster.isPrimary = false;
cluster._getServer = function(obj, options, cb) {
// ...
};
// ...
这里我们主要记住工作进程中的cluster有个_getServer函数，后续流程走到这个函数的时候，会详细看里面的代码。
接下来进入正题，看下net server listen函数：
// lib/net.js
Server.prototype.listen = function(...args) {
// ...
if (typeof options.port === 'number' || typeof options.port === 'string') {
// 如果是向最开始那种直接调用listen时直接传入一个端口，就会直接进入else，我们也主要看else中的逻辑
if (options.host) {
// ...
} else {
// listen(8080, () => {...})调用方式，将运行这条分支
listenInCluster(this, null, options.port | 0, 4, backlog, undefined, options.exclusive);
}
return this;
}
// ...
}
function listenInCluster(server, address, port, addressType, backlog, fd, exclusive, flags) {
// ...
// 这里就用到cluster初始时写入的isPrimary参数，当前如果在主进程isPrimary就为true，反之为false。主进程会直接去执行server._listen2函数，工作进程之后也会执行这个函数，等下一起看server._listen2内部的功能。
if (cluster.isPrimary || exclusive) {
server._listen2(address, port, addressType, backlog, fd, flags);
return;
}
// 后面的代码只有在工作进程中才会执行
const serverQuery = {
address: address,
port: port,
addressType: addressType,
fd: fd,
flags,
};
// 这里执行的是internal/cluster/child.js中的cluster._getServer，同时会传入listenOnPrimaryHandle这个回调函数，这个回调函数会在主进程添加端口监听，同时将工作进程绑定到对应的TCP服务后才会执行，里面工作就是对net server listen等函数进行hack。
cluster._getServer(server, serverQuery, listenOnPrimaryHandle);
function listenOnPrimaryHandle(err, handle) {
// ...
server._handle = handle;
server._listen2(address, port, addressType, backlog, fd, flags);
}
}
// 等工作进程执行这个函数的时候再一起讲
Server.prototype._listen2 = setupListenHandle;
function setupListenHandle(...) {
// ...
}
从上面代码中可以得知，主进程和工作进程中执行net server listen都会进入到一个setupListenHandle函数中。不过区别是，主进程是直接执行该函数，而工作进程需要先执行cluster._getServer函数，让主进程监听工作进程端口，同时对listen函数进行hack处理，然后再执行setupListenHandle函数。接下来我们看下cluster._getServer函数的内部实现。
// lib/internal/cluster/child.js
cluster._getServer = function(obj, options, cb) {
// ...
// 这个是工作进程第一次发送内部消息的内容。
// 注意这里act值为queryServer
const message = {
act: 'queryServer',
index,
data: null,
...options
};
// ...
// send函数内部使用IPC信道向工作进程发送内部消息。主进程在使用cluster.fork新建工作进程的时候，会让工作进程监听内部消息事件，下面会展示具体代码
// send调用传入的回调函数会被写入到lib/internal/cluster/utils.js文件中的callbacks map中，等后面要用的时候，再提取出来。
send(message, (reply, handle) => {
if (typeof obj._setServerData === 'function')
obj._setServerData(reply.data);
if (handle)
shared(reply, handle, indexesKey, index, cb);
else
// 这个函数内部会定义一个listen函数，用来hack net server listen函数
rr(reply, indexesKey, index, cb);
});
// ...
}
function send(message, cb) {
return sendHelper(process, message, null, cb);
}
// lib/internal/cluster/utils.js
// ...
const callbacks = new SafeMap();
let seq = 0;
function sendHelper(proc, message, handle, cb) {
message = { cmd: 'NODE_CLUSTER', ...message, seq };
if (typeof cb === 'function')
// 这里将传入的回调函数记录下来。
// 注意这里的key是递增数字
callbacks.set(seq, cb);
seq += 1;
// 利用IPC信道，给当前工作进程发送内部消息
return proc.send(message, handle);
}
// ...
工作进程中cluster._getServer函数执行，将生成一个回调函数，将这个回调函数存放起来，并且会使用IPC信道，向当前工作进程发送内部消息。主进程执行cluster.fork生成工作进程的时候，会在工作进程中注册internalMessage事件。接下来我们看下cluster.fork中与工作进程注册内部消息事件的代码。
// internal/cluster/primary.js
cluster.fork = function(env) {
// ...
// internal函数执行会返回一个接收message对象的回调函数。
// 可以先看下lib/internal/cluster/utils.js中的internal函数，了解内部的工作
worker.process.on('internalMessage', internal(worker, onmessage));
// ...
}
const methodMessageMapping = {
close,
exitedAfterDisconnect,
listening,
online,
queryServer,
};
// 第一次触发internalMessage执行的回调是这个函数。
// 此时message的act为queryServer
function onmessage(message, handle) {
// internal内部在执行onmessage时会将这个函数执行上下文绑定到工作进程的work上
const worker = this;
// 工作进程传入的
const fn = methodMessageMapping[message.act];
if (typeof fn === 'function')
fn(worker, message);
}
function queryServer(worker, message) {
// ...
}
// lib/internal/cluster/utils.js
// ...
const callbacks = new SafeMap();
function internal(worker, cb) {
return function onInternalMessage(message, handle) {
let fn = cb;
// 工作进程第一次发送内部消息：ack为undefined，callback为undefined，直接执行internal调用传入的onmessage函数，message函数只是用于解析消息的，实际会执行queryServer函数
// 工作进程第二次发送内部消息：主进程queryServer函数执行会用工作进程发送内部消息，并向message中添加ack参数，让message.ack=message.seq
if (message.ack !== undefined) {
const callback = callbacks.get(message.ack);
if (callback !== undefined) {
fn = callback;
callbacks.delete(message.ack);
}
}
ReflectApply(fn, worker, arguments);
};
}
工作进程第一次发送内部消息时，由于传入的message.ack（这里注意分清act和ack）为undefind，因此没办法直接拿到cluster._getServer中调用send写入的回调函数，因此只能先执行internal/cluster/primary.js中的queryServer函数。接下来看下queryServer函数内部逻辑。
// internal/cluster/primary.js
// hadles中存放的就是TCP服务器。
// 主进程在代替工作进程监听端口生成新的TCP服务器前，
// 需要先判断该服务器是否有创建，如果有，就直接复用之前的服务器，然后将工作进程绑定到相应的服务器上；如果没有，就新建一个TCP服务器，然后将工作进程绑定到新建的服务器上。
function queryServer(worker, message) {
// 这里key就是服务器的唯一标识
const key = `${message.address}:${message.port}:${message.addressType}:` +
`${message.fd}:${message.index}`;
// 从现存的服务器中查看是否有当前需要的服务器
let handle = handles.get(key);
// 如果没有需要的服务器，就新建一个
if (handle === undefined) {
// ...
// RoundRobinHandle构建函数中，会新建一个TCP服务器
let constructor = RoundRobinHandle;
handle = new constructor(key, address, message);
// 将这个服务器存放起来
handles.set(key, handle);
}
if (!handle.data)
handle.data = message.data;
// 可以先看下下面关于RoundRobinHandle构建函数的代码，了解内部机制
handle.add(worker, (errno, reply, handle) => {
const { data } = handles.get(key);
if (errno)
handles.delete(key);
// 这里会向工作进程中发送第二次内部消息。
// 这里只传了worker和message，没有传入handle和cb
send(worker, {
errno,
key,
ack: message.seq, // 注意这里增加了ack属性
data,
...reply
}, handle);
});
}
function send(worker, message, handle, cb) {
return sendHelper(worker.process, message, handle, cb);
}
// internal/cluster/round_robin_handle.js
function RoundRobinHandle(key, address, { port, fd, flags }) {
// ...
this.server = net.createServer(assert.fail);
if (fd >= 0)
this.server.listen({ fd });
else if (port >= 0) {
this.server.listen({
port,
host: address,
ipv6Only: Boolean(flags & constants.UV_TCP_IPV6ONLY),
});
} else
this.server.listen(address);
// 当服务处于监听状态，就会执行这个回调。
this.server.once('listening', () => {
this.handle = this.server._handle;
this.handle.onconnection = (err, handle) => this.distribute(err, handle);
this.server._handle = null;
// 注意：如果监听成功，就会将server删除
this.server = null;
});
}
RoundRobinHandle.prototype.add = function(worker, send) {
const done = () => {
if (this.handle.getsockname) {
// ...
send(null, { sockname: out }, null);
} else {
send(null, null, null); // UNIX socket.
}
// ...
};
// 如果在add执行前server就已经处于listening状态，this.server就会为null
if (this.server === null)
return done();
// 如果add执行后，server才处于listening，就会走到这里，始终都会执行add调用时传入的回调
this.server.once('listening', done);
}
在这一步，主进程替工作进程生成或者是获取了一个可用的TCP服务器，并将工作进程与相应的服务器绑定在一起（方便后续请求任务分配）。当工作进程绑定完成以后，就向工作进程中发送了第二次内部消息，接下来我们再次进入lib/internal/cluster/utils.js看看内部流程：
// lib/internal/cluster/utils.js
const callbacks = new SafeMap();
function internal(worker, cb) {
// 注意这里handle为undefined
return function onInternalMessage(message, handle) {
let fn = cb;
// 第二次工作进程内部消息执行的时候message.ack已经被赋值为message.seq
// 因此这次能够获取到之前lib/cluster.child.js cluster._getServer函数执行是调用send写入的回调函数
if (message.ack !== undefined) {
const callback = callbacks.get(message.ack);
if (callback !== undefined) {
fn = callback;
callbacks.delete(message.ack);
}
}
ReflectApply(fn, worker, arguments);
};
}
工作进程第二次接受到内部消息时，cluster._getServer函数执行是调用send写入的回调函数会被执行，接下来看下send写入的回调函数内容：
// lib/internal/cluster/child.js
send(message, (reply, handle) => {
// 此时handle为undefined，流程会直接运行rr函数
if (handle)
shared(reply, handle, indexesKey, index, cb);
else
// 这里的cb是lib/net.js在执行cluster._getServer时传入listenOnPrimaryHandle函数，后面会介绍他的工作。
rr(reply, indexesKey, index, cb);
});
function rr(message, indexesKey, index, cb) {
let key = message.key;
// 这里定义的listen用于hack net server.listen，在工作进程中执行listen，工作进程并不会真正去监听端口
function listen(backlog) {
return 0;
}
function close() {...}
function getsockname(out) {...}
const handle = { close, listen, ref: noop, unref: noop };
handles.set(key, handle);
// 执行传入的listenOnPrimaryHandle函数
cb(0, handle);
}
rr函数执行，会新建几个与net server中同名的函数，并通过handle传入listenOnPrimaryHandle函数。
// lib/net.js
function listenInCluster(...) {
cluster._getServer(server, serverQuery, listenOnPrimaryHandle);
// listenOnPrimaryHandle函数中将工作进程生成的server._handle对象替换成自定义的handle对象，后续server listen执行的就是server._handle中的listen函数，因此这里就完成了对工作进程中的listen函数hack
function listenOnPrimaryHandle(err, handle) {
// ...
// handle：{ listen: ..., close: ...., ... }
server._handle = handle;
server._listen2(address, port, addressType, backlog, fd, flags);
}
}
下面看下server._listen2函数执行内容
Server.prototype._listen2 = setupListenHandle;
function setupListenHandle(address, port, addressType, backlog, fd, flags) {
// 忽略，只要是从工作进程进来的，this._handle就是自己定义的对象内容
if (this._handle) {
debug('setupListenHandle: have a handle already');
} else {
// 主进程会进入这一层逻辑，会在这里生成一个服务器
// ...
rval = createServerHandle(address, port, addressType, fd, flags);
// ...
this._handle = rval;
}
const err = this._handle.listen(backlog || 511);
// ...
}
至此，工作进程端口监听相关的源码就看完了，现在差不多了解到工作进程中执行net server listen时，工作进程并不会真正去监听端口，端口监听工作始终会交给主进程来完成。主进程在接到工作进程发来的端口监听的时候，首先会判断是否有相同的服务器，如果有，就直接将工作进程绑定到对应的服务器上，这样就不会出现端口被占用的问题；如果没有对应的服务器，就生成一个新的服务。主进程接受到请求的时候，就会将请求任务分配给工作进程，如何分配，就需要看具体使用的哪种负载均衡了。
网友评论