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

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

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

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

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

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

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

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

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

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

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

rpcManager.create(workers => {
    //首先定義一個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)

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

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

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

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

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

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

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

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

//一段時間后。。。。。。

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

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

//收到worker發(fā)來的init消息之后
var workers = {
    __send: this.__send.bind(this), //這個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];
        
        //這個__functionCall調(diào)用了Manager底層的通信，具體在后面解釋
        this.__functionCall("${funcName}",params,callback);
    `)
})
//將workers注冊到Manager內(nèi)部
this.__workers = workers;
//如果此時Manager已經(jīng)在等待開始了，那么開始任務(wù)
if (this.__waitingForInit) {
    this.start();
}

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

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

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

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

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

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

2
4
6

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

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

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

2
4
content of xxx

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

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

workers.add(1, 1, stupid_callback);

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

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

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

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

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

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

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

this.__callbackStore[id](result);

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

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

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

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

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

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

全部的Manager代碼在這里（抱歉還沒時間補注釋）：

Maus/manager.js at master · starkwang/Maus

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

Worker的實現(xiàn)遠比Manager簡單，因為它只需要對唯一一個Manager通信，它的邏輯只有：

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

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

發(fā)送返回值

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

Maus/worker.js at master · starkwang/Maus

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

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

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

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

//這里的x, y可能是數(shù)字，也可能是個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));
        }))
}

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

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

然后你可以嘗試用你的電腦里、樹莓派里、服務(wù)器里的nodejs、手機平板上的瀏覽器作為一個Worker，總之集合所有的計算能力，一起來計算這個傻傻的算法（事實上相比于單機算法會慢很多很多，因為通信上的延遲遠大于單機的加法計算，但只是為了演示啦）：

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