So verwenden Sie worker_threads zum Erstellen neuer Threads in nodejs

const { Worker, istHauptthread } = erforderlich('worker_threads');

wenn (istHauptthread) {
 console.log('im Hauptthread');
 neuer Worker(__Dateiname);
} anders {
 console.log('im Arbeitsthread');
 console.log(isMainThread); // gibt „false“ aus.
}

Klasse MessageChannel {
  schreibgeschützter Port1: MessagePort;
  schreibgeschützter Port2: MessagePort;
 }

const { MessageChannel } = erfordern ('worker_threads');

const { port1, port2 } = neuer MessageChannel();
port1.on('Nachricht', (Nachricht) => console.log('empfangen', Nachricht));
port2.postMessage({ foo: 'bar' });
// Druckt: empfangen { foo: 'bar' } vom Listener `port1.on('message')`

Klasse MessagePort erweitert EventEmitter {
  schließen(): ungültig;
  postMessage(Wert: beliebig, Übertragungsliste?: Array<ArrayBuffer | MessagePort>): void;
  ref(): ungültig;
  unref(): ungültig;
  start(): ungültig;

  addListener(Ereignis: "schließen", listener: () => void): dies;
  addListener(Ereignis: "Nachricht", Listener: (Wert: beliebig) => void): dies;
  addListener(Ereignis: Zeichenfolge | Symbol, Listener: (...Argumente: beliebig[]) => void): dies;

  emittieren(Ereignis: "schließen"): boolesch;
  emittieren(Ereignis: "Nachricht", Wert: beliebig): boolesch;
  emittieren(Ereignis: Zeichenfolge | Symbol, ...Argumente: beliebig[]): Boolesch;

  bei (Ereignis: "schließen", Listener: () => void): dies;
  bei (Ereignis: „Nachricht“, Listener: (Wert: beliebig) => ungültig): dies;
  bei (Ereignis: Zeichenfolge | Symbol, Listener: (...Argumente: beliebig[]) => void): dies;

  einmal (Ereignis: „schließen“, Listener: () => void): dies;
  einmal (Ereignis: „Nachricht“, Listener: (Wert: beliebig) => ungültig): dies;
  einmal (Ereignis: Zeichenfolge | Symbol, Listener: (...Argumente: beliebig[]) => ungültig): dies;

  prependListener(Ereignis: "schließen", listener: () => void): dies;
  prependListener(Ereignis: "Nachricht", Listener: (Wert: beliebig) => void): dies;
  prependListener(Ereignis: Zeichenfolge | Symbol, Listener: (...Argumente: beliebig[]) => void): dies;

  prependOnceListener(Ereignis: "schließen", listener: () => void): dies;
  prependOnceListener(Ereignis: "Nachricht", Listener: (Wert: beliebig) => void): dies;
  prependOnceListener(Ereignis: Zeichenfolge | Symbol, Listener: (...Argumente: beliebig[]) => void): dies;

  removeListener(Ereignis: "schließen", listener: () => void): dies;
  removeListener(Ereignis: "Nachricht", Listener: (Wert: beliebig) => void): dies;
  removeListener(Ereignis: Zeichenfolge | Symbol, Listener: (...Argumente: beliebig[]) => void): dies;

  aus(Ereignis: "schließen", Listener: () => void): dies;
  aus (Ereignis: „Nachricht“, Listener: (Wert: beliebig) => ungültig): dies;
  aus (Ereignis: Zeichenfolge | Symbol, Listener: (...Argumente: beliebig[]) => ungültig): dies;
 }

const { MessageChannel } = erfordern ('worker_threads');
const { port1, port2 } = neuer MessageChannel();

// Druckt:
// foobar
// geschlossen!
port2.on('Nachricht', (Nachricht) => console.log(Nachricht));
port2.on('schließen', () => console.log('geschlossen!'));

port1.postMessage('foobar');
port1.schließen();

port.postMessage(Wert[, Übertragungsliste])

const { MessageChannel } = erfordern ('worker_threads');
const { port1, port2 } = neuer MessageChannel();

port1.on('Nachricht', (Nachricht) => console.log(Nachricht));

const circularData = {};
circularData.foo = kreisförmigeDaten;
// Druckt: { foo: [Rundschreiben] }
port2.postMessage(circularData);

const { MessageChannel } = erfordern ('worker_threads');
const { port1, port2 } = neuer MessageChannel();

port1.on('Nachricht', (Nachricht) => console.log(Nachricht));

const uint8Array = neues Uint8Array([ 1, 2, 3, 4 ]);
// Kopie von uint8Array posten:
port2.postMessage(uint8Array);

port2.postMessage(uint8Array, [ uint8Array.buffer ]);

//port2.postMessage(uint8Array);

DOMException [DataCloneError]: Ein ArrayBuffer wurde getrennt und konnte nicht geklont werden.

const { MessageChannel, markAsUntransferable } = require('worker_threads');

const pooledBuffer = neuer ArrayBuffer(8);
const typedArray1 = neues Uint8Array(pooledBuffer);
const typedArray2 = neues Float64Array(pooledBuffer);

markAlsNichtÜbertragbarMarkieren(pooledBuffer);

const { port1 } = neuer MessageChannel();
port1.postMessage(typedArray1, [typedArray1.buffer]);

console.log(typedArray1);
console.log(typedArray2);

const { Worker, SHARE_ENV } = erfordern('worker_threads');
neuer Worker('process.env.SET_IN_WORKER = "foo"', { eval: true, env: SHARE_ENV })
 .on('exit', () => {
 console.log(process.env.SET_IN_WORKER); // Gibt „foo“ aus.
 });

const { Worker, isMainThread, workerData } = erfordern('worker_threads');

wenn (istHauptthread) {
 const worker = new Worker(__filename, { workerData: 'Hallo Welt!' });
} anders {
 console.log(workerData); // Gibt „Hallo Welt!“ aus.
}

 Klasse Worker erweitert EventEmitter {
  schreibgeschützte Standardeingabe: Beschreibbar | null;
  schreibgeschützter Standardout: Lesbar;
  schreibgeschützter stderr: Lesbar;
  schreibgeschützte Thread-ID: Nummer;
  schreibgeschützte Ressourcengrenzen?: Ressourcengrenzen;

  Konstruktor (Dateiname: Zeichenfolge | URL, Optionen?: WorkerOptions);

  postMessage(Wert: beliebig, Übertragungsliste?: Array<ArrayBuffer | MessagePort>): void;
  ref(): ungültig;
  unref(): ungültig;

  beenden(): Versprechen<Zahl>;

  getHeapSnapshot(): Promise<Lesbar>;

  addListener(Ereignis: "Fehler", Listener: (err: Fehler) => void): dies;
  addListener(Ereignis: "exit", Listener: (ExitCode: Nummer) => void): dies;
  addListener(Ereignis: "Nachricht", Listener: (Wert: beliebig) => void): dies;
  addListener(Ereignis: "online", listener: () => void): dies;
  addListener(Ereignis: Zeichenfolge | Symbol, Listener: (...Argumente: beliebig[]) => void): dies;

  ... 
 }

const assert = erfordern('assert');
Konstante {
 Worker, Nachrichtenkanal, Nachrichtenport, isMainThread, übergeordneter Port
} = erfordern('worker_threads');
wenn (istHauptthread) {
 const worker = neuer Worker(__Dateiname);
 const subChannel = neuer MessageChannel();
 worker.postMessage({ hierIstIhrPort: subChannel.port1 }, [subChannel.port1]);
 subChannel.port2.on('Nachricht', (Wert) => {
 console.log('Empfangen:', Wert);
 });
} anders {
 parentPort.once('Nachricht', (Wert) => {
 assert(Wert.hierIstIhrPort Instanzvon MessagePort);
 value.hereIsYourPort.postMessage('Der Arbeitsthread sendet diese Nachricht');
 Wert.hierIstIhrPort.schließen();
 });
}

const { MessageChannel, receiveMessageOnPort } = erfordern('worker_threads');
const { port1, port2 } = neuer MessageChannel();
port1.postMessage({ hallo: 'Welt' });

console.log(Nachricht auf Port empfangen(Port2));
// Druckt: { Nachricht: { Hallo: 'Welt' } }
console.log(Nachricht auf Port empfangen(Port2));
// Druckt: undefiniert

const vm = erfordern('vm');

konstant x = 1;

const Kontext = { x: 2 };
vm.createContext(context); // Kontextisolationsobjekt.

const-Code = 'x += 40; var y = 17;';
// „x“ und „y“ sind globale Variablen im Kontext.
// Der Wert von x ist anfangs 2, da dies der Wert von context.x ist.
vm.runInContext(Code, Kontext);

console.log(Kontext.x); // 42
konsole.log(Kontext.y); // 17

console.log(x); // 1; y ist nicht definiert.

worker.moveMessagePortToContext(Port, kontextbezogene Sandbox)

const { parentPort } = erfordern('worker_threads');
parentPort.on('Nachricht', (Aufgabe) => {
 parentPort.postMessage(task.a + task.b);
});

const { AsyncResource } = erfordern('async_hooks');
const { EventEmitter } = erfordern('Ereignisse');
const path = require('Pfad');
const { Worker } = erfordern ('worker_threads');

const kTaskInfo = Symbol('kTaskInfo');
const kWorkerFreedEvent = Symbol('kWorkerFreedEvent');

Klasse WorkerPoolTaskInfo erweitert AsyncResource {
 Konstruktor(Rückruf) {
 super('WorkerPoolTaskInfo');
 this.callback = Rückruf;
 }

 erledigt(Fehler, Ergebnis) {
 this.runInAsyncScope(this.callback, null, err, Ergebnis);
 this.emitDestroy(); // `TaskInfo`s werden nur einmal verwendet.
 }
}

Klasse WorkerPool erweitert EventEmitter {
 Konstruktor(AnzahlThreads) {
 super();
 this.numThreads = AnzahlThreads;
 diese.arbeiter = [];
 dies.freieWorker = [];

 für (lass i = 0; i < numThreads; i++)
  dies.addNewWorker();
 }

 addNewWorker() {
 const worker = neuer Worker(Pfad.resolve(__dirname, 'task_processor.js'));
 worker.on('Nachricht', (Ergebnis) => {
  // Im Erfolgsfall: Rufen Sie den Callback auf, der an `runTask` übergeben wurde,
  // Entfernen Sie die mit dem Worker verknüpfte „TaskInfo“ und markieren Sie sie als frei
  // wieder.
  Arbeiter[kTaskInfo].done(null, Ergebnis);
  Arbeiter[kTaskInfo] = null;
  dies.freeWorkers.push(Arbeiter);
  dies.emit(kWorkerFreedEvent);
 });
 worker.on('Fehler', (Fehler) => {
  // Im Falle einer nicht abgefangenen Ausnahme: Rufen Sie den Callback auf, der an
  // `runTask` mit dem Fehler.
  wenn (Arbeitnehmer[kTaskInfo])
  Arbeiter[kTaskInfo].done(err, null);
  anders
  dies.emit('Fehler', err);
  // Entfernen Sie den Worker aus der Liste und starten Sie einen neuen Worker, um den
  // aktuelles.
  dies.workers.splice(dieses.workers.indexOf(worker), 1);
  dies.addNewWorker();
 });
 dies.Arbeiter.push(Arbeiter);
 dies.freeWorkers.push(Arbeiter);
 dies.emit(kWorkerFreedEvent);
 }

 runTask(Aufgabe, Rückruf) {
 wenn (this.freeWorkers.length === 0) {
  // Keine freien Threads, warten bis ein Worker-Thread frei wird.
  dies.once(kWorkerFreedEvent, () => dies.runTask(Aufgabe, Rückruf));
  zurückkehren;
 }

 const worker = this.freeWorkers.pop();
 Arbeiter[kTaskInfo] = neue WorkerPoolTaskInfo(Rückruf);
 Arbeiter.postMessage(Aufgabe);
 }

 schließen() {
 für (const worker von this.workers) worker.terminate();
 }
}

module.exports = ArbeiterPool;

const WorkerPool = erfordern('./worker_pool.js');
const os = erfordern('os');

const pool = neuer WorkerPool(os.cpus().length);

lass fertig = 0;
für (sei i = 0; i < 10; i++) {
 pool.runTask({ a: 42, b: 100 }, (fehler, ergebnis) => {
 console.log(i, err, Ergebnis);
 wenn (++fertig === 10)
  pool.schließen();
 });
}