摘要：歡迎關(guān)注我的知乎專欄這幾天寫了個(gè)小型的框架，最初只是想用寫個(gè)純平臺(tái)的東西，后來無意中開了個(gè)腦洞，如果基于把瀏覽器當(dāng)做，那豈不是只要是能運(yùn)行瀏覽器或者的設(shè)備，都可以作為分布式計(jì)算中的一個(gè)了嗎打開一張網(wǎng)頁，就能成為分布式計(jì)算的一個(gè)節(jié)點(diǎn)，看起

歡迎關(guān)注我的知乎專欄： https://zhuanlan.zhihu.com/starkwang

starkwang/Maus: A Simple JSON-RPC Framework running in NodeJS or Browser, based on websocket.

這幾天寫了個(gè)小型的RPC框架，最初只是想用 TCP-JSON 寫個(gè)純 NodeJS 平臺(tái)的東西，后來無意中開了個(gè)腦洞，如果基于 Websocket 把瀏覽器當(dāng)做 RPC Server ，那豈不是只要是能運(yùn)行瀏覽器（或者nodejs）的設(shè)備，都可以作為分布式計(jì)算中的一個(gè) Worker 了嗎？

打開一張網(wǎng)頁，就能成為分布式計(jì)算的一個(gè)節(jié)點(diǎn)，看起來還是挺酷炫的。

可以參考：誰能用通俗的語言解釋一下什么是RPC框架？ - 知乎

簡單地說就是你可以這樣注冊一個(gè)任意數(shù)量的worker（姑且叫這個(gè)名字好了），它里面聲明了具體的方法實(shí)現(xiàn)：

var rpcWorker = require("maus").worker;
rpcWorker.create({
    add: (x, y) => x + y
}, "http://192.168.1.100:8124");

然后你可以在另一個(gè)node進(jìn)程里這樣調(diào)用：

var rpcManager = require("maus").manager;
rpcManager.create(workers => {
    workers.add(1, 2, result => console.log(result));
}, 8124)

這里我們封裝了底層的通信細(xì)節(jié)（可以是tcp、http、websocket等等）和任務(wù)分配，只需要用異步的方式去調(diào)用worker提供的方法即可，通過這個(gè)我們可以輕而易舉地做到分布式計(jì)算的map和reduce：

rpcManager.create(workers => {
    //首先定義一個(gè)promise化的add
    var add = function(x, y){
        return new Promise((resolve, reject)=>{
            workers.add(x, y, result => resolve(result));
        })
    }
    //map&reduce
    Promise.all([add(1,2), add(3,4), add(4,5)])
        .then(result => result.reduce((x, y) => x + y))
        .then(sum => console.log(sum)) //19
}, 8124)

如果我們有三個(gè)已經(jīng)注冊的Worker（可能是本地的另一個(gè)nodejs進(jìn)程、某個(gè)設(shè)備上的瀏覽器、另一個(gè)機(jī)器上的nodejs），那么我們這里會(huì)分別在這三個(gè)機(jī)器上分別計(jì)算三個(gè)add，并且將三個(gè)結(jié)果在本地相加，得到最后的值，這就是分布式計(jì)算的基礎(chǔ)。

要實(shí)現(xiàn)雙向的通信，我們首先要定義這樣一個(gè)“遠(yuǎn)程調(diào)用”的通信標(biāo)準(zhǔn)，在我的實(shí)現(xiàn)中比較簡單：

{
    [id]: uuid          //在某些通信中需要唯一標(biāo)識(shí)碼
    message: "......"   //消息類別
    body: ......        //攜帶的數(shù)據(jù)
}

首先我們要解決的問題是，如何讓Manager知道Worker提供了哪些方法可供調(diào)用？

這個(gè)問題其實(shí)很簡單，只要在 websocket 建立的時(shí)刻發(fā)送一個(gè)init消息就可以了，init消息大概長這樣：

{
    message: "init",
    body: ["add", "multiply"] //body是方法名組成的數(shù)組
}

同時(shí)，我們要將Manager傳入的回調(diào)函數(shù)，記錄到Manager.__workersStaticCallback中，以便延遲調(diào)用：

manager.create(callback, port) //記錄下這個(gè)callback

//一段時(shí)間后。。。。。。

manager.start() //任務(wù)開始

現(xiàn)在我們的Manager收到了一個(gè)遠(yuǎn)程可調(diào)用的方法名組成的數(shù)組，我們接下來需要在Manager中生成一個(gè)workers實(shí)例，它應(yīng)該包含所有這些方法名，但底層依然是調(diào)用一個(gè)webpack通信。這里我們可以用類似元編程的奇技淫巧，下面的是部分代碼：

//收到worker發(fā)來的init消息之后
var workers = {
    __send: this.__send.bind(this), //這個(gè)this指向Manager，而不是自己
    __functionCall: this.__functionCall.bind(this) //同上
};
var funcNames = data.body; //比如["add", "multiply"]
funcNames.forEach(funcName => {
    //使用new Function的奇技淫巧
    rpc[funcName] = new Function(`
        //截取參數(shù)
        var params = Array.prototype.slice.call(arguments,0,arguments.length-1);
        var callback = arguments[arguments.length-1];
        
        //這個(gè)__functionCall調(diào)用了Manager底層的通信，具體在后面解釋
        this.__functionCall("${funcName}",params,callback);
    `)
})
//將workers注冊到Manager內(nèi)部
this.__workers = workers;
//如果此時(shí)Manager已經(jīng)在等待開始了，那么開始任務(wù)
if (this.__waitingForInit) {
    this.start();
}

還記得上面我們有個(gè)start方法么？它是這樣寫的：

start: function() {
    if (this.__workers != undefined) {
        //如果初始化完畢，workers實(shí)例存在
        this.__workersStaticCallback(this.__workers);
        this.__waitingForInit = false;
    } else {
        //否則將等待初始化完畢
        this.__waitingForInit = true;
    }
},

如果只是單個(gè)Worker和單個(gè)Manager，并且遠(yuǎn)程方法都是同步而非異步的，那么我們顯然不需要考慮返回值順序的問題：

比如我們的Manager調(diào)用了下面一堆方法：

workers.add(1, 1, callback);
workers.add(2, 2, callback);
workers.add(3, 3, callback);

由于Worker中add的是同步的方法，那么顯然我們收到返回值的順序是：

2
4
6

但如果Worker中存在一個(gè)異步調(diào)用，那么這個(gè)順序就會(huì)被打亂：

workers.readFile("xxx", callback);
workers.add(1, 1, callback);
workers.add(2, 2, callback);

顯然我們收到的返回值順序是：

2
4
content of xxx

所以這里就需要對發(fā)出的函數(shù)調(diào)用做一個(gè)序列化，具體的方法就是對于每一個(gè)調(diào)用都給一個(gè)uuid（唯一標(biāo)識(shí)碼）。

比如我們調(diào)用了：

workers.add(1, 1, stupid_callback);

那么首先Manager會(huì)對這個(gè)調(diào)用生成一個(gè) uuid ：

9557881b-25d7-4c94-84c8-2463c53b67f4

然后在__callbackStore中將這個(gè) uuid 和stupid_callback 綁定，然后向選中的某個(gè)Worker發(fā)送函數(shù)調(diào)用信息（具體怎么選Worker我們后面再說）：

{
    id: "9557881b-25d7-4c94-84c8-2463c53b67f4",
    message: "function call",
    body: { 
        funcName: "add", 
        params: [1, 1] 
    }
}

Worker執(zhí)行這個(gè)函數(shù)之后，發(fā)送回來一個(gè)函數(shù)返回值的信息體，大概是這樣：

{
    id: "9557881b-25d7-4c94-84c8-2463c53b67f4",
    message: "function call",
    body: { 
        result: 2 
    }
}

然后我們就可以在__callbackStore中找到這個(gè) uuid 對應(yīng)的 callback ，并且執(zhí)行它：

this.__callbackStore[id](result);

這就是workers.add(1, 1, stupid_callback)這行代碼背后的原理。

如果存在多個(gè)Worker，顯然我們不能把所有的調(diào)用都傻傻地發(fā)送到第一個(gè)Worker身上，所以這里就需要有一個(gè)任務(wù)分配機(jī)制，我的機(jī)制比較簡單，大概說就是在一張表里對每個(gè)Worker記錄下它是否繁忙的狀態(tài)，每次當(dāng)有調(diào)用需求的時(shí)候，先遍歷這張表，

如果找到有空閑的Worker，那么就將對它發(fā)送調(diào)用；

如果所有Worker都繁忙，那么先把這個(gè)調(diào)用暫存在一個(gè)隊(duì)列之中；

當(dāng)收到某個(gè)Worker的返回值后，會(huì)檢查隊(duì)列中是否有任務(wù)，有的話，那么就對這個(gè)Worker發(fā)送最前的函數(shù)調(diào)用，若沒有，就把這個(gè)Worker設(shè)為空閑狀態(tài)。

具體任務(wù)分配的代碼比較冗余，分散在各個(gè)方法內(nèi)，所以只介紹方法，就不貼上來了/w

全部的Manager代碼在這里（抱歉還沒時(shí)間補(bǔ)注釋）：

Maus/manager.js at master · starkwang/Maus

這里要再說一遍，我們的RPC框架是基于websocket的，所以Worker可以是一個(gè)PC瀏覽器?。。】梢允且粋€(gè)手機(jī)瀏覽器?。?！可以是一個(gè)平板瀏覽器?。?！

Worker的實(shí)現(xiàn)遠(yuǎn)比Manager簡單，因?yàn)樗恍枰獙ξㄒ灰粋€(gè)Manager通信，它的邏輯只有：

接收Manager發(fā)來的數(shù)據(jù)；

根據(jù)數(shù)據(jù)做出相應(yīng)的反應(yīng)（函數(shù)調(diào)用、初始化等等）；

發(fā)送返回值

所以我們也不放代碼了，有興趣的可以看這里：

Maus/worker.js at master · starkwang/Maus

假設(shè)我們的加法是通過這個(gè)框架異步調(diào)用的，那么我們該怎么寫算法呢？

在單機(jī)情況下，寫個(gè)斐波拉契數(shù)列簡直跟喝水一樣簡單（事實(shí)上這種暴力遞歸的寫法非常非常傻逼且性能低下，只是作為范例演示用）：

var fib = x => x>1 ? fib(x-1)+fib(x-2) : x

但是在分布式環(huán)境下，我們要將workers.add方法封裝成一個(gè)Promise化的add：

//這里的x, y可能是數(shù)字，也可能是個(gè)Promise，所以要先調(diào)用Promise.all
var add = function(x, y){
    return Promise.all([x, y])
        .then(arr => new Promise((resolve, reject) => {
            workers.add(arr[0], arr[1], result => resolve(result));
        }))
}

然后我們就可以用類似同步的遞歸方法這樣寫一個(gè)分布式的fib算法：

var fib = x => x>1 ? add(fib(x-1), fib(x-2)) : x;

然后你可以嘗試用你的電腦里、樹莓派里、服務(wù)器里的nodejs、手機(jī)平板上的瀏覽器作為一個(gè)Worker，總之集合所有的計(jì)算能力，一起來計(jì)算這個(gè)傻傻的算法（事實(shí)上相比于單機(jī)算法會(huì)慢很多很多，因?yàn)橥ㄐ派系难舆t遠(yuǎn)大于單機(jī)的加法計(jì)算，但只是為了演示啦）：

//分布式計(jì)算fib(40)
fib(40).then(result => console.log(result));