前言
最近,明學是一個火熱的話題,而我,卻也想當那么一回明學家,那就是,把JavaScript和多線程並發這兩個八竿子打不找的東西,給硬湊了起來,還寫了一個並發庫concurrent-thread-js。尷尬的是,當我發現其中的不合理之處,即這個東東的應用場景究竟是什么時,我發現我已經把代碼寫完了。
⚠️注意! 本文中的線程指的都是用JS異步函數模擬的“假線程”,不是真正意義上的多線程,請不要誤解⚠️
github地址
本文的目的
事實上,這個庫用處很小,但是在寫的過程中,我對Promise,Async函數以及event事件流的使用產生了新的認識,同時也逐漸去學習和了解怎么去從零開始去寫一個非業務的,通用的npm模塊,所以希望拿出來和大家分享一下,這才是本文的真正的目的。
好,我們從一個故事開始。


github地址
https://github.com/penghuwan/concurrent-thread.jsgithub.com
注意!倘若不考慮webworker這種解決方案,我們一般都認為JS是單線程的。
concurrent-thread-js功能簡介
為單線程的JavaScript實現並發協調的功能,語意,命名和作用性質上參考Java的實現,提sleep/join/interupt等API以及鎖和條件變量等內容,並提供線程間通信的功能,依賴ES6語法,基於Promise和Async函數實現,故需要Babel編譯才能運行。JavaScrpt本來就是單線程的,所以這只是在API的層面實現了模擬,在下文的介紹中,每條所謂的線程其實就是普通的異步函數,並在此基礎上實現不同線程的協調配合。
為什么不選用webworker實現?
沒錯,一般來說JS中模擬多線程我們也許會選用webworker,但是它必須要求你手動創建額外的webworker腳本文件,並通過new work('work.js')這種方式使用,這並不能達到我項目中想要的API的效果,而且注意:webwork中的環境不是window!很多方法你調用不了的。你只能采取這種方案,也即在主線程完成該功能,這是我沒有選擇webworker的另一個原因。
說是這樣說,但其實在大多數時候還是用webworker就夠了
什么時候使用concurrent-thread-js
這個問題真是靈魂拷問,可是既然代碼寫都寫了,我怎么也得編一個理由出來啊!額。。。讓我想想哈
它的作用是:當JS工程需要讓兩個函數在執行上不互相干擾,同時也不希望它們會阻塞主線程,與此同時,還希望這兩個函數實現類似並發多線程之間的協調的需求的時候,你可以使用這個並發模擬庫,實際上這種應用場景。。。這尼瑪有這種應用場景嗎?!(扎心了呀)。
API總覽
-
submit(function,[namespace]): 接收一個函數,普通函數或Async函數均可,並異步執行"線程"
-
sleep(ms): "線程"休眠,可指定休眠時間ms,以毫秒計算
-
join(threadName): "線程"同步,調用此方法的"線程"函數將在threadName執行結束后繼續執行
-
interupt(threadName): "線程"中斷,影響"線程"內部調this.isInterrupted()的返回值
-
Lock.lock: 加鎖,一個時刻只能有一個"線程"函數進入臨界區,其他"線程"函數需要等待,鎖是非公平的,也就是說后面排隊的線程函數沒有先后,以隨機的方式進行競爭。
-
Lock.unlock:解除非公平鎖
-
Condition.wait:不具備執行條件,"線程"進入waiting狀態,等待被喚醒
-
Condition.notify:隨機喚醒一個wait的"線程"
-
Condition.notifyAll: 尚未編寫,喚醒所有wait的"線程"
-
getState: 還沒寫完 獲取"線程"狀態,包括RUNNALE(運行),WAITING(等待),BLOCKED(阻塞),TERMINATED(終止)
三個類:ThreadPool,Lock和Condition
我們的API分別寫入三個類中,分別是
-
ThreadPool類:包含submit/sleep/join/interrupt/getState方法
-
Lock類:包含Lock.lock和Lock.unLock方法
-
Condition類:包含Condition.wait和Condition.notify方法
注:以下所說的"線程"都是指JS中模擬的異步函數
A1.submit方法
submit模擬提交線程至線程池
// 備注:為循序漸進介紹,以下為簡化代碼 // 存儲每個線程函數的狀態,例如是否中斷,以及線程狀態等 const threadMap = {}; class ThreadPool { // 模擬線程中斷 interrupt(threadName) { } // 模擬線程同步 join(threadName, targetThread) { } // 模擬線程休眠 sleep(ms) { } }; function submit(func, name) { if (!func instanceof Function) return; // 方式1:傳入一個具名函數;方式2:傳入第二個參數,即線程命名空間 const threadName = func.name || name; // threadMap負責存儲線程狀態數據 threadMap[threadName] = { state: RUNNABLE, isInterrupted: false }; // 讓func異步調用,同時將傳入函數的作用域綁定為 ThreadPool原型 Promise.resolve({ then: func.bind(ThreadPool.prototype); }) }
首先,我們做了三件事情:
-
獲取線程函數的命名空間,並初始化線程初始數據,不同線程狀態由threadMap全局存儲
-
將提交的函數func作為Promise.resolve方法中的一個thenable對象的then參數,這相當於立即"完成"一個Promise,同時在then方法中執行func,func會以異步而不是同步的方式進行執行,你也可以簡單的理解成類似於執行了setTimeOut(func,0);
-
將func的作用域綁定為新生成的ThreadPool實例,ThreadPool中定義了我們上面我們介紹到的方法,如sleep/join/interupt等,這有什么好處呢?這意味着我們可以直接在函數中通過調用this.interrupt的方式去調用我們定義的API了,符合我們的使用習慣(注意,class中定義的除箭頭函數外的普通函數實際上都存放在原型中)
submit(async function example() { this.interrupt(); });
但問題在於:現在因為所有的函數通過this調用的都是ThreadPool原型中的方法,我們要在調用唯一的interrupt方法,需要在異步函數中傳入"線程"標識,如線程名。這顯然不方便,也不優雅,例如下面的命名為example的線程函數
submit(async function example() { this.interrupt('example'); });
使用這個模塊用戶會感到奇怪:我明明在example函數中,為什么還要給調用方法傳example這個名字參數??難道不能在模塊內部把這事情干了嗎?
對!我們下面做的就是這件事情,我們編寫一個delegateThreadPool方法,由它為ThreadPool代理處理不同“線程“函數的函數名
// 返回代理后的ThreadPool function delegateThreadPool(threadName) { // threadName為待定的線程名,在submit方法調用時候傳入 // 代理后的ThreadPool const proxyClass = {}; // 獲取ThreadPool原來的所有的方法,賦給props數組 var props = Object.getOwnPropertyNames(ThreadPool.prototype); for (let prop of props) { // 代理ThreadPool,為其所有方法增加threadName這個參數 let fnName = prop; proxyClass[fnName] = (...args) => { const fn = baseClass[fnName]; return fn(threadName, ...args); }; } return proxyClass; } function submit(func, name) { // 省略其他代碼 。。。 const proxyScope = delegateThreadPool(threadName); // 讓func異步調用,不阻塞主線程,同時實現並發 Promise.resolve({ then: function () { // 給func綁定this為代理后的ThreadPool對象,以便調用方法 func.call(proxyScope); } }); } // 調用this.sleep方法時,已經無需增加函數命名作為參數了 submit(async function example() { this.interrupt(); });
也就是說,我們的線程函數func綁定的已經不是ThreadPool.prototype了,而是delegateThreadPool處理后返回的對象:proxyScope。這時候,我們在“線程”函數體里調用this.interrupt方法時,已經無需增加函數命名作為參數了,因為這個工作,proxyScope對象幫我們做了,其實它的工作很簡單——就是它的每個函數,都在一個返回的閉包里面調用ThreadPool的同名函數,並傳遞線程名作為第一個參數。
A2. sleep方法
作用:線程休眠
sleep方法很簡單,無非就是返回一個Promise實例,在Promise的函數里面調setTimeOut,等時間到了執行resolve函數,這段時間里修飾Promise的await語句會阻塞一段時間,resolve后又await語句又繼續向下執行了,能滿足我們想要的休眠效果
// 模擬“線程”休眠 sleep(ms) { return new Promise(function (resolve) { setTimeout(resolve, ms); }) } // 提交“線程” submit(async function example() { // 阻塞停留3秒,然后才輸出1 await this.sleep(3000); console.log(1); });
A3. interrupt方法
作用:線程中斷,可用於處理線程停止等操作
這里要先介紹一下Java里面的interrupt方法:在JAVA里,你不能通過調用terminate方法停掉一個線程,因為這有可能會因為處理邏輯突然中斷而導致數據不一致的問題,所以要通過interrupt方法把一個中斷標志位置為true,然后通過isInterrupted方法作為判斷條件跳出關鍵代碼。
所以為了模擬,我在JS中處理“線程”中斷也是這么去做的,但是我們這樣做的根本原因是:我們壓根沒有可以停掉一個線程函數的方法!(JAVA是有但是不准用,即廢棄了而已)
// 模擬線程中斷 interrupt(threadName) { if (!threadName) { throw new Error('Miss function parameters') } if (threadMap[threadName]) { threadMap[threadName].isInterrupted = true; } } // 獲取線程中斷狀態 isInterrupted(threadName) { if (!threadName) { throw new Error('Miss function parameters') } // !!的作用是:將undefined轉為false return !!threadMap[threadName].isInterrupted; }
A4. join方法
join(threadName): "線程"同步,調用此方法的"線程"函數將在threadName執行結束后繼續執行
join方法和上面的sleep方法是一樣的道理,我們讓它返回一個Promise,只要我們不調resolve,那么外部修飾Promise的await語句就會一直暫停,等到join的那個另一個線程執行完了,我們看准時機!把這個Promise給resolve,這時候外部修飾Promise的await語句不就又可以向下執行了嗎?
但問題在於:我們如何實現這個“一個函數執行完通知另一個函數的功能呢”?沒錯!那就是我們JavaScript最喜歡的套路: 事件流! 我們下面使用event-emitter這個前后端通用的模塊實現事件流。
我們只要在任何一個函數結束的時候觸發結束事件(join-finished),同時傳遞該線程的函數名作為參數,然后在join方法內部監聽該事件,並在響應時候調用resolve方法不就可以了嘛。
首先是在join方法內部監聽線程函數的結束事件
import ee from 'event-emitter'; const emitter = ee(); // 模擬線程同步 join(threadName, targetThread) { return new Promise((resolve) => { // 監聽其他線程函數的結束事件 emitter.on('join-finished', (finishThread) => { // 根據結束線程的線程名finishThread做判斷 if (finishThread === targetThread) { resolve(); } }) }) }
同時在線程函數執行結束時觸發join-finished事件,傳遞線程名做參數
import ee from 'event-emitter'; const emitter = ee(); function submit(func, name) { // ... Promise.resolve({ then: func().then(() => { emitter.emit('join-finished', threadName); }) }); }
使用如下:
submit(async function thread1 () { this.join('thread2'); console.log(1); }); submit(async function thread2 () { this.sleep(3000); console.log(2) }) // 3s后,依次輸出 2 1
A5. Lock.lock & Lock.unlock(非公平鎖)
我們主要是要編寫兩個方法:lock和unlock方法。我們需要設置一個Boolean屬性isLock
-
lock方法:lock方法首先會判斷isLock是否為false,如果是,則代表沒有線程占領臨界區,那么允許該線程進入臨界區,同時把isLock設置為true,不允許其他線程函數進入。其他線程進入時,由於判斷isLock為true,會setTimeOut每隔一段時間遞歸調用判斷isLock是否為false,從而以較低性能消耗的方式模擬while死循環。當它們檢測到isLock為false時候,則會進入臨界區,同時設置isLock為true。因為后面的線程沒有先后順序,所以這是一個非公平鎖
-
unLock方法:unlock則是把isLock屬性設置為false,解除鎖定就可以了
// 這是一個非公平鎖 class Lock { constructor() { this.isLock = false; } //加鎖 lock() { if (this.isLock) { const self = this; // 循環while死循環,不停測試isLock是否等於false return new Promise((resolve) => { (function recursion() { if (!self.isLock) { // 占用鎖 self.isLock = true; // 使外部await語句繼續往下執行 resolve(); return; } setTimeout(recursion, 100); })(); }); } else { this.isLock = true; return Promise.resolve(); } } // 解鎖 unLock() { this.isLock = false; } } const lockObj = new Lock(); export default lockObj;
運行示例如下:
async function commonCode() { await Lock.lock(); await Executor.sleep(3000); Lock.unLock(); } submit(async function example1() { console.log('example1 start') await commonCode(); console.log('example1 end') }); submit(async function example2() { console.log('example2 start') await commonCode(); console.log('example2 end') });
輸出
// 立即輸出 example1 start example2 start // 3秒后輸出 example1 end // 再3秒后輸出 example2 end
A6. Condition.wait & Condition.notify(條件變量)
-
Condition.wait:不具備執行條件,線程進入waiting狀態,等待被喚醒
-
Condition.notify: 喚醒線程
對不起!寫到這里,我實在是口干舌燥,寫不下去了,但是道理和前面是一樣的:
無非是:事件監聽 + Promise + Async函數組合拳,一套搞定
import ee from 'event-emitter'; const ev = ee(); class Condition { constructor() { this.n = 0; this.list = []; } // 當不滿足條件時,讓線程處於等待狀態 wait() { return new Promise((resolve) => { const eventName = `notify-${this.n}`; this.n++; const list = this.list; list.push(eventName); ev.on(eventName, () => { // 從列表中刪除事件名 const i = list.indexOf(eventName); list.splice(i, 1); // 讓外部函數恢復執行 debugger; resolve(); }) }) } // 選擇一個線程喚醒 notify() { const list = this.list; let i = Math.random() * (this.list.length - 1); i = Math.floor(i); ev.emit(list[i]) } }
測試代碼
async function testCode() { console.log('i will be wait'); if (true) { await Condition.wait(); }; console.log('i was notified '); } submit(async function example() { testCode(); setTimeout(() => { Condition.notify(); }, 3000); });
輸出
i will be wait // 3秒后輸出 i was notified
最后的大總結
其實說到底,我想和大家分享的不是什么並發啊,什么多線程啦。
其實我想表達的是:事件監聽 + Promise + Async函數這套組合拳很好用啊
-
你想讓一段代碼停一下?OK!寫個返回Promise的函數,用await修飾,它就停啦!
-
你想控制它(await)不要停了,繼續往下走?OK! 把Promise給resolve掉,它就往下走啦
-
你說你不知道怎么控制它停,因為監聽和發射事件的代碼分布在兩個地方?OK!那就使用事件流
本文完,下面是全部項目代碼(剛寫了文章才發現有bug,待會改改)