توضيح مكتبة Nexus.js لتشغيل متعدد الأنماط في JavaScript

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

现在，让我们开始吧。

去年，我开始实现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('test الأول تم!');
 await test.emit('test', { payload: 'test 2' });
 console.log('test الثاني تم!');
 const values = await test.emit('returns-a-value', 10);
 console.log('test الثالث تم، القيم المنقولة هي:'); console.inspect(values);
}
start().catch(console.error);

I/O

تم إكمال جميع الإدخال/الإخراج من خلال ثلاثة كلمات أساسية: Device، Filter و Stream.

تم تنفيذ جميع الكلمات الأساسية الإدخال/الإخراج لفئة EventEmitter.

للإستخدام Device، يجب عليك إنشاء ReadableStream أو WritableStream على Device.

للعمل على البيانات، يمكنك إضافة Filters إلى ReadableStream أو WritableStream.

في النهاية، استخدم source.pipe(...destinationStreams) ثم انتظر source.resume() لمعالجة البيانات.

جميع العمليات الإدخال/الإخراج تتم باستخدام عناصر ArrayBuffer.

Filter جرب process(buffer) لمعالجة البيانات.

مثالًا: استخدم ملفين م输出تقديمي مستقلي لتحويل 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!`);
 }
  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';
 للمحصول على i < 13;
  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

واجهات أساسية

عندما يكمل الخادم تحليل/تحقق الرأس الأساسي لاتصال المستلم، سيتم إطلاق حدث الاتصال باستخدام نفس المعلومات والاتصال

كل نموذج اتصال يحتوي على هدف طلب وهدف رد. هذه هي أجهزة الإدخال/الخروج.

يمكنك بناء تدفق قابل للقراءة وتدفق قابل للكتابة للتعامل مع الطلب/الرد.

إذا كنت تقوم بربط مجرى إلى هدف الرد، سيستخدم تدفق الإدخال نمط التكديس. في غير ذلك، يمكنك استخدام response.write() لتحرير سلسلة عادية.

مثال معقد: (خادم HTTP الأساسي مع التكديس، التفاصيل مكتسبة)

....
/**
 * يخلق تدفق إدخال من مسار.
 * @param path
 * @returns {Promise<ReadableStream>}
 */
async function createInputStream(path) {
 if (path.startsWith('/')) // إذا بدأ بـ '/', تُتجاهله.
  path = path.substr(1);
 if (path.startsWith('.')) // إذا بدأ بالنقطة، رفضه.
  throw new NotFoundError(الطريق);
 if (path === '/' || !path) // إذا كان فارغًا، أعد إلى index.html.
  path = 'index.html';
 /**
  * `import.meta.dirname` و `import.meta.filename` تُستبدل بالقديم CommonJS `__dirname` و `__filename`.
  */
 const filePath = Nexus.FileSystem.join(import.meta.dirname, 'server_root', path);
 try {
  // قم بتحديد المسار الهدف.
  const {type} = await Nexus.FileSystem.stat(filePath);
  if (type === Nexus.FileSystem.FileType.Directory) // إذا كان مجلدًا، أعد return 'index.html'
   return createInputStream(Nexus.FileSystem.join(filePath, 'index.html'));
  else if (type === Nexus.FileSystem.FileType.Unknown || type === Nexus.FileSystem.FileType.NotFound)
   // إذا لم يتم العثور عليه، ألقي NotFound.
   throw new NotFoundError(الطريق);
 }
  if (e.code)
   throw e;
  throw new NotFoundError(الطريق);
 }
 try {
  // أولاً، نقوم بإنشاء جهاز.
  const fileDevice = new Nexus.IO.FilePushDevice(filePath);
  // ثم نرجع إلى ReadableStream الجديدة التي تم إنشاؤها باستخدام جهاز المصدر الخاص بنا.
  return new Nexus.IO.ReadableStream(fileDevice);
 }
  throw new InternalServerError(e.message);
 }
}
/**
 * معدير الاتصالات.
 */
let connections = 0;
/**
 * أنشئ خادم HTTP جديد.
 * @type {Nexus.Net.HTTP.Server}
 */
const server = new Nexus.Net.HTTP.Server();
// يعني خطأ الخادم أن هناك خطأ حدث أثناء استماع الخادم إلى الاتصالات.
// يمكننا تجاهل مثل هذه الأخطاء، نعرضها على أي حال.
server.on('error', e => {
 console.error(FgRed + Bright + 'Server Error: ' + e.message + '\n' + e.stack, Reset);
});
/**
 * استمع إلى الاتصالات.
 */
server.on('connection', async (connection, peer) => {
 // يبدأ مع معرف الاتصال 0، يزيد في كل اتصال جديد.
 const connId = connections++;
 // سجّل وقت البدء لهذه الاتصال.
 const startTime = Date.now();
 // دعم تفكيك الدوال مدعوم، لماذا لا تستخدمه؟
 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); });
  // قم بتعيين نوع المحتوى ووضع حالة الطلب.
  response
   .set('Content-Type', mimeType(path))
   .status(200);
  // ربط الدخل إلى الخروج(الخارج).
  const disconnect = inStream.pipe(outStream);
  try {
   // استئناف تدفق ملفنا، هذا يجعل التدفق يتحول إلى تشفير قطع HTTP.
   // هذا سيقوم بإرجاع وعد (promise) سيتم حله فقط بعد كتابة آخر بيت (قطعة HTTP).
   await inStream.resume();
  }
   // التقط أي أخطاء تحدث أثناء التدفق.
   errors.push(e);
  }
  // قطع جميع الدالات المكتوبة بواسطة `.pipe()`.
  return disconnect();
 }
  // إذا حدث خطأ، أضفه إلى الصف.
  errors.push(e);
  // قم بتعيين نوع المحتوى، حالة، وكتابة رسالة أساسية.
  response
   .set('Content-Type', 'text/plain')
   .status(e.code || 500)
   .send(e.message || 'An error has occurred.');
 }
  // قم بإغلاق تدفق البيانات يدويًا. هذا مهم لأنه قد نفد من معالجات الملف إذا لم نفعل ذلك.
  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` +
   (if (errors.length) " " + FgRed + Bright + errors.map(error => error.message).join(', ') + Reset : Reset));
 }
});
/**
 * IP والمنفذ للاستماع إليه.
 */
const ip = '0.0.0.0', port = 3000;
/**
 * ما إذا كان يجب إعداد علم `reuse` أم لا. (اختياري، الافتراضي=false)
 */
const portReuse = true;
/**
 * أقصى عدد من الاتصالات المتوازية المسموح بها. الافتراضي هو 128 على نظامي. (اختياري، نظام معين)
 * @type {number}
 */
const maxConcurrentConnections = 1000;
/**
 * ربط العنوان والمنفذ المحدد.
 */
server.bind(ip, port, portReuse);
/**
 * تبدأ في الاستماع إلى الطلبات.
 */
server.listen(maxConcurrentConnections);
/**
 * استمتع بالبث!
 */
console.log(FgGreen + `Nexus.js HTTP server listening at ${ip}:${port}` + Reset);

تقييم

أعتقد أنني قد غطيت كل شيء تم تحقيقه حتى الآن. لذا دعونا نتحدث عن الأداء الآن.

هذا هو التقييم الحالي للخادم HTTP المذكور أعلاه، مع 100 اتصال متوازي و10000 طلب في المجموع:

هذا هو ApacheBench، إصدار 2.3 <$Revision: 1796539 $>
حقوق النسخ محفوظة 1996 Adam Twiss، Zeus Technology Ltd، http://www.zeustech.net/
مصرح له ببرنامج Apache Software Foundation، 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 [ملي ثانية] (المعدل)
الوقت لكل طلب:    0.999 [ملي ثانية] (المعدل، عبر جميع الطلبات المتزامنة)
سرعة التحويل:     8590.14 [كيلو بايت/ثانية] مُستقبل
أوقات الاتصال (ملي ثانية)
       الحد الأدنى المعدل[+/-sd] الوسطي  الأعلى
الإتصال:    0  0  0.1   0    1
المعالجة:   6  99 36.6   84   464
الانتظار:    5  99 36.4   84   463
المجموع:     6 100 36.6   84   464
نسبة الطلبات التي تم خدمتها داخل وقت معين (ملي ثانية)
 50%   84
 66%   97
 75%  105
 80%  112
 90%  134
 95%  188
 98%  233
 99%  238
 100%  464 (أطول طلب)

كل ثانية 1000 طلب. على معالج قديم من نوع i7، يعمل عليه هذا البرنامج الأساسي للتحليل، وIDE يأخذ 5 جيجا بايت من الذاكرة، بالإضافة إلى الخادم نفسه.

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   : نعم
fpu_exception  : نعم
cpuid level : 13
wp   : نعم
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
الخطأ: }}
bogomips: 6784.18
حجم clflush: 64
تساوي التخزين المؤقت: 64
أحجام العناوين: 39 بت مادي، 48 بت افتراضي
إدارة الطاقة:

لقد حاولت 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: تحذير: الحد الأقصى لمقدار ملفات الفتح > FD_SETSIZE؛ تقييد الحد الأقصى لمقدار ملفات الفتح إلى FD_SETSIZE
طول انفجار الاتصال الأقصى: 262
المجموع: الاتصالات 9779 الطلبات 9779 الردود 9779 وقت الاختبار 10.029 ثانية
معدل الاتصال: 975.1 اتصال/ثانية (1.0 مللي ثانية/اتصال، <=1022 اتصالات متزامنة)
زمن الاتصال [مللي ثانية]: الأدنى 0.5 الوسطي 337.9 الأعلى 7191.8 الوسط 79.5 التباين 848.1
زمن الاتصال [مللي ثانية]: الاتصال 207.3
طول الاتصال [ردود/اتصال]: 1.000
معدل الطلب: 975.1 طلب/ثانية (1.0 مللي ثانية/طلب)
حجم الطلب [ب]: 62.0
معدل الرد [رد/ثانية]: الأدنى 903.5 المتوسط 974.6 الأعلى 1045.7 الاختلاف المعياري 100.5 (2 عينة)
وقت الرد [ملي ثانية]: الرد 129.5 نقل 1.1
حجم الرد [بايت]: العنوان 89.0 المحتوى 8660.0 التذييل 2.0 (إجمالي 8751.0)
حالة الرد: 1xx=0 2xx=9779 3xx=0 4xx=0 5xx=0
وقت المعالجة للمعالج [ثانية]: المستخدم 0.35 النظام 9.67 (المستخدم 3.5% النظام 96.4% إجمالي 99.9%)
إدخال الشبكة: 8389.9 KB/s (68.7*10^6 بيت/ثانية)
الخطأ: إجمالي 221 client-timo 0 socket-timo 0 connrefused 0 connreset 0
الخطأ: fd-unavail 221 addrunavail 0 ftab-full 0 other 0

كما ترونه، لازال يعمل. على الرغم من أن بعض الاتصالات قد تتخطى وقت الاستجابة بسبب الضغط. أنا ما زلت أبحث عن السبب في هذه المشكلة.

هذا هو محتوى دراسة Nexus.js الكامل لهذا المقال، إذا كان لديك أي أسئلة، يمكنك ترك تعليق في الأسفل للمناقشة، شكراً لدعم دروس النفخ.

البيان: محتوى هذا المقال تم جمعه من الإنترنت، وله حقوق الملكية الأصلية للمالك، تم إضافة المحتوى من قبل المستخدمين عبر الإنترنت، ويحمل هذا الموقع حقوق الملكية، لم يتم تعديل المحتوى بشكل يدوي، ولا يتحمل هذا الموقع أي مسؤولية قانونية. إذا كنت قد وجدت محتوى يشتبه في انتهاك حقوق النسخ، فلا تتردد في إرسال بريد إلكتروني إلى: notice#oldtoolbag.com (عند إرسال البريد الإلكتروني، يرجى استبدال '#' ب '@') لإبلاغنا، وقدم الدليل على الادعاء، إذا تم التحقق من صحة الادعاء، سيتم حذف المحتوى المزعوم فوراً.