本文详细介绍Nexus.js，一个基于Webkit内核的多线程服务端JavaScript运行库。内容涵盖其事件循环机制、ES6特性支持（如async/await）、模块系统、基于Promise的EventEmitter、以及Device/Filter/Stream组成的I/O模型。此外，还解析了TCP/HTTP服务器的实现及性能基准测试结果，为构建高性能网络应用提供参考。
若对Nexus.js项目尚不熟悉，建议先阅读作者此前撰写的系列文章以获取背景知识；即便跳过前作，本文也将涵盖相关要点。
接下来，我们正式进入主题。
Nexus.js架构与事件循环机制
去年，我开始开发Nexus.js，这是一个基于Webkit和JavaScript内核构建的多线程服务端JavaScript运行库。期间因某些不可控因素曾暂停开发，具体原因在此不予赘述，主要原因在于难以长期维持高强度的工作状态。
现在，让我们从Nexus的架构设计及其工作原理开始讨论。
事件循环机制
无事件循环设计
系统包含一个由无锁任务对象组成的线程池。
每次调用setTimeout、setImmediate或创建Promise时，相关任务会被加入任务队列。
一旦任务被调度，线程池中首个可用的线程将选取并执行该任务。
Promise的处理直接在CPU内核上并行进行，调用Promise.all()可实现Promise的并行解决。
ES6语言特性与模块系统支持
ES6语言特性支持
全面支持async/await语法，且官方推荐使用该方式。
支持for await(...)循环结构。
支持对象解构赋值。
支持async环境下的try/catch/finally异常处理机制。
模块系统
不支持CommonJS规范（即不支持require(...)和module.exports）。
所有模块均强制使用ES6标准的import/export语法。
支持通过import('file-or-package').then(...)进行动态导入。
支持import.meta元数据，例如import.meta.filename以及import.meta.dirname等属性。
附加功能：支持直接从URL导入资源，具体示例如下：
import { h } from 'https://unpkg.com/preact/dist/preact.esm.js';基于Promise的EventEmitter实现
EventEmitter事件发射器
Nexus实现了基于Promise的EventEmitter类。
事件处理程序在所有可用线程上按序排列，并并行执行处理逻辑。
EventEmitter.emit(...)方法的返回值是一个Promise，该Promise解析后返回由所有事件处理器返回值构成的数组。
具体示例代码如下：
class EmitterTest extends Nexus.EventEmitter {
constructor() {
super();
for(let i = 0; i < 4; i++)
this.on('test', value => { console.log(`fired test ${i}!`); console.inspect(value); });
for(let i = 0; i < 4; i++)
this.on('returns-a-value', v => `${v + i}`);
}
}
const test = new EmitterTest();
async function start() {
await test.emit('test', { payload: 'test 1' });
console.log('first test done!');
await test.emit('test', { payload: 'test 2' });
console.log('second test done!');
const values = await test.emit('returns-a-value', 10);
console.log('third test done, returned values are:'); console.inspect(values);
}
start().catch(console.error);I/O模型：Device、Filter与Stream
输入/输出（I/O）处理
所有的输入/输出操作均通过三个核心原语完成：Device（设备）、Filter（过滤器）和Stream（流）。
所有I/O原语均实现了EventEmitter接口。
使用Device时，需在其上层构建ReadableStream（可读流）或WritableStream（可写流）。
数据处理方面，可将Filters添加到ReadableStream或WritableStream中。
最终，通过source.pipe(...destinationStreams)连接源与目标流，并调用source.resume()以开始处理数据。
所有I/O操作均基于ArrayBuffer对象进行。
Filter通过process(buffer)方法处理数据缓冲区。
示例：利用2个独立的输出文件将UTF-8编码转换为UTF-16LE（注：原文UTF6疑为笔误，代码中显示为UTF-16LE）。
const startTime = Date.now();
try {
const device = new Nexus.IO.FilePushDevice('enwik8');
const stream = new Nexus.IO.ReadableStream(device);
stream.pushFilter(new Nexus.IO.EncodingConversionFilter("UTF-8", "UTF-16LE"));
const wstreams = [0,1,2,3]
.map(i => new Nexus.IO.WritableStream(new Nexus.IO.FileSinkDevice('enwik16-' + i)));
console.log('piping...');
stream.pipe(...wstreams);
console.log('streaming...');
await stream.resume();
await stream.close();
await Promise.all(wstreams.map(stream => stream.close()));
console.log(`finished in ${(Date.now() * startTime) / 1000} seconds!`);
} catch (e) {
console.error('An error occurred: ', e);
}
}
start().catch(console.error);TCP/HTTP网络通信与服务器实现
TCP/UDP网络通信
Nexus.js提供了Acceptor类，负责绑定IP地址/端口并监听连接请求。
每当收到连接请求，将触发connection事件，并提供一个Socket设备实例。
每个Socket实例均为全双工I/O设备。
可通过ReadableStream和WritableStream对Socket进行操作。
最基础示例：向客户端发送“Hello World”。
const acceptor = new Nexus.Net.TCP.Acceptor();
let count = 0;
acceptor.on('connection', (socket, endpoint) => {
const connId = count++;
console.log(`connection #${connId} from ${endpoint.address}:${endpoint.port}`);
const rstream = new Nexus.IO.ReadableStream(socket);
const wstream = new Nexus.IO.WritableStream(socket);
const buffer = new Uint8Array(13);
const message = 'Hello World!\n';
for(let i = 0; i < 13; i++)
buffer[i] = message.charCodeAt(i);
rstream.pushFilter(new Nexus.IO.UTF8StringFilter());
rstream.on('data', buffer => console.log(`got message: ${buffer}`));
rstream.resume().catch(e => console.log(`client #${connId} at ${endpoint.address}:${endpoint.port} disconnected!`));
console.log(`sending greeting to #${connId}!`);
wstream.write(buffer);
});
acceptor.bind('127.0.0.1', 10000);
acceptor.listen();
console.log('server ready');HTTP协议支持
Nexus提供了Nexus.Net.HTTP.Server类，该类基本继承自TCP Acceptor。
基础接口说明
当服务器完成对传入连接的基本HTTP头解析与校验后，将触发connection事件，并传递连接对象及相关信息。
每个连接实例包含request（请求）和response（响应）对象，二者均为I/O设备。
可通过构造ReadableStream和WritableStream来操纵request/response对象。
若将流管道连接至Response对象，输入流将采用分块编码模式；否则，可使用response.write()写入常规字符串。
复杂示例：基本的HTTP服务器实现（含分块编码，细节省略）。
....
/**
* Creates an input stream from a path.
* @param path
* @returns {Promise<ReadableStream>}
*/
async function createInputStream(path) {
if (path.startsWith('/')) // If it starts with '/', omit it.
path = path.substr(1);
if (path.startsWith('.')) // If it starts with '.', reject it.
throw new NotFoundError(path);
if (path === '/' || !path) // If it's empty, set to index.html.
path = 'index.html';
/**
* `import.meta.dirname` and `import.meta.filename` replace the old CommonJS `__dirname` and `__filename`.
*/
const filePath = Nexus.FileSystem.join(import.meta.dirname, 'server_root', path);
try {
// Stat the target path.
const {type} = await Nexus.FileSystem.stat(filePath);
if (type === Nexus.FileSystem.FileType.Directory) // If it's a directory, return its 'index.html'
return createInputStream(Nexus.FileSystem.join(filePath, 'index.html'));
else if (type === Nexus.FileSystem.FileType.Unknown || type === Nexus.FileSystem.FileType.NotFound)
// If it's not found, throw NotFound.
throw new NotFoundError(path);
} catch(e) {
if (e.code)
throw e;
throw new NotFoundError(path);
}
try {
// First, we create a device.
const fileDevice = new Nexus.IO.FilePushDevice(filePath);
// Then we return a new ReadableStream created using our source device.
return new Nexus.IO.ReadableStream(fileDevice);
} catch(e) {
throw new InternalServerError(e.message);
}
}
/**
* Connections counter.
*/
let connections = 0;
/**
* Create a new HTTP server.
* @type {Nexus.Net.HTTP.Server}
*/
const server = new Nexus.Net.HTTP.Server();
// A server error means an error occurred while the server was listening to connections.
// We can mostly ignore such errors, we display them anyway.
server.on('error', e => {
console.error(FgRed + Bright + 'Server Error: ' + e.message + '\n' + e.stack, Reset);
});
/**
* Listen to connections.
*/
server.on('connection', async (connection, peer) => {
// Start with a connection ID of 0, increment with every new connection.
const connId = connections++;
// Record the start time for this connection.
const startTime = Date.now();
// Destructuring is supported, why not use it?
const { request, response } = connection;
// Parse the URL parts.
const { path } = parseURL(request.url);
// Here we'll store any errors that occur during the connection.
const errors = [];
// inStream is our ReadableStream file source, outStream is our response (device) wrapped in a WritableStream.
let inStream, outStream;
try {
// Log the request.
console.log(`> #${FgCyan + connId + Reset} ${Bright + peer.address}:${peer.port + Reset} ${
FgGreen + request.method + Reset} "${FgYellow}${path}${Reset}"`, Reset);
// Set the 'Server' header.
response.set('Server', `nexus.js/0.1.1`);
// Create our input stream.
inStream = await createInputStream(path);
// Create our output stream.
outStream = new Nexus.IO.WritableStream(response);
// Hook all `error` events, add any errors to our `errors` array.
inStream.on('error', e => { errors.push(e); });
request.on('error', e => { errors.push(e); });
response.on('error', e => { errors.push(e); });
outStream.on('error', e => { errors.push(e); });
// Set content type and request status.
response
.set('Content-Type', mimeType(path))
.status(200);
// Hook input to output(s).
const disconnect = inStream.pipe(outStream);
try {
// Resume our file stream, this causes the stream to switch to HTTP chunked encoding.
// This will return a promise that will only resolve after the last byte (HTTP chunk) is written.
await inStream.resume();
} catch (e) {
// Capture any errors that happen during the streaming.
errors.push(e);
}
// Disconnect all the callbacks created by `.pipe()`.
return disconnect();
} catch(e) {
// If an error occurred, push it to the array.
errors.push(e);
// Set the content type, status, and write a basic message.
response
.set('Content-Type', 'text/plain')
.status(e.code || 500)
.send(e.message || 'An error has occurred.');
} finally {
// Close the streams manually. This is important because we may run out of file handles otherwise.
if (inStream)
await inStream.close();
if (outStream)
await outStream.close();
// Close the connection, has no real effect with keep-alive connections.
await connection.close();
// Grab the response's status.
let status = response.status();
// Determine what colour to output to the terminal.
const statusColors = {
'200': Bright + FgGreen, // Green for 200 (OK),
'404': Bright + FgYellow, // Yellow for 404 (Not Found)
'500': Bright + FgRed // Red for 500 (Internal Server Error)
};
let statusColor = statusColors[status];
if (statusColor)
status = statusColor + status + Reset;
// Log the connection (and time to complete) to the console.
console.log(`< #${FgCyan + connId + Reset} ${Bright + peer.address}:${peer.port + Reset} ${
FgGreen + request.method + Reset} "${FgYellow}${path}${Reset}" ${status} ${(Date.now() * startTime)}ms` +
(errors.length ? " " + FgRed + Bright + errors.map(error => error.message).join(', ') + Reset : Reset));
}
});
/**
* IP and port to listen on.
*/
const ip = '0.0.0.0', port = 3000;
/**
* Whether or not to set the `reuse` flag. (optional, default=false)
*/
const portReuse = true;
/**
* Maximum allowed concurrent connections. Default is 128 on my system. (optional, system specific)
* @type {number}
*/
const maxConcurrentConnections = 1000;
/**
* Bind the selected address and port.
*/
server.bind(ip, port, portReuse);
/**
* Start listening to requests.
*/
server.listen(maxConcurrentConnections);
/**
* Happy streaming!
*/
console.log(FgGreen + `Nexus.js HTTP server listening at ${ip}:${port}` + Reset);性能基准测试与结果分析
性能基准测试
至此已涵盖Nexus.js目前实现的所有功能，接下来讨论其性能表现。
以下是上述HTTP服务器的当前基准测试结果，配置为100个并发连接和总计10000个请求：
This is ApacheBench, Version 2.3 <$Revision: 1796539 $>
Copyright 1996 Adam Twiss, Zeus Technology Ltd, http://www.zeustech.net/
Licensed to The Apache Software Foundation, http://www.apache.org/
Benchmarking localhost (be patient).....done
Server Software: nexus.js/0.1.1
Server Hostname: localhost
Server Port: 3000
Document Path: /
Document Length: 8673 bytes
Concurrency Level: 100
Time taken for tests: 9.991 seconds
Complete requests: 10000
Failed requests: 0
Total transferred: 87880000 bytes
HTML transferred: 86730000 bytes
Requests per second: 1000.94 [#/sec] (mean)
Time per request: 99.906 [ms] (mean)
Time per request: 0.999 [ms] (mean, across all concurrent requests)
Transfer rate: 8590.14 [Kbytes/sec] received
Connection Times (ms)
min mean[+/-sd] median max
Connect: 0 0 0.1 0 1
Processing: 6 99 36.6 84 464
Waiting: 5 99 36.4 84 463
Total: 6 100 36.6 84 464
Percentage of the requests served within a certain time (ms)
50% 84
66% 97
75% 105
80% 112
90% 134
95% 188
98% 233
99% 238
100% 464 (longest request)测试结果显示每秒处理1000个请求。该测试在一台老旧的i7处理器上运行，同时启动了基准测试软件、占用5G内存的IDE以及服务器本身。
voodooattack@voodooattack:~$ cat /proc/cpuinfo
processor : 0
vendor_id : GenuineIntel
cpu family : 6
model : 60
model name : Intel(R) Core(TM) i7-4770 CPU @ 3.40GHz
stepping : 3
microcode : 0x22
cpu MHz : 3392.093
cache size : 8192 KB
physical id : 0
siblings : 8
core id : 0
cpu cores : 4
apicid : 0
initial apicid : 0
fpu : yes
fpu_exception : yes
cpuid level : 13
wp : yes
flags : fpu vme de pse tsc msr pae mce cx8 apic sep mtrr pge mca cmov pat pse36 clflush dts acpi mmx fxsr sse sse2 ss ht tm pbe syscall nx pdpe1gb rdtscp lm constant_tsc arch_perfmon pebs bts rep_good nopl xtopology nonstop_tsc cpuid aperfmperf pni pclmulqdq dtes64 monitor ds_cpl vmx smx est tm2 ssse3 sdbg fma cx16 xtpr pdcm pcid sse4_1 sse4_2 x2apic movbe popcnt tsc_deadline_timer aes xsave avx f16c rdrand lahf_lm abm cpuid_fault tpr_shadow vnmi flexpriority ept vpid fsgsbase tsc_adjust bmi1 avx2 smep bmi2 erms invpcid xsaveopt dtherm ida arat pln pts
bugs :
bogomips : 6784.18
clflush size : 64
cache_alignment : 64
address sizes : 39 bits physical, 48 bits virtual
power management:尝试进行1000个并发请求时，ApacheBench因打开过多套接字而超时。改用httperf进行测试，结果如下：
voodooattack@voodooattack:~$ httperf --port=3000 --num-conns=10000 --rate=1000
httperf --client=0/1 --server=localhost --port=3000 --uri=/ --rate=1000 --send-buffer=4096 --recv-buffer=16384 --num-conns=10000 --num-calls=1
httperf: warning: open file limit > FD_SETSIZE; limiting max. # of open files to FD_SETSIZE
Maximum connect burst length: 262
Total: connections 9779 requests 9779 replies 9779 test-duration 10.029 s
Connection rate: 975.1 conn/s (1.0 ms/conn, <=1022 concurrent connections)
Connection time [ms]: min 0.5 avg 337.9 max 7191.8 median 79.5 stddev 848.1
Connection time [ms]: connect 207.3
Connection length [replies/conn]: 1.000
Request rate: 975.1 req/s (1.0 ms/req)
Request size [B]: 62.0
Reply rate [replies/s]: min 903.5 avg 974.6 max 1045.7 stddev 100.5 (2 samples)
Reply time [ms]: response 129.5 transfer 1.1
Reply size [B]: header 89.0 content 8660.0 footer 2.0 (total 8751.0)
Reply status: 1xx=0 2xx=9779 3xx=0 4xx=0 5xx=0
CPU time [s]: user 0.35 system 9.67 (user 3.5% system 96.4% total 99.9%)
Net I/O: 8389.9 KB/s (68.7*10^6 bps)
Errors: total 221 client-timo 0 socket-timo 0 connrefused 0 connreset 0
Errors: fd-unavail 221 addrunavail 0 ftab-full 0 other 0如结果所示，系统仍能正常工作，尽管在高压力下部分连接会出现超时。目前仍在排查导致此问题的具体原因。
以上即为关于Nexus.js技术详解的全部内容，如有疑问欢迎在下方留言讨论，感谢对本站的支持。