跨平台模块tagg2，让nodejs多线程支持

我积极解决

转载自： http://snoopyxdy.blog.163.com/blog/static/6011744020134186614267/


上一篇文章详细介绍了Jorge开发的Threads A GoGo模块，这个模块让node支持了多线程的模型，让node可以更好的胜任cpu密集型的应用场景。

上篇文章的传送门：

nodejs多线程，真正的非阻塞

 



 

 

不过Threads A GoGo模块（以下简称tagg）开发的比较早，而且已经1年多没有更新了，所以tagg模块还在使用node-waf编译addon，而且tagg模块并不支持windows，对于我个人来讲tagg模块的api并不是我想要的，不友好。比如我要传入到线程的执行，需要写方法名的字符串进去，而且返回值只能通过线程中的return来获得，等等。


于是我就有了对tagg模块进行改造的想法，当然其中不乏遇到一些坑，这些坑在下一篇博客中再进行总结吧，本文主要介绍我对tagg模块进行改造而创建的tagg2模块，毕竟流的还是tagg模块的血，不想改名字了。

 

安装：npm install tagg2

请保证nodejs的版本在0.8.x以上

github项目地址：

https://github.com/DoubleSpout/node-threads-a-gogo2

 

 

1、tagg2模块对tagg模块改造最主要的一点就是跨平台支持，对windows、linux和mac都很友好，我在win8（node v0.10.6），centos 2.6.4（node v0.8.16）,以及mac ml 10.8.1（node v0.10.6）都经过测试，可以编译通过和运行测试代码。

 

2、tagg2模块第二个对tagg模块的改造就是node v0.10.x的支持，由于新的node版本将全部支持node-gyp而放弃node-waf，所以编译的方式也有所改变，tagg2模块放弃了node-waf的支持，采用node-gyp进行编译代码。

 

3、对tagg模块的api进行改造，新改造的api更加友好，比如我们执行一个执行40次斐波那契算法的线程：

 

//加载tagg2的模块

var tagg = require('tagg2'); 

 

//子线程工作函数

var th_func = function(){

var fibo =function fibo (n) {

return n > 1 ? fibo(n - 1) + fibo(n - 2) : 1;
}thread.end(fibo(40));
}

 

//创建子线程,并且注册回调

var thread = tagg.create(th_func, function(err, res){

if(err) throw new(err);//如果在线程中throw异常，err就会得到相应的错误

console.log(res);//fibo(40)的结果

thread.destroy();//摧毁线程

});



当然实际情况可能更加复杂，fibo的次数需要主线程传递进去，没关系，tagg2模块完全可以做到，tagg2可以通过buffer将参数传递给子线程，目前0.1.x版本仅支持字符串，以后会陆续支持2进制buffer支持等。

虽然tagg2使用起来更加简单，不过还是有些地方需要注意的，比如上述代码 th_func 是独立执行在子线程中的，上下文并不是node的实例，所以在子线程中是无法使用node的api，比如无法require('fs')等，也无法访问主线程的变量和模块，为什么这样下文对tagg2多线程原理会有所介绍。


 

4、tagg2模块同时还支持线程池，创建固定大小的线程时，让线程支持复用，不会让过多的创建线程造成主程序内存不足而崩溃。多余的任务会自动进行排队。

 

var thread = tagg.create(3);
thread.dirname = __dirname;

thread.pool(th_func,buf,thread_cb);//将任务放入线程池



thread.pool(th_func,buf,thread_cb);



thread.pool(th_func,buf,thread_cb);



thread.pool(th_func,buf,thread_cb);



thread.pool(th_func,buf,thread_cb);



 


console.log('thread.totalThreads: '+thread.totalThreads())//线程池中的所有线程数量



console.log('thread.idleThreads: '+thread.idleThreads()) //线程池闲置线程的数量



console.log('thread.pendingJobs: '+thread.pendingJobs()) //线程池工作的线程数量


 


thread.destory();//摧毁线程池


 





 

4、为了完善线程的局限性，tagg2还有一个特性，当我们调用 tagg.create 函数创建子线程时，可以对线程做一个配置，比如我们这样创建一个子线程：


var thread = tagg.create(th_func,{buffer:buf,fastthread:false,dirname:__dirname},function(err, res){if(err)thrownew(err);
console.log(res);//thread.end
thread.destroy();//摧毁线程});



我们设置option中的 fastthread属性为false，就会创建一个慢线程来执行任务，所谓慢线程将是一个新的nodejs的实例，拥有完全的nodejs的api，比如require('fs')等。这个慢线程实际上就是一个nodejs进程，tagg2模块式利用child.fork()来进行的。所以严格上来说是创建了一个nodejs的进程而不是线程。


 

5、在线程中tagg2设置了一些全局变量来帮助使用者更好的完成任务

 

console.log(param);用来向输出流输出内容，主要用于线程的调试等

throw(errstr);抛出一个异常，会触发回调函数，赋值其中的error参数

thread对象，保存了一些有用的属性和方法。

thread.id 标识了一条线程的id，当使用thread pool时无用

thread.buffer 保存了主线程传递过来的buffer对象，由于目前只支持string，所以此buffer对象只有toString()方法。

thread.buffer.toString()返回主线程传递过来的buffer字符串

thread.end(param)表示线程任务执行完毕，将参数传递给回调函数，会赋值回调函数的第二个参数

thread.nextTick(function)在线程中异步执行


require(filepath)当在fastthread中，require参数 filepath 请输入相对地址，表示相对于本目录的文件，比如在本目录中有a.js，所以在线程中就得写require('a.js')，fastthread 目前不支持../../和绝对路径/user等样式，下个版本将会加入。


其他一些详细的用法请参阅github上的readme。



 

6、为什么 tagg2 中的 fastthread 不能够访问全局的变量和使用node的api？


这有必要简单介绍下tagg2模块的工作原理和流程，tagg2模块是接受用户的任务，然后将任务进行包装，利用pthread创建一个线程，然后用node的libuv库，写好回调函数然后丢入到异步池中，同时tagg2模块维护了一个队列，在丢入libuv异步池的同时，也将任务的指针丢入了队列，并且告知队列有新任务。队列调用libuv的api，异步执行回调函数。

回调函数将利用v8引擎的isolate类，创建一个新的v8实例，同时v8手册上有说一个isolate实例同时只能够让一个线程进行访问，所以我们无法对当前主线程的isolate进行多线程同时读写，这也就是为什么我们无法访问当前主线程的变量的原因。

 

介绍就先写到这里吧，tagg2模块还在实验中，请不要用于对外的生产环境，以后我会丰富tagg2模块的功能。如果在安装或者使用中有任何疑问或者错误，请联系我，qq：53822985。

 

下篇再介绍本次改造碰到的坑，以及如何使用c++ addon来为node实现多线程异步的模型。