?2016-2026 Matools All rights reserved,
粵ICP備17059708號
前言 最近，明學是一個火熱的話題，而我，卻也想當那么一回明學家，那就是，把JavaScript和多線程并發(fā)這兩個八竿子打不找的東西，給硬湊了起來
，還寫了一個并發(fā)庫concurrent-thread-js
<https://github.com/penghuwan/concurrent-thread.js>
。尷尬的是，當我發(fā)現(xiàn)其中的不合理之處，即這個東東的應(yīng)用場景究竟是什么時，我發(fā)現(xiàn)我已經(jīng)把代碼寫完了。 ? ? ? ??注意！
本文中的線程指的都是用JS異步函數(shù)模擬的“假線程”，不是真正意義上的多線程，請不要誤解?? ? ?
github地址
https://github.com/penghuwan/concurrent-thread.js
<https://github.com/penghuwan/concurrent-thread.js>
<https://github.com/penghuwan/concurrent-thread.js>
本文的目的
事實上，這個庫用處很小，但是在寫的過程中，我對Promise,Async函數(shù)以及event事件流的使用產(chǎn)生了新的認識，同時也逐漸去學習和了解怎么去從零開始去寫一個非業(yè)務(wù)的，通用的npm模塊，所以希望拿出來和大家分享一下，這才是本文的真正的目的。
? 好，我們從一個故事開始。 場景一 場景二 ?
github地址
https://github.com/penghuwan/concurrent-thread.js?github.com
注意！倘若不考慮webworker這種解決方案，我們一般都認為JS是單線程的。
concurrent-thread-js功能簡介
為單線程的JavaScript實現(xiàn)并發(fā)協(xié)調(diào)的功能，語意，命名和作用性質(zhì)上參考Java的實現(xiàn)，提sleep/join/interupt等API以及鎖和條件變量等內(nèi)容，并提供線程間通信的功能，依賴ES6語法，基于Promise和Async函數(shù)實現(xiàn)，故需要Babel編譯才能運行。JavaScrpt本來就是單線程的，所以這只是在API的層面實現(xiàn)了模擬，
在下文的介紹中，每條所謂的線程其實就是普通的異步函數(shù)，并在此基礎(chǔ)上實現(xiàn)不同線程的協(xié)調(diào)配合。 ?
為什么不選用webworker實現(xiàn)？
沒錯，一般來說JS中模擬多線程我們也許會選用webworker，但是它必須要求你手動創(chuàng)建額外的webworker腳本文件，并通過new
work('work.js')這種方式使用，這并不能達到我項目中想要的API的效果，而且注意：webwork中的環(huán)境不是window!很多方法你調(diào)用不了的。你只能采取這種方案，也即在主線程完成該功能，這是我沒有選擇webworker的另一個原因。
說是這樣說，但其實在大多數(shù)時候還是用webworker就夠了 ?
什么時候使用concurrent-thread-js
這個問題真是靈魂拷問，可是既然代碼寫都寫了，我怎么也得編一個理由出來?。☆~。。。讓我想想哈 它的作用是
：當JS工程需要讓兩個函數(shù)在執(zhí)行上不互相干擾，同時也不希望它們會阻塞主線程，與此同時，還希望這兩個函數(shù)實現(xiàn)類似并發(fā)多線程之間的協(xié)調(diào)的需求的時候，你可以使用這個并發(fā)模擬庫，
實際上這種應(yīng)用場景。。。這尼瑪有這種應(yīng)用場景嗎？?。ㄔ牧搜剑?。 ?
API總覽
* submit(function,[namespace]): 接收一個函數(shù)，普通函數(shù)或Async函數(shù)均可，并異步執(zhí)行"線程"
* sleep(ms): "線程"休眠,可指定休眠時間ms,以毫秒計算
* join(threadName): "線程"同步，調(diào)用此方法的"線程"函數(shù)將在threadName執(zhí)行結(jié)束后繼續(xù)執(zhí)行
* interupt(threadName): "線程"中斷，影響"線程"內(nèi)部調(diào)this.isInterrupted()的返回值
* Lock.lock:
加鎖，一個時刻只能有一個"線程"函數(shù)進入臨界區(qū)，其他"線程"函數(shù)需要等待，鎖是非公平的，也就是說后面排隊的線程函數(shù)沒有先后，以隨機的方式進行競爭。
* Lock.unlock:解除非公平鎖
* Condition.wait:不具備執(zhí)行條件，"線程"進入waiting狀態(tài)，等待被喚醒
* Condition.notify:隨機喚醒一個wait的"線程"
* Condition.notifyAll: 尚未編寫，喚醒所有wait的"線程"
* getState: 還沒寫完 獲取"線程"狀態(tài)，包括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中模擬的異步函數(shù)
A1.submit方法
submit模擬提交線程至線程池 // 備注：為循序漸進介紹，以下為簡化代碼 // 存儲每個線程函數(shù)的狀態(tài)，例如是否中斷，以及線程狀態(tài)等 const
threadMap = {}; class ThreadPool { // 模擬線程中斷 interrupt(threadName) { } //
模擬線程同步 join(threadName, targetThread) { } // 模擬線程休眠 sleep(ms) { } }; function
submit(func, name) {if (!func instanceof Function) return; //
方式1：傳入一個具名函數(shù)；方式2：傳入第二個參數(shù)，即線程命名空間 const threadName = func.name || name; //
threadMap負責存儲線程狀態(tài)數(shù)據(jù) threadMap[threadName] = { state: RUNNABLE, isInterrupted:
false }; // 讓func異步調(diào)用，同時將傳入函數(shù)的作用域綁定為 ThreadPool原型 Promise.resolve({ then:
func.bind(ThreadPool.prototype); }) }
?
首先，我們做了三件事情：
* 獲取線程函數(shù)的命名空間，并初始化線程初始數(shù)據(jù)，不同線程狀態(tài)由threadMap全局存儲
*
將提交的函數(shù)func作為Promise.resolve方法中的一個thenable對象的then參數(shù)，這相當于立即"完成"一個Promise，同時在then方法中執(zhí)行func，
func會以異步而不是同步的方式進行執(zhí)行，你也可以簡單的理解成類似于執(zhí)行了setTimeOut(func,0)；
*
將func的作用域綁定為新生成的ThreadPool實例，ThreadPool中定義了我們上面我們介紹到的方法，如sleep/join/interupt等，
這有什么好處呢？這意味著我們可以直接在函數(shù)中通過調(diào)用this.interrupt的方式去調(diào)用我們定義的API了
，符合我們的使用習慣（注意,class中定義的除箭頭函數(shù)外的普通函數(shù)實際上都存放在原型中） submit(async function example() {
this.interrupt(); });
?
但問題在于：現(xiàn)在因為所有的函數(shù)通過this調(diào)用的都是ThreadPool原型中的方法，我們要在調(diào)用唯一的interrupt方法，
需要在異步函數(shù)中傳入"線程"標識，如線程名。這顯然不方便，也不優(yōu)雅,例如下面的命名為example的線程函數(shù) submit(async function
example() {this.interrupt('example'); });
?
使用這個模塊用戶會感到奇怪：我明明在example函數(shù)中，為什么還要給調(diào)用方法傳example這個名字參數(shù)？？難道不能在模塊內(nèi)部把這事情干了嗎？
對！我們下面做的就是這件事情，我們編寫一個delegateThreadPool方法，由它為ThreadPool代理處理不同“線程“函數(shù)的函數(shù)名 //
返回代理后的ThreadPool function delegateThreadPool(threadName) { //
threadName為待定的線程名，在submit方法調(diào)用時候傳入 // 代理后的ThreadPool const proxyClass = {}; //
獲取ThreadPool原來的所有的方法,賦給props數(shù)組 var props =
Object.getOwnPropertyNames(ThreadPool.prototype);for (let prop of props) { //
代理ThreadPool，為其所有方法增加threadName這個參數(shù) 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異步調(diào)用，不阻塞主線程，同時實現(xiàn)并發(fā)
Promise.resolve({ then:function () { // 給func綁定this為代理后的ThreadPool對象，以便調(diào)用方法
func.call(proxyScope); } }); }// 調(diào)用this.sleep方法時，已經(jīng)無需增加函數(shù)命名作為參數(shù)了 submit(async
function example() { this.interrupt(); });
?
也就是說，我們的線程函數(shù)func綁定的已經(jīng)不是ThreadPool.prototype了，而是delegateThreadPool處理后返回的對象：proxyScope。這時候，我們在“線程”函數(shù)體里調(diào)用this.interrupt方法時，已經(jīng)無需增加函數(shù)命名作為參數(shù)了，因為這個工作，proxyScope對象幫我們做了，其實它的工作很簡單——就是它的每個函數(shù)，都在一個返回的閉包里面調(diào)用ThreadPool的同名函數(shù)，并傳遞線程名作為第一個參數(shù)。
A2. sleep方法
作用：線程休眠
sleep方法很簡單,無非就是返回一個Promise實例，在Promise的函數(shù)里面調(diào)setTimeOut,等時間到了執(zhí)行resolve函數(shù),這段時間里修飾Promise的await語句會阻塞一段時間，resolve后又await語句又繼續(xù)向下執(zhí)行了，能滿足我們想要的休眠效果
// 模擬“線程”休眠 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里，你不能通過調(diào)用terminate方法停掉一個線程，因為這有可能會因為處理邏輯突然中斷而導致數(shù)據(jù)不一致的問題，所以要通過interrupt方法把一個中斷標志位置為true，然后通過isInterrupted方法作為判斷條件跳出關(guān)鍵代碼。
所以為了模擬，我在JS中處理“線程”中斷也是這么去做的,但是我們這樣做的根本原因是：我們壓根沒有可以停掉一個線程函數(shù)的方法！（JAVA是有但是不準用，即廢棄了而已）
// 模擬線程中斷 interrupt(threadName) { if (!threadName) { throw new Error('Miss
function parameters') } if (threadMap[threadName]) {
threadMap[threadName].isInterrupted= true; } } // 獲取線程中斷狀態(tài)
isInterrupted(threadName) {if (!threadName) { throw new Error('Miss function
parameters') } // !!的作用是：將undefined轉(zhuǎn)為false return !!
threadMap[threadName].isInterrupted; }
?
A4. join方法 join(threadName): "線程"同步，調(diào)用此方法的"線程"函數(shù)將在threadName執(zhí)行結(jié)束后繼續(xù)執(zhí)行
join方法和上面的sleep方法是一樣的道理，我們讓它返回一個Promise，只要我們不調(diào)resolve，那么外部修飾Promise的await語句就會一直暫停，等到j(luò)oin的那個另一個線程執(zhí)行完了，我們看準時機！把這個Promise給resolve,這時候外部修飾Promise的await語句不就又可以向下執(zhí)行了嗎？
? ? 但問題在于：我們?nèi)绾螌崿F(xiàn)這個“一個函數(shù)執(zhí)行完通知另一個函數(shù)的功能呢”？沒錯！那就是我們JavaScript最喜歡的套路: 事件流！
我們下面使用event-emitter這個前后端通用的模塊實現(xiàn)事件流。
我們只要在任何一個函數(shù)結(jié)束的時候觸發(fā)結(jié)束事件（join-finished），同時傳遞該線程的函數(shù)名作為參數(shù)，然后在join方法內(nèi)部監(jiān)聽該事件，并在響應(yīng)時候調(diào)用resolve方法不就可以了嘛。
? 首先是在join方法內(nèi)部監(jiān)聽線程函數(shù)的結(jié)束事件 import ee from 'event-emitter'; const emitter = ee();
// 模擬線程同步 join(threadName, targetThread) { return new Promise((resolve) => { //
監(jiān)聽其他線程函數(shù)的結(jié)束事件 emitter.on('join-finished', (finishThread) => { //
根據(jù)結(jié)束線程的線程名finishThread做判斷 if (finishThread === targetThread) { resolve(); } })
}) }
?
同時在線程函數(shù)執(zhí)行結(jié)束時觸發(fā)join-finished事件，傳遞線程名做參數(shù) import ee from 'event-emitter'; const
emitter= ee(); function submit(func, name) { // ... Promise.resolve({ then:
func().then(()=> { emitter.emit('join-finished', threadName); }) }); } 使用如下：
submit(asyncfunction thread1 () { this.join('thread2'); console.log(1); });
submit(asyncfunction thread2 () { this.sleep(3000); console.log(2) }) //
3s后，依次輸出 2 1
A5. Lock.lock & Lock.unlock（非公平鎖）
我們主要是要編寫兩個方法：lock和unlock方法。我們需要設(shè)置一個Boolean屬性isLock
* lock方法：
lock方法首先會判斷isLock是否為false，如果是，則代表沒有線程占領(lǐng)臨界區(qū)，那么允許該線程進入臨界區(qū)，同時把isLock設(shè)置為true，不允許其他線程函數(shù)進入。其他線程進入時，由于判斷isLock為true，會setTimeOut每隔一段時間遞歸調(diào)用判斷isLock是否為false，從而以較低性能消耗的方式模擬while死循環(huán)。當它們檢測到isLock為false時候，則會進入臨界區(qū)，同時設(shè)置isLock為true。因為后面的線程沒有先后順序，所以這是一個非公平鎖
* unLock方法：unlock則是把isLock屬性設(shè)置為false,解除鎖定就可以了 // 這是一個非公平鎖 class Lock {
constructor() {this.isLock = false; } //加鎖 lock() { if (this.isLock) { const
self= this; // 循環(huán)while死循環(huán)，不停測試isLock是否等于false return new Promise((resolve) => {
(function recursion() { if (!self.isLock) { // 占用鎖 self.isLock = true; //
使外部await語句繼續(xù)往下執(zhí)行 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(asyncfunction 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:不具備執(zhí)行條件，線程進入waiting狀態(tài)，等待被喚醒
* Condition.notify: 喚醒線程 對不起！寫到這里，我實在是口干舌燥，寫不下去了，但是道理和前面是一樣的： 無非是：事件監(jiān)聽 +
Promise + Async函數(shù)組合拳，一套搞定 import ee from 'event-emitter'; const ev = ee();
class Condition { constructor() {this.n = 0; this.list = []; } //
當不滿足條件時，讓線程處于等待狀態(tài) 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); // 讓外部函數(shù)恢復執(zhí)行 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
?
最后的大總結(jié)
其實說到底，我想和大家分享的不是什么并發(fā)啊，什么多線程啦。 其實我想表達的是：事件監(jiān)聽 + Promise + Async函數(shù)這套組合拳很好用啊
* 你想讓一段代碼停一下？OK!寫個返回Promise的函數(shù)，用await修飾，它就停啦！
* 你想控制它(await)不要停了，繼續(xù)往下走？OK! 把Promise給resolve掉，它就往下走啦
* 你說你不知道怎么控制它停，因為監(jiān)聽和發(fā)射事件的代碼分布在兩個地方？OK！那就使用事件流 ?
本文完，下面是全部項目代碼（剛寫了文章才發(fā)現(xiàn)有bug，待會改改）
熱門工具 換一換