JavaScript多线程运行库Nexus.js详解

首先,如果你不熟悉这个项目,建议先阅读之前写的一系列文章。如果你不想阅读这些,不用担心。这里面也会涉及到那些内容。

现在,让我们开始吧。

去年,我开始实现Nexus.js,这是一个基于Webkit/JavaScript内核的多线程服务端JavaScript运行库。有一段时间我放弃了做这件事,由于一些我无法控制的原因,我不打算在这里讨论,主要是:我无法让自己长时间工作。

所以,让我们从讨论Nexus的架构开始,以及它是如何工作的。

事件循环

没有事件循环

有一个带有(无锁)任务对象的线程池

每次调用setTimeout或setImmediate或创建一个Promise时,任务就排队到任务队列钟。

每当计划任务时,第一个可用的线程将选择任务并执行它。

在CPU内核上处理Promise。对Promise.all()的调用将并行的解决Promise。

ES6

支持async/await,并且推荐使用

支持for await(...)

支持解构

支持async try/catch/finally

模块

不支持CommonJS。(require(...)和module.exports)

所有模块使用ES6的import/export语法

支持动态导入通过import('file-or-packge').then(...)

支持import.meta,例如:import.meta.filename以及import.meta.dirname等等

附加功能:支持直接从URL中导入,例如:

import { h } from 'https://unpkg.com/preact/dist/preact.esm.js';

EventEmitter

Nexus实现了基于Promise的EventEmitter类

事件处理程序在所有线程上排序,并将并行处理执行。

EventEmitter.emit(...)的返回值是一个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。

所有输入/输出原语都实现了EventEmitter类

要使用Device,你需要在Device之上创建一个ReadableStream或WritableStream

要操作数据,可以将Filters添加到ReadableStream或WritableStream中。

最后,使用source.pipe(...destinationStreams),然后等待source.resume()来处理数据。

所有的输入/输出操作都是使用ArrayBuffer对象完成的。

Filter试了了process(buffer)方法来处理数据。

例如:使用2个独立的输出文件将UTF-8转换为UTF6。

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/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类,该类基本上继承了TCPAcceptor

一些基础接口

当服务器端完成了对传入连接的基本的Http头的解析/校验时,将使用连接和同样的信息触发connection事件

每一个连接实例都又一个request和一个response对象。这些是输入/输出设备。

你可以构造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);

基准

我想我已经涵盖了到目前为止所实现的一切。那么现在我们来谈谈性能。

这里是上诉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学习知识的全部内容,大家有问题可以在下方留言讨论,感谢对我们的支持。

(0)

相关推荐

  • nexus服务器搭建的详细步骤

    一.搭建nexus服务的意义 作为内网的统一代理,团队合作开发时不用每人都去外网下载一次: 解决部分变态公司内网管制无法访问外网的问题,选一台有外网权限的机器搭建服务器,其余人直接内网访问此服务: 团队合作开发时解决maven项目依赖问题: 二.安装nexus # 下载安装包: wget https://sonatype-download.global.ssl.fastly.net/nexus/oss/nexus-2.14.4-03-bundle.tar.gz # 解压 mkdir /data/

  • 浅谈android nexus私服的使用

    先唠叨几句啊,由于公司私服已经搭好了我就不费那劲琢磨搭建私服的事了,直接开撸上传lib.下图是我放组件库的地方,本来想一个module拉出一个项目来维护,后来想想实在是麻烦,同时维护多个库就要down多个库下来,于是就统一管理了.当然统一维护module多了自然会影响编译的速度,可以在settings.gradle 里将不编译的module注释掉就可以了.现在就有图片选择.表情.输入框.星星.视频录制.裁切等七七八八个库,至于后面组件库多了可能会分类按类型拉出几个项目来管理,要不版本控制也是个麻

  • 如何使用nexus在局域网内搭建maven私服及idea的使用

    使用Maven进行项目的声明周期管理,方便对项目依赖和插件进行管理,同时便于项目的统一管理,提升项目开发效率.但是实际工作中会有以下问题: 1  因为安全考虑,有些公司不给开发人员提供外网.因此不能使用maven访问远程的仓库地址. 2  大家同时上网,导致网速比较慢,maven下载构件效率比较低. 3  由于版权的问题,有些第三方jar包无法从远程仓库下载. 4  有些jar包是公司内部使用的,自然也无法从远程maven仓库下载. 怎么办呢? 使用Nexus搭建私服 所谓私服,是一种特殊的远程

  • Maven入门之使用Nexus搭建Maven私服及上传下载jar包

    一. 私服搭建及配置 1 . 私服简介 私服是架设在局域网的一种特殊的远程仓库,目的是代理远程仓库及部署第三方构件.有了私服之后,当 Maven 需要下载构件时,直接请求私服,私服上存在则下载到本地仓库:否则,私服请求外部的远程仓库,将构件下载到私服,再提供给本地仓库下载. 我们可以使用专门的 Maven 仓库管理软件来搭建私服,比如:Apache Archiva,Artifactory,Sonatype Nexus.这里我们使用 Sonatype Nexus. 2 . 安装Nexus 2 .

  • CentOS7 Nexus安装步骤详细介绍

    CentOS7 Nexus安装 CentOS7 Download Nexus安装下载nexus-2.11.2-03-bundle.tar.gz Install 安装 上传RPM文件到/tmp目录下 mkdir /opt/nexus cd /opt/nexus tar -zxvf /tmp/nexus-2.11.2-03-bundle.tar.gz ln -sf /opt/nexus/nexus-2.11.2-03/opt/nexus/latest ln -sf /opt/nexus/latest

  • 如何通过网页方式将jar包上传到nexus的方法步骤

    在使用Maven管理项目时,如何将jar包上传到私有仓库,下面简单介绍一下上传方法. 环境:nexus私有库 步骤如下: 1.登陆已经安装好的nexus私有仓库,如图: 2.点击左边菜单"Repositories",选择右边列表"3rd party" 3.点击"3rd party",选择artifact Upload,如下图: 4.在artifact Upload界面显示的有GAV Parameter,groupId,artifactId,Ver

  • Maven nexus 安装nexus私服出现的问题和解决办法

    1. 在win10中安装nexus时提示:wrapper | OpenSCManager failed - 拒绝访问. (0x5) 主要是没有权限.需要以管理员的身份运行 如果你是直接点击 start-nexus.bat 来启动服务的话: 如果你是 通过 cmd窗口来执行的话,需要设置命令提示符窗口的权限来执行 2. wrapper | The nexus-webapp service is not installed - The specified service does not exist

  • Nexus使用Api进行操作

    Nexus提供了RestApi,但是一部分Api仍然需要结合Groovy等进行操作,在3.3及其以后的版本上进行了强化,但是和普通的RestApi相比仍有一些"使用上的特性".这篇文章以创建raw的repository为例,来确认一下相关的使用方法. 前提准备 准备可运行的Nexus,这里的版本为3.2.1 Api机制 这篇文章介绍的是Nexus3推出之初的Api的设计思路,主要是通过结合groovy脚本来完成相关的功能,简单的来说需要用户自行提供groovy脚本,nexus提供上传和

  • 详解Maven私服Nexus的安装与使用

    本文介绍了详解Maven私服Nexus的安装与使用,分享给大家,具体如下: 1.安装 1.1 安装docker并加速 yum update && yum install docker sudo mkdir -p /etc/docker sudo tee /etc/docker/daemon.json <<-'EOF' { "registry-mirrors": ["https://y7u9p3i0.mirror.aliyuncs.com"

  • 使用Nexus添加jar包到私服里的方法步骤

    为什么要搭建nexus私服,原因很简单,有些公司都不提供外网给项目组人员,因此就不能使用maven访问远程的仓库地址,所以很有必要在局域网里找一台有外网权限的机器,搭建nexus私服,然后开发人员连到这台私服上,这样的话就可以通过这台搭建了nexus私服的电脑访问maven的远程仓库. 1.首先确定我们的环境安装好maven,jdk等必须的环境 2.这些都准备好之后,去下载最新版本的nexus    下载地址:http://www.sonatype.org/nexus/go 我本地安装的是 ne

随机推荐