nodeJs多線程 -- 學習記錄

Fibers 和 Threads

Fibers 稱纖程，可以理解為協同程序，類似py和lua都有這樣的模型。使用Fibers可以避免對資源的互搶，減少cpu和內存的消耗，但是Fibers並不能夠真正的並行執行，同一時刻只有一個Fibers在執行，如果在其中一個Fibers中執行過多的cpu操作或者寫了個死循環，則整個主程序將卡死住。node中的異步事件循環模型就有點象這個。

Threads 又稱線程，他可以在同一時刻並行的執行，他們共享主進程的內存，在其中某一時刻某一個threads鎖死了，是不會影響主線程以及其他線程的執行。但是為了實現這個模型，我們不得不消耗更多的內存和cpu為線程切換的開銷，同時也存在可能多個線程對同一內存單元進行讀寫而造成程序崩潰的問題。

1. cluster

創始人Ryan Dahl建議，運行多個Nodejs進程，利用某些通信機制來協調各項任務。目前，已經有不少第三方的Node.js多進程支持模塊發布，而NodeJS 0.6.x 以上的版本提供了一個cluster模塊 ,允許創建“共享同一個socket”的一組進程,用來分擔負載壓力。本篇文章就基於該cluster模塊來講述Node.js在多核CPU下的編程

A single instance of Node.js runs in a single thread. To take advantage of multi-core systems the user will sometimes want to launch a cluster of Node.js processes to handle the load.

The cluster module allows you to easily create child processes that all share server ports.

 

官網實例

const cluster = require('cluster');
const http = require('http');
const numCPUs = require('os').cpus().length;

if (cluster.isMaster) {
  // Fork workers.
  for (var i = 0; i < numCPUs; i++) {
    cluster.fork();
  }

  cluster.on('exit', (worker, code, signal) => {
    console.log('worker' + worker.id +'died');
  });
} else {
  // Workers can share any TCP connection
  // In this case it is an HTTP server
  http.createServer((req, res) => {
    res.writeHead(200);
    res.end('hello world\n');
  }).listen(8000);
}

 這段代碼是有主線程 根據cups數目創建子進程,可以根據cluster.work.id打印出是哪個進程處理該請求。

 

 

2.node-threads-a-gogo

   a.安裝  

npm install threads_a_gogo
實例
A.- Here's a program that makes Node's event loop spin freely and as fast as possible: it simply prints a dot to the console in each turn:

//cat examples/quickIntro_loop.js

1 (function spinForever () {
   process.stdout.write(".");
   process.nextTick(spinForever);
})();

 

B.- Here's another program that adds to the one above a fibonacci(35) call in each turn, a CPU-bound task that takes quite a while to complete and that blocks the event loop making it spin slowly and clumsily. The point is simply to show that you can't put a job like that in the event loop because Node will stop performing properly when its event loop can't spin fast and freely due to a callback/listener/nextTick()ed function that's blocking.

//cat examples/quickIntro_blocking.js
function fibo (n) {
  return n > 1 ? fibo(n - 1) + fibo(n - 2) : 1;
}

(function fiboLoop () {
  process.stdout.write(fibo(35).toString());
  process.nextTick(fiboLoop);
})();

(function spinForever () {
  process.stdout.write(".");
  process.nextTick(spinForever);
})();

C.- The program below uses threads_a_gogo to run the fibonacci(35) calls in a background thread, so Node's event loop isn't blocked at all and can spin freely again at full speed:

1 //cat examples/quickIntro_oneThread.js
 2 function fibo (n) {
 3   return n > 1 ? fibo(n - 1) + fibo(n - 2) : 1;
 4 }
 5 
 6 function cb (err, data) {
 7   process.stdout.write(data);
 8   this.eval('fibo(35)', cb);
 9 }
10 
11 var thread= require('threads_a_gogo').create();
12 
13 thread.eval(fibo).eval('fibo(35)', cb);
14 
15 (function spinForever () {
16   process.stdout.write(".");
17   process.nextTick(spinForever);
18 })();

 

D.- This example is almost identical to the one above, only that it creates 5 threads instead of one, each running a fibonacci(35) in parallel and in parallel too with Node's event loop that keeps spinning happily at full speed in its own thread:

//cat examples/quickIntro_fiveThreads.js

function fibo (n) {

  return n > 1 ? fibo(n - 1) + fibo(n - 2) : 1;  3 }  4 
function cb (err, data) {  6   process.stdout.write(" ["+ this.id+ "]"+ data);  7   this.eval('fibo(35)', cb);  8 }  9 
var threads_a_gogo= require('threads_a_gogo'); 11 
threads_a_gogo.create().eval(fibo).eval('fibo(35)', cb); 13 threads_a_gogo.create().eval(fibo).eval('fibo(35)', cb); 14 threads_a_gogo.create().eval(fibo).eval('fibo(35)', cb); 15 threads_a_gogo.create().eval(fibo).eval('fibo(35)', cb); 16 threads_a_gogo.create().eval(fibo).eval('fibo(35)', cb); 17 
(function spinForever () { 19   process.stdout.write("."); 20  process.nextTick(spinForever); 21 })();

 

 E.- The next one asks threads_a_gogo to create a pool of 10 background threads, instead of creating them manually one by one:

//cat examples/multiThread.js
function fibo (n) {
  return n > 1 ? fibo(n - 1) + fibo(n - 2) : 1;
}

var numThreads= 10;
var threadPool= require('threads_a_gogo').createPool(numThreads).all.eval(fibo);

threadPool.all.eval('fibo(35)', function cb (err, data) {
  process.stdout.write(" ["+ this.id+ "]"+ data);
  this.eval('fibo(35)', cb);
});

(function spinForever () {
  process.stdout.write(".");
  process.nextTick(spinForever);
})();

 

F.- This is a demo of the threads_a_gogo eventEmitter API, using one thread:

//cat examples/quickIntro_oneThreadEvented.js
var thread= require('threads_a_gogo').create();
thread.load(__dirname + '/quickIntro_evented_childThreadCode.js');

/*
  This is the code that's .load()ed into the child/background thread:

  function fibo (n) {
    return n > 1 ? fibo(n - 1) + fibo(n - 2) : 1;
  }

  thread.on('giveMeTheFibo', function onGiveMeTheFibo (data) {
    this.emit('theFiboIs', fibo(+data)); //Emits 'theFiboIs' in the parent/main thread.
  });

*/

//Emit 'giveMeTheFibo' in the child/background thread.
thread.emit('giveMeTheFibo', 35);

//Listener for the 'theFiboIs' events emitted by the child/background thread.
thread.on('theFiboIs', function cb (data) {
  process.stdout.write(data);
  this.emit('giveMeTheFibo', 35);
});

(function spinForever () {
  process.stdout.write(".");
  process.nextTick(spinForever);
})();

G.- This is a demo of the threads_a_gogo eventEmitter API, using a pool of threads:

 

//cat examples/quickIntro_multiThreadEvented.js
var numThreads= 10;
var threadPool= require('threads_a_gogo').createPool(numThreads);
threadPool.load(__dirname + '/quickIntro_evented_childThreadCode.js');

/*
  This is the code that's .load()ed into the child/background threads:

  function fibo (n) {
    return n > 1 ? fibo(n - 1) + fibo(n - 2) : 1;
  }

  thread.on('giveMeTheFibo', function onGiveMeTheFibo (data) {
    this.emit('theFiboIs', fibo(+data)); //Emits 'theFiboIs' in the parent/main thread.
  });

*/

//Emit 'giveMeTheFibo' in all the child/background threads.
threadPool.all.emit('giveMeTheFibo', 35);

//Listener for the 'theFiboIs' events emitted by the child/background threads.
threadPool.on('theFiboIs', function cb (data) {
  process.stdout.write(" ["+ this.id+ "]"+ data);
  this.emit('giveMeTheFibo', 35);
});

(function spinForever () {
  process.stdout.write(".");
  process.nextTick(spinForever);
})();