李軼

（江漢大學(xué) 數(shù)學(xué)與計算機(jī)科學(xué)學(xué)院，湖北 武漢 430056）

基于Node.js的JavaScript并發(fā)控制流框架

李軼

（江漢大學(xué) 數(shù)學(xué)與計算機(jī)科學(xué)學(xué)院，湖北 武漢 430056）

Node.js因其異步I/O的特性，非常適合于服務(wù)器端的JavaScript開發(fā)。然而為實現(xiàn)此環(huán)境下異步I/O的并發(fā)控制，開發(fā)者不得不手工編寫繁瑣的代碼，因而給開發(fā)者造成了障礙。以并發(fā)計數(shù)器為基礎(chǔ)，可以設(shè)計一個并發(fā)控制流框架。該框架以直觀的調(diào)用形式，實現(xiàn)了異步I/O間的并發(fā)控制；其不僅有助于Node.js環(huán)境下的JavaScript開發(fā)，更提高了開發(fā)者的開發(fā)效率。

異步I/O；并發(fā)；同步；并發(fā)計數(shù)器；JavaScript

0 引言

時至今日，JavaScript的應(yīng)用領(lǐng)域早已從前端的Web瀏覽器，延伸到后臺服務(wù)器。作為一個基于V8［1］引擎的服務(wù)器端JavaScript環(huán)境，Node.js［2］從發(fā)布之初就備受矚目。Node.js以V8引擎的輕量、高效、快速為基礎(chǔ)，同時又提供了基于事件驅(qū)動的異步I/O（Asynchronous I/O）特性［3］，使其非常適合于開發(fā)不同規(guī)模的數(shù)據(jù)密集型、實時型分布式應(yīng)用。

1 異步I/O與并發(fā)控制

異步I/O又稱為非阻塞式I/O，它是Node.js具有的獨(dú)特特性。不同于傳統(tǒng)的每客戶一服務(wù)線程的服務(wù)器架構(gòu)，Node.js是單線程腳本環(huán)境。其核心思想是用一個服務(wù)線程應(yīng)對多個客戶請求。當(dāng)服務(wù)進(jìn)程響應(yīng)客戶請求執(zhí)行I/O操作（如網(wǎng)絡(luò)I/O、文件I/O或數(shù)據(jù)庫I/O等）時，異步I/O可避免因服務(wù)線程被I/O操作阻塞而無法響應(yīng)其他用戶請求的情況。因此，異步I/O不僅避免了多服務(wù)線程對服務(wù)器資源的過多占用，又提高了服務(wù)線程的執(zhí)行效率，從而使一個服務(wù)線程就能應(yīng)對大量的并發(fā)客戶請求。

在調(diào)用異步I/O函數(shù)時，通常都需要指定一個或多個回調(diào)函數(shù)作為其參數(shù)。這是因為異步I/O的非阻塞特性，異步函數(shù)在調(diào)用后會立即返回。當(dāng)I/O操作完成后，需要回調(diào)函數(shù)通知調(diào)用者I/O操作完成。此外，回調(diào)函數(shù)可能還帶有若干參數(shù)，其是I/O操作所返回的數(shù)據(jù)。

有時為實現(xiàn)某一特定功能或為提高程序執(zhí)行效率，多個異步I/O之間既需要并發(fā)（concurrency）［4］，又需要在彼此間進(jìn)行同步（synchronization）［5］。例如，有3個異步I/O可并發(fā)執(zhí)行，但同時要求3個異步I/O全部完成后，才能執(zhí)行下一步操作。此時，最通常的辦法是通過設(shè)置并檢查多個標(biāo)志位來實現(xiàn)。假設(shè)3個異步I/O函數(shù)分別為ioA、ioB和ioC。為達(dá)成此目標(biāo)，需要設(shè)置3個I/O標(biāo)志位，分別為：tagA、tagB、tagC，并將其初值設(shè)為false。此后，在每個I/O函數(shù)的異步回調(diào)中，再將該標(biāo)識位置設(shè)為true，并檢查其余標(biāo)志位。若所有標(biāo)志位全為true，則表明并發(fā)全部完成，可進(jìn)行下一步操作。例程如下：

由此可以看出，這種復(fù)雜繁瑣的代碼，雖然在執(zhí)行邏輯上正確，但是在程序編寫，代碼可讀性及程序調(diào)試上存在諸多不便。特別是在開發(fā)較為復(fù)雜邏輯的系統(tǒng)時，隨著標(biāo)志位的增加，復(fù)雜繁瑣的代碼已經(jīng)成為開發(fā)者的阻礙。因此需要一種方法，將復(fù)雜的代碼簡化；也就是說需要一種并發(fā)控制流框架，實現(xiàn)異步I/O間的并發(fā)控制。

2 任務(wù)對象與并發(fā)任務(wù)緩沖區(qū)

為解決上述問題，可考慮應(yīng)用并發(fā)計數(shù)器機(jī)制。每個異步I/O的調(diào)用，可封裝成為一個任務(wù)對象；依據(jù)任務(wù)對象的數(shù)目，設(shè)置并發(fā)計數(shù)器的初值。當(dāng)某個任務(wù)完成時（即當(dāng)其I/O函數(shù)的回調(diào)函數(shù)被調(diào)用時）將并發(fā)計數(shù)器值減1；若并發(fā)計數(shù)器的值減1后為0，則表明所有并發(fā)任務(wù)均執(zhí)行完畢。如此，多個異步I/O間的并發(fā)控制就可抽象為任務(wù)及并發(fā)任務(wù)計數(shù)器的相關(guān)操作。

對異步I/O任務(wù)而言，其具體屬性包括：任務(wù)名、任務(wù)函數(shù)和任務(wù)參數(shù)；此外，還需要為多個并發(fā)任務(wù)建立一個并發(fā)緩沖區(qū)，其狀態(tài)圖如圖1所示。并發(fā)緩沖區(qū)創(chuàng)建后為就緒狀態(tài)，當(dāng)需要同步的多個并發(fā)任務(wù)加入緩沖區(qū)后，就可啟動并發(fā)同步過程。由于任務(wù)函數(shù)中包含了異步調(diào)用，任務(wù)函數(shù)執(zhí)行完后會立即返回，但此時并不能認(rèn)為該任務(wù)完成，而是在未來某個時刻，在該任務(wù)中的異步回調(diào)得到執(zhí)行后，才可認(rèn)為該任務(wù)執(zhí)行完畢。在該任務(wù)完成后，并發(fā)計數(shù)器的值會自動減1，并檢查其值。若為0則表明所有并發(fā)任務(wù)均執(zhí)行完畢，緩沖區(qū)返回就緒態(tài)。此外，當(dāng)某任務(wù)執(zhí)行出錯時，緩沖區(qū)進(jìn)入終止?fàn)顟B(tài)；當(dāng)某任務(wù)超時時，緩沖區(qū)進(jìn)入超時狀態(tài)。在此兩種狀態(tài)下，都可重新將緩沖區(qū)復(fù)位為就緒態(tài)。

圖1 并發(fā)任務(wù)緩沖區(qū)狀態(tài)圖Fig.1 State chart of concurrent task buffer

3 基于計數(shù)器的并發(fā)控制流框架

以上述討論作為基礎(chǔ)，可構(gòu)造一個以并發(fā)計數(shù)器為核心的并發(fā)控制流框架。該框架以Node.js的模塊形式為用戶提供調(diào)用接口，模塊命名為“concurrentBuf”。另外，由于Node.js為單線程腳本環(huán)境，因此也無需考慮計數(shù)器操作的線程安全問題。

3.1 模塊接口規(guī)范（Module interface specifications）

3.1.1 create函數(shù) 模塊的唯一調(diào)用接口為函數(shù)“create”，該函數(shù)用于創(chuàng)建一個新的并發(fā)緩沖區(qū)對象。其函數(shù)聲明為：function create（name，timeoutMs）。

其中參數(shù)name為并發(fā)緩沖區(qū)的名稱；timeoutMs為并發(fā)任務(wù)超時毫秒數(shù)，即在所有任務(wù)中，最長的任務(wù)耗時不得超過timeoutMs的毫秒數(shù)；否則認(rèn)為并發(fā)同步超時。

3.1.2 并發(fā)任務(wù)緩沖區(qū)對象的方法 并發(fā)任務(wù)緩沖區(qū)對象有如下方法。

1）push方法

該方法將一個并發(fā)任務(wù)放入緩沖區(qū)。

其方法聲明為：push（name，task，argsAry）。

其中參數(shù)name為并發(fā)任務(wù)名；task為任務(wù)函數(shù)；argsAry為函數(shù)task的參數(shù)數(shù)組，其中每個元素為一個參數(shù)對象。參數(shù)對象具有屬性type和屬性value；其中type屬性為字符串類型，其用于指示參數(shù)的類型。特別需要注意的是，參數(shù)的類型除包括JavaScript標(biāo)準(zhǔn)類型之外，還包括類型“callback”，其表示該參數(shù)為一回調(diào)函數(shù)；屬性value則用來存儲實際的參數(shù)值。

2）reset方法

該方法將并發(fā)任務(wù)緩沖區(qū)重新復(fù)位，無需任何參數(shù)。

3）start方法

該方法啟動并發(fā)緩沖區(qū)中的所有任務(wù)，無需任何參數(shù)。

4）onError方法

其聲明格式為：onError（errorCb）。

該方法用于指定當(dāng)某個并發(fā)任務(wù)發(fā)生執(zhí)行錯誤時的事件處理函數(shù)，通過參數(shù)errorCB指定。該方法返回并發(fā)緩沖區(qū)本身的引用。錯誤事件處理函數(shù)errorCb的函數(shù)聲明為：function（taskName，e）。其中參數(shù)taskName為發(fā)生錯誤的任務(wù)名；參數(shù)e為錯誤對象。

5）onComplete方法

其聲明格式為：onComplete（cb）。

該方法用于指定并發(fā)任務(wù)全部完成時的事件處理函數(shù)，通過參數(shù)cb指定。

6）onTimeout方法

其聲明格式為：onTimeout（cb）。

該方法用于指定并發(fā)任務(wù)超時的事件處理函數(shù)，通過參數(shù)cb指定。

3.1.3 并發(fā)任務(wù)緩沖區(qū)對象的事件 并發(fā)任務(wù)緩沖區(qū)對象有如下事件。

1）error事件

當(dāng)某個并發(fā)任務(wù)執(zhí)行中發(fā)生錯誤時，該事件被觸發(fā)。該事件的處理函數(shù)聲明為：function（taskName，e），由緩沖區(qū)對象的onError方法指定。

其中參數(shù)taskName為發(fā)生錯誤的任務(wù)名，參數(shù)e為錯誤對象，其具有的屬性包括：code，message和stack；其中code屬性為錯誤代碼；message屬性為出錯信息字符串；stack屬性為出錯時的函數(shù)調(diào)用棧。

2）complete事件

當(dāng)所有并發(fā)任務(wù)完成時，該事件被觸發(fā)。該事件處理函數(shù)由緩沖區(qū)對象的onComplete方法進(jìn)行指定，該事件無其他參數(shù)。

3）timeout事件

當(dāng)并發(fā)任務(wù)同步超時時，該事件被觸發(fā)。該事件處理函數(shù)由緩沖區(qū)對象的onComplete方法進(jìn)行指定，該事件無其他參數(shù)。

3.2 模塊實現(xiàn)

3.2.1 create函數(shù) 函數(shù)create用于創(chuàng)建一個并發(fā)緩沖對象，此函數(shù)也是模塊對外暴露的唯一接口函數(shù)。其函數(shù)聲明為：function create（name，timeoutMs）。其中參數(shù)name為并發(fā)緩沖區(qū)名稱；timeoutMs為任務(wù)超時毫秒數(shù)。函數(shù)的返回值，即為并發(fā)緩沖區(qū)對象，其具有的方法和事件已經(jīng)在上述模塊接口規(guī)范中進(jìn)行了說明。

3.2.2 并發(fā)任務(wù)計數(shù)器 本框架使用create函數(shù)的局部變量pCunter存儲引用計數(shù)值。由于閉包（closure）［6］的作用，pCunter受到保護(hù)，防止了其從外部被訪問，并具有和緩沖區(qū)對象相同的生存期。

3.2.3 并發(fā)任務(wù)存儲 本框架使用create函數(shù)的局部變量pBuffer存儲任務(wù)隊列。pBuffer為一數(shù)組，其中的每個成員，都是一個任務(wù)對象。任務(wù)對象具有3個屬性：fName、fn和argary。其中fName為任務(wù)名；fn為任務(wù)函數(shù)；argary為任務(wù)函數(shù)參數(shù)值數(shù)組。

同樣由于閉包的作用，pBuffer受到保護(hù)，防止了其從外部被訪問，并具有和緩沖區(qū)對象相同的生存期。

3.2.4 并發(fā)任務(wù)緩沖區(qū)狀態(tài) 本框架使用模塊全局對象STATES，枚舉緩沖區(qū)的所有狀態(tài)。緩沖區(qū)對象的狀態(tài)轉(zhuǎn)換，如圖2所示。對象STATES包含以下屬性：ready、running、terminated和expired。屬性ready表示緩沖區(qū)就緒，其值為0；屬性running表示并發(fā)任務(wù)運(yùn)行中，其值為1；屬性terminated表示緩沖區(qū)因某任務(wù)執(zhí)行異常而終止，其值為2；屬性expired表示任務(wù)執(zhí)行超時導(dǎo)致緩沖區(qū)并發(fā)同步超時，其值為3。

圖2 并發(fā)任務(wù)緩沖區(qū)對象狀態(tài)圖Fig.2 Object state chart of concurrent task buffer

對于緩沖區(qū)狀態(tài)的保存，則由模塊函數(shù)create的局部變量pState存儲。

3.2.5 并發(fā)任務(wù)的加入 并發(fā)任務(wù)的加入由緩沖區(qū)對象的push方法完成，具體描述已在模塊接口定義規(guī)范中進(jìn)行了說明，執(zhí)行流程較為簡單。push方法首先判斷緩沖區(qū)的當(dāng)前狀態(tài)，若不處于就緒態(tài)，則方法直接返回；否則，push方法將任務(wù)名、任務(wù)函數(shù)以及任務(wù)參數(shù)數(shù)組封裝成為一個任務(wù)對象，并將之追加到pBuffer數(shù)組中。

3.2.6 任務(wù)對象的封裝 如上所述，push方法會將任務(wù)名、任務(wù)函數(shù)以及任務(wù)參數(shù)數(shù)組封裝成為一任務(wù)對象。在對任務(wù)進(jìn)行封裝時，有兩個需要注意的地方。

首先，為捕獲任務(wù)函數(shù)中可能發(fā)生的執(zhí)行異常，需要對任務(wù)函數(shù)進(jìn)行二次封裝。即在原任務(wù)函數(shù)的基礎(chǔ)上構(gòu)造出一個新的函數(shù)，如圖3所示。此函數(shù)通過try…catch語句調(diào)用原任務(wù)函數(shù)，若發(fā)生異常，則在catch子句中終止超時計時器，并改變緩沖區(qū)狀態(tài)為終止態(tài)（STATUS.terminated），然后觸發(fā)緩沖區(qū)的error事件，并將捕獲的錯誤對象交由用戶進(jìn)行處理。

其次，為保證并發(fā)引用計數(shù)器在任務(wù)完成后能自動減1，還需要對任務(wù)參數(shù)數(shù)組中的回調(diào)函數(shù)進(jìn)行二次封裝。圖4為任務(wù)回調(diào)的二次封裝函數(shù)執(zhí)行流程。二次封裝函數(shù)首先檢查當(dāng)前緩沖區(qū)狀態(tài)；若為運(yùn)行態(tài)（STATUS.running），則將計數(shù)器減1，然后執(zhí)行用戶的任務(wù)回調(diào)函數(shù)。在用戶回調(diào)函數(shù)返回后，檢查計數(shù)器的值，若為0則表示所有并發(fā)任務(wù)均已完成，因此終止超時計時器，并設(shè)置緩沖區(qū)狀態(tài)為就緒態(tài)，最后觸發(fā)緩沖區(qū)complete事件；若當(dāng)前緩沖區(qū)狀態(tài)為非運(yùn)行態(tài)，說明有某任務(wù)執(zhí)行異常或超時，并發(fā)執(zhí)行失敗，因此直接返回即可。

圖3 任務(wù)函數(shù)的二次封裝函數(shù)執(zhí)行流程圖Fig.3 Executive flow chart of the second wrapper function of task function

圖4 任務(wù)回調(diào)的二次封裝函數(shù)執(zhí)行流程Fig.4 Flow chart of the second wrapper function of task callback

由此可見，任務(wù)對象中的任務(wù)函數(shù)及參數(shù)數(shù)組中的回調(diào)函數(shù)都是其原函數(shù)的二次封裝函數(shù)。

3.2.7 并發(fā)任務(wù)的啟動 并發(fā)任務(wù)的啟動由并發(fā)緩沖區(qū)對象的start方法完成。該方法的執(zhí)行流程較為簡單，首先啟動一個任務(wù)超時計時器，并設(shè)置好其超時回調(diào)函數(shù)；然后設(shè)置引用計數(shù)器pCunter的初值為pBuffer數(shù)組長度，并設(shè)置緩沖區(qū)狀態(tài)為運(yùn)行態(tài)。最后，依次調(diào)用pBuffer數(shù)組中每個任務(wù)對象的任務(wù)函數(shù)。

3.2.8 并發(fā)任務(wù)緩沖區(qū)復(fù)位 并發(fā)緩沖區(qū)復(fù)位由并發(fā)緩沖區(qū)對象的reset方法實現(xiàn)，其執(zhí)行流程較為簡單。首先檢查當(dāng)前隊列狀態(tài)是否為終止態(tài)或超時態(tài)。若是，則清空pBuffer數(shù)組并更改狀態(tài)為就緒態(tài)（STATUS.ready）。

3.3 模塊接口導(dǎo)出

本模塊的唯一導(dǎo)出接口為函數(shù)create，通過Node.js的模塊對象module的exports屬性進(jìn)行導(dǎo)出，示例如下：

module.exports=｛"create"：create｝；

4 示例

以一個簡單的實例演示本框架的使用方法。假設(shè)有兩個文件讀取操作，分別為文件A.txt和B.txt，需要并發(fā)控制完成，示例如下：

程序首先通過require函數(shù)，分別獲得并發(fā)同步模塊parallelBuffer和Node.js的文件系統(tǒng)模塊fs。接著調(diào)用parallelBuffer的create函數(shù)，創(chuàng)建一個并發(fā)同步緩沖區(qū)buf；通過指定timeoutMs參數(shù)為5 000設(shè)定任務(wù)的最長超時為5 000 ms；接著調(diào)用其onError方法，onTimeout方法和onComplete方法，分別指定了任務(wù)在出錯、超時和所有任務(wù)完成時的事件處理函數(shù)。

接下來，通過緩沖區(qū)對象的push方法，添加了兩個并發(fā)任務(wù)"read_fileA"及"read_fileB"；其任務(wù)函數(shù)通過Node.js的fs文件系統(tǒng)模塊，分別讀取文件A.txt和文件B.txt的內(nèi)容。任務(wù)函數(shù)的調(diào)用參數(shù)，被封裝為1個數(shù)組；本例中的任務(wù)函數(shù)需要2個參數(shù)，因此其參數(shù)數(shù)組包含2個元素，分別為文件名以及文件讀取完畢的回調(diào)函數(shù)。

最后，程序通過緩沖區(qū)對象的start方法啟動并發(fā)任務(wù)。若所有并發(fā)任務(wù)執(zhí)行無誤，則程序?qū)⒃诳刂婆_輸出“所有任務(wù)并發(fā)同步完成！”。

5 結(jié)語

通過并發(fā)計數(shù)器的應(yīng)用，在Node.js環(huán)境下成功地設(shè)計了一個并發(fā)控制流框架。解決了因多個異步I/O并發(fā)控制所導(dǎo)致的代碼繁瑣、邏輯復(fù)雜等的問題?？蚣茉谠O(shè)計上簡潔易用，語義明確清晰，代碼可讀性強(qiáng)，符合開發(fā)者的一般邏輯；同時又采用純JavaScript實現(xiàn)，不依賴第三方模塊或組件，具有較好的通用性，稍加修改就能適用于各種JavaScript環(huán)境。

（

）

［1］WIKIPEDIA.V8（JavaScript engine）［EB/OL］.（2014-6-23）［2014-9-10］http：//en.wikipedia.org/wiki/V8_（JavaScript_engine）.

［2］WIKIPEDIA.Node.js［EB/OL］.（2014-7-10）［2014-9-10］http：//en.wikipedia.org/wiki/Node.js.

［3］WIKIPEDIA.Asynchronous I/O［EB/OL］.（2014-6-11）［2014-9-10］http：//en.wikipedia.org/wiki/Asynchronous_I/O.

［4］WIKIPEDIA.Concurrency（computer science）［EB/OL］.（2014-10-11）［2014-10-11］http：//en.wikipedia.org/wiki/Concurrency_（computer_science）.

［5］WIKIPEDIA.Synchronization（computer science）［EB/OL］.（2014-10-15）［2014-10-15］http：//en.wikipedia.org/wiki/Synchronization_（computer_science）.

［6］WIKIPEDIA.Closure（computer science）［EB/OL］.（2014-10-17）［2014-10-17］http：//en.wikipedia.org/wiki/Closure_（computer_programming）.

［7］BURNHAM T.JavaScript異步編程：設(shè)計快速響應(yīng)的網(wǎng)絡(luò)應(yīng)用［M］.北京：人民郵電出版社，2013.

（責(zé)任編輯：范建鳳）

JavaScript Concurrent Flow-Control Framework Based on Node.js

LI Yi
（School of Mathematics and Computer Science，Jianghan University，Wuhan 430056，Hubei，China）

Because of the asynchronous characteristic of I/O，Node.js is quite suitable for developing server-side JavaScript applications.To implement concurrent asynchronous I/O control，developers have to write tedious code manually，it becomes a barrier to developers.Based on the concurrent counter，a concurrent flow-control framework can be constructed.The framework achieved concurrent control between multiple asynchronous I/O in a direct viewing way.Thus，it benefits server-side JavaScript developments and also promotes the efficiency of development.

asynchronous I/O；concurrency；synchronization；concurrent counter；JavaScript

TP393.01

A

1673-0143（2015）02-0170-07

10.16389/j.cnki.cn42-1737/n.2015.02.013

2014-10-27

李 軼（1976—），男，實驗員，碩士，研究方向：網(wǎng)絡(luò)管理。