Detaillierte Erklärung der JavaScript-Multithreading-Bibliothek Nexus.js

Zunächst, wenn Sie mit diesem Projekt nicht vertraut sind, wird empfohlen, zunächst eine Reihe von Artikeln zu lesen, die ich zuvor geschrieben habe. Wenn Sie diese nicht lesen möchten, brauchen Sie sich keine Sorgen zu machen. Dies wird auch auf diese Inhalte eingehen.

Lassen Sie uns anfangen.

Letztes Jahr habe ich mit der Implementierung von Nexus.js begonnen, das auf Webkit basiert/JavaScript-Kern-Dienstleistungs-Server-JavaScript-Bibliothek mit Multithreading. Einige Zeit habe ich es aufgegeben, dies zu tun, aufgrund einiger Gründe, die ich nicht kontrollieren kann, ich werde dies hier nicht diskutieren, hauptsächlich: Ich kann mich nicht lange beschäftigen.

Also, lassen Sie uns mit der Diskussion über die Architektur von Nexus beginnen und wie sie funktioniert.

Event-Loop

Keine Event-Loop

Es gibt einen Thread-Pool mit (lockfreien) Task-Objekten

Bei jedem Aufruf von setTimeout oder setImmediate oder beim Erstellen eines Promises wird der Task in die Task-Warteschlange eingereiht.

Bei jedem geplanten Task wählt der erste verfügbare Thread den Task aus und führt ihn aus.

Verarbeitet Promises auf CPU-Kernen. Der Aufruf von Promise.all() löst Promises parallel auf.

ES6

Unterstützung von async/await und wird empfohlen

Unterstützung von for await(...)

Unterstützung von Dekomposition

Unterstützung von async try/catch/finally

Modul

Unterstützt nicht CommonJS. (require(...) und module.exports)

Alle Module verwenden ES6des Imports/Export-Syntax

Unterstützung dynamischer Imports durch import('file-oder-packge).then(...)

Unterstützung von import.meta, z.B. import.meta.filename sowie import.meta.dirname usw.

Zusätzliche Funktionen: Unterstützung des direkten Imports aus URLs, z.B.:

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

EventEmitter

Nexus implementiert die auf Promise basierende Klasse EventEmitter

Der Ereignisverarbeitungsprozess wird auf allen Threads sortiert und parallel ausgeführt.

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('enwik',8');}}
  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(`Fertig in ${(Date.now() * startTime) / 1000} Sekunden!`);
 } catch (e) {
  console.error('Ein Fehler ist aufgetreten: ', e);
 }
}
start().catch(console.error);

TCP/UDP

Nexus.js stellt eine Acceptor-Klasse bereit, die für die Bindung von IP-Adressen verantwortlich ist/Port und Überwachung von Verbindungen

Bei jedem empfangenen Verbindungsanfrage wird das connection-Ereignis ausgelöst und ein Socket-Gerät bereitgestellt.

Jede Socket-Instanz ist bidirektionaler I/O-Gerät.

Sie können ReadableStream und WritableStream verwenden, um Sockets zu verwalten.

Schlichtester Beispiel: (Senden von "Hallo Welt" an den Client)

const acceptor = new Nexus.Net.TCP.Acceptor();
let count = 0;
acceptor.on('connection', (socket, endpoint) => {
 const connId = count++;
 console.log(`Verbindung #${connId} von ${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 bietet die Klasse Nexus.Net.HTTP.Server, die im Wesentlichen von TCPAcceptor abgeleitet ist

Einige grundlegende Schnittstellen

Wenn der Server die grundlegende Analyse der Http-Header der eingehenden Verbindung abgeschlossen hat/Bei der Überprüfung wird der Verbindung und derselben Information ein connection-Ereignis ausgelöst

Jede Verbindungsinstanz hat eine Anfrage- und eine Response-Objekt. Dies sind die Eingaben/Ausgabegerät。

Sie können ReadableStream und WritableStream konstruieren, um die Anfrage zu steuern/response。

Wenn Sie eine Response-Objekt über einen Pipe-Verbindung verbinden, wird der Eingabestrom im Modus der Blöcke-Kodierung verwendet. Andernfalls können Sie response.write() verwenden, um einen regulären String zu schreiben.

Komplexes Beispiel: (Basis-HTTP-Server und Blockkodierung, Details werden weggelassen)}

....
/**
 * Erstellt einen Eingabestrom von einem Pfad.
 * @param path
 * @returns {Promise<ReadableStream>}
 */
async function createInputStream(path) {
 if (path.startsWith('/')) // Wenn es mit '/', ignorere es.
  path = path.substr(1);
 if (path.startsWith('.')) // Wenn es mit '.', abweisen.
  throw new NotFoundError(path);
 if (path === '/|| !path) // Wenn es leer ist, setze es auf index.html.
  path = 'index.html';
 /**
  * `import.meta.dirname` und `import.meta.filename` ersetzen die alten CommonJS `__dirname` und `__filename`.
  */
 const filePath = Nexus.FileSystem.join(import.meta.dirname, 'server_root', path);
 try {
  // Stat den Zielpfad.
  const {type} = await Nexus.FileSystem.stat(filePath);
  if (type === Nexus.FileSystem.FileType.Directory) // Falls es ein Verzeichnis ist, gibt es 'index.html' zurück
   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 wird unterstützt, warum nutzen Sie es nicht?
 const { request, response } = connection;
 // Verarbeiten Sie die URL-Teile.
 const { path } = parseURL(request.url);
 // Hier speichern wir alle Fehler, die während der Verbindung auftreten.
 const errors = [];
 // inStream ist unsere lesbare Stream-Dateiquelle, outStream ist unsere Antwort (Gerät) in einem schreibbaren Stream verpackt.
 let inStream, outStream;
 try {
  // Protokollieren Sie die Anfrage.
  console.log(`> #${FgCyan + connId + Reset} ${Bright + peer.address}:${peer.port + Reset} ${
   FgGreen + request.method + Reset} "${FgYellow}${path}${Reset}"`, Reset);
  // Setzen Sie den 'Server'-Header.
  response.set('Server', `nexus.js/0.1.1`);
  // Erstellen Sie unseren Eingabestrom.
  inStream = await createInputStream(path);
  // Erstellen Sie unseren Ausgabestrom.
  outStream = new Nexus.IO.WritableStream(response);
  // Hooken Sie alle `error`-Events, fügen Sie alle Fehler unserer `errors`-Array hinzu.
  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); });
  // Setzen Sie den Inhaltstyp und den Anfragestatus.
  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, it has no real effect with keep-active 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;
/**
 * 绑定选定的地址和端口。
 */
server.bind(ip, port, portReuse);
/**
 * 开始监听请求。
 */
server.listen(maxConcurrentConnections);
/**
 * 愉快的流媒体播放！
 */
console.log(FgGreen + `Nexus.js HTTP服务器监听于${ip}:${port}` + 重置);

基准

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

这里是上述Http服务器的当前基准，有100个并发连接和总共10000个请求：

这是ApacheBench，版本 2.3 <$修订： 1796539 $>
版权 1996 Adam Twiss, Zeus Technology Ltd, http：//www.zeustech.net/
许可给Apache软件基金会，http：//www.apache.org/
基准测试localhost（请耐心等待）.....完成
服务器软件：    nexus.js/0.1.1
服务器主机名：    localhost
服务器端口：      3000
文档路径：     /
文档长度：    8673 字节
并发级别：   100
测试所用时间：  9.991 秒
完成的请求：   10000
失败的请求：    0
总传输量：   87880000字节
传输的HTML：    86730000字节
每秒请求数：  1000.94 [#/秒] (平均)
请求时间：    99.906 [ms] (平均)
请求时间：    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
Fehler    :
bogomips  : 6784.18
clflush-Größe  : 64
Cache-Ausrichtung : 64
Adressgrößen  : 39 bits physisch, 48 bits virtuell
Energieverwaltung:

Ich habe versucht1000 parallele Anfragen, aber APacheBench aufgrund vieler geöffneter Sockets abgelaufen. Ich habe httperf versucht, hier sind die Ergebnisse:

voodooattack@voodooattack:~$ httperf --Port=3000 --Anzahl-Verbindungen=10000 --Rate=1000
httperf --Client=0/1 --Server=localhost --Port=3000 --URI=/ --Rate=1000 --gesendet-Puffer=4096 --empfangen-Puffer=16384 --Anzahl-Verbindungen=10000 --Anzahl-Aufrufe=1
httperf: Warnung: Dateiöffnungsbeschränkung > FD_SETSIZE; maximale Anzahl offener Dateien auf FD_SETSIZE begrenzt
Maximale Länge des Verbindungsbursts: 262
Insgesamt: Verbindungen 9779 Anfragen 9779 Antworten 9779 Test-Dauer 10.029 s
Verbindungsrate: 975.1 verbindungen/s (1.0 ms/verbindungen, <=1022 gleichzeitige Verbindungen)
Verbindungszeit [ms]: min 0.5 avg 337.9 max 7191.8 Mittelwert 79.5 stddev 848.1
Verbindungszeit [ms]: verbinden 207.3
Verbindungszeit [Antworten]}/verbindung]: 1.000
Anfrage rate: 975.1 anfr/s (1.0 ms/anfr)
Anfragegröße [B]: 62.0
Antwortrate [Antworten/s]: min 903.5 avg 974.6 max 1045.7 stddev 100.5 (2 samples)
Antwortzeit [ms]: Antwort 129.5 Übertragung 1.1
Antwortgröße [B]: Kopfzeile 890.0 Inhalt 8660.0 Fußnote 2.0 (gesamt 8751.0)
Antwortstatus: 1xx=0 2xx=9779 3xx=0 4xx=0 5xx=0
CPU-Zeit [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)
Fehler: total 221 client-timo 0 socket-timo 0 connrefused 0 connreset 0
Fehler: fd-unavail 221 addrunavail 0 ftab-voll 0 andere 0

Wie Sie sehen können, funktioniert es immer noch. Trotz des Drucks wird einige Verbindungen abgebrochen. Ich untersuche immer noch die Ursachen dieses Problems.

Das ist der gesamte Inhalt dieses Artikels über Nexus.js. Wenn Sie Fragen haben, können Sie unten einen Kommentar hinterlassen und die Unterstützung des呐喊教程 danken.

Erklärung: Der Inhalt dieses Artikels wurde aus dem Internet übernommen und gehört dem Urheberrechtsinhaber. Der Inhalt wurde von Internetnutzern freiwillig beigesteuert und hochgeladen. Diese Website besitzt keine Eigentumsrechte und hat den Inhalt nicht manuell bearbeitet. Sie übernimmt auch keine rechtlichen Verantwortlichkeiten. Wenn Sie Inhalte finden, die urheberrechtlich geschützt sind, freuen wir uns über eine E-Mail an: notice#oldtoolbag.com (bei der E-Mail senden Sie bitte # durch @ ersetzen und beschreiben Sie den Vorfall, sowie die betreffenden Beweise. Sobald die Informationen überprüft wurden, wird die Website den fraglichen Inhalt sofort löschen, der urheberrechtlich geschützt ist.)