JavaScript多线程之二~~Node.js中的Web Worker

上次研究了Web Worker在浏览器中的现状和使用方法，顺便感慨这么好用的东西为啥Node.js中没有呢，没过几天就发现原来国外早在2011年11月的时候就有人在github上开源了Web Worker规范在Node.js上的实现。这个模块是基于node-threads-a-gogo模块（简称TAGG）做的，TAGG是个挺不错的模块，可以让Node.js实现多线程编程，充分使用服务器的资源。Node.js从此不再限于I/O密集型的场景，也使它的触角可以拓展到更加广阔的领域。相比于TAGG，webworker-threads（目前版本是0.6.2）提供了更加易用的功能，重要的是它的接口实现是参考HTML5中定义的Web Worker标准，易于掌握前后端一致的开发方法。

先看一张关于这个模块使用后整个Node.js系统的图示，点击查看出处：



参考webworker-threads的说明文档中的例子：

[code]var Worker = require('webworker-threads').Worker;
// var w = new Worker('worker.js'); // Standard API 

// You may also pass in a function: 
var worker = new Worker(function(){
  postMessage("I'm working before postMessage('ali').");
  this.onmessage = function(event) {
    postMessage('Hi ' + event.data);
    self.close();
  };
});
worker.onmessage = function(event) {
  console.log("Worker said : " + event.data);
};
worker.postMessage('ali');

这段代码是让Node.js主线程建立一个工作线程，并且通过

worker.postMessage('ali');

向工作线程发出消息，工作线程接受到消息后加上”Hi “后返回给主线程。运行结果如下：

[code]Worker said : I'm working before postMessage('ali').
Worker said : Hi ali

从这段代码可以看出，Node.js中也可以通过webworker-threads模块，让编码者参考HTML5中的Web Worker规范来实现多线程编程应用。主线程和工作线程通过postMessage和onmessage发送和监听消息，使用方式非常简便。

为了证明Web Worker的运行和主线程确实是个异步的过程，可以通过CPU密集型的计算来检验：

[code]var Worker = require('webworker-threads').Worker;
require('http').createServer(function(req, res) {
  var fibo = new Worker(function() {
    function fibo(n) {
      return n > 1 ? fibo(n - 1) + fibo(n - 2) : 1;
    }
    this.onmessage = function(event) {
      postMessage(fibo(event.data));
    }
  });
  fibo.onmessage = function(event) {
    console.log('====computing end===');
    res.end('fib(40) = ' + event.data);
  };

  fibo.postMessage(40);
  console.log('====computing started===');
  interval();
}).listen(3333);

console.log("server started");

function interval(){
  setInterval(function(){
    console.log("主线程在运行");
  },100);
}

上面的代码中，在Node.js的主线程建立了一个HTTP服务器，监听3333端口，当接受到客户端请求后，服务器端打印“====computing started===”并开始计算斐波那契数列40个值的和，这个计算工作在另外的线程中进行，主线程持续打印“主线程在运行”的字符串，斐波那契数列计算完成后主线程接受到计算结果并打印“====computing end===”字符串。运行结果如下：

[code]server started
====computing started===
主线程在运行
主线程在运行
主线程在运行
主线程在运行
主线程在运行
主线程在运行
主线程在运行
主线程在运行
主线程在运行
主线程在运行
主线程在运行
主线程在运行
主线程在运行
主线程在运行
主线程在运行
====computing end===
主线程在运行
主线程在运行
主线程在运行
主线程在运行
....

从运算结果可以看出，当工作线程运算时，主线程并没有被阻塞。

由于webworker-threads是在TAGG的基础上实现的，因此也可以不使用Web Worker，而直接创建线程对象Thread或线程池ThreadPool来实现多线程编程。

API附录：

模块API：

引入webworker-threads的方式：

var Threads= require('webworker-threads');

Threads.Worker()

返回Worker对象

Threads.create()

返回thread对象

Threads.createPool(numThreads)

返回threadPool对象

Web Worker API:

使用三种方式生成Web Worker对象：

[code]var worker= new Threads.Worker('worker.js');
var worker= new Threads.Worker(function(){ ... });
var worker= new Threads.Worker();

worker.postMessage(data)

向worker发送消息

worker.onmessage

从worker接受消息

worker.terminate()

结束worker线程

worker.addEventListener(type,callback)

通常用于

worker.addEventListener('message', callback)

dispatchEvent(event)

未实现

removeEventListener(type)

未实现

worker.thread

返回worker的thread对象

Thread API：

创建方式：

var thread= Threads.create();

thread.id

线程序列号

thread.load(absolutePath[,callback])

读取绝对路径上的文件并执行文件内容和callback回调函数

thread.eval(program[,callback])

在thread全局上下文中执行program代码，之后调用callback(error,completionValue)

thread.on(eventType,listener)

注册事件监听

thread.once(eventType,listener)

注册单次事件监听

thread.removeAllListeners([eventType])

移除事件监听

thread.emit(eventType,eventData[,eventData……])

发出eventType类型的事件

thread.destroy()

摧毁thread对象

Thread Pool API:

创建方式：

var threadPool=Threads.createPool(numberOfThreads);

threadPool.load(absolutePath[,callback])

在所有池内线程上执行

thread.load(absolutePath[,callback])

threadPool.any.eval(program,callback)

在任意池内线程执行

thread.eval()

threadPool.any.emit(eventType,eventData[,eventData...])

在任意池内线程执行

thread.emit()

threaPool.all.eval(program,callback)

在所有池内线程上执行

thread.eval()

threadPool.all.emit(eventType,eventData[,eventData...])

在所有池内线程执行

thread.emit()

threadPool.on(eventType,listener)

在任意池内线程注册事件监听

threadPool.totalThread()

返回池内线程数目

threadPool.idleThread()

返回池内空闲线程数目

threadPool.pendingJobs()

返回尚未执行的工作数目

threadPool.destroy([rudely])

摧毁线程池，鉴于rudely等待pendingJobs数量为0时摧毁还是立刻摧毁

Web Worker内部全局属性和方法：

self.postMessage(data)

向主线程发送消息

self.onmessage

定义接受主线程消息的回调函数

self.close()

关闭自身线程

self.addEventListener(type,callback)

定义事件监听

self.dispatchListener(event)

发出事件

self.removeEventListener(type)

未实现

self.importScripts(file[,file,...])

从磁盘加载js文件并在当前线程上下文中执行

self.thread

返回webworker的线程对象

Thread全局属性和方法：

self.id

线程序列号

self.on(eventType,listener)

类似于

thread.on()

self.once(eventType,listener)

类似于

thread.once()

self.emit(eventType,eventData[,eventData...])

类似于

thread.emit()

self.removeAllListener([eventType])

类似于

thread.removeAllListeners()

self.nextTick(function)

类似于

process.nextTick()

，但是比它更快

其他全局方法：

在每个worker中都可以使用下列方法协助调试：

console.log(arg1[,arg2...])

console.err(arg1[,arg2...])

打印到stderr

puts(arg1[,arg2...])

打印到stdout

更多的例子和API可以看参考文献。

参考文献：

1. webworker-threads

2. node-threads-a-gogo的GitHub链接

3. webworker-threads的GitHub链接

4. nodejs多线程，真正的非阻塞

5. Web Workers in HTML5