柚子快報邀請碼778899分享：前端并發(fā)控制

http://yzkb.51969.com/

本文講解Promise，callback，RxJS多種方式實現(xiàn)并發(fā)限制

1.Promise

目前來說，Promise是最通用的方案，一般我們最先想到Promise.all，當然最好是使用新出的Promise.allsettled。

下面簡單介紹下二者的區(qū)別，假如存在某個請求失敗時，all會整體失敗，而allsettled只會讓單個請求失敗，對于大部分情況來說，allsettled的是更好的選擇，因為allsettled更為靈活，一般來說面對這種情況，總共有三種處理方式，如下所示，all只能支持第一種，而allsettled三種都支持：

整體失敗 最終結果，過濾失敗的選項 將單個失敗的保留，并渲染到UI中

方法1 全部并發(fā)

直接使用Promise.all是最簡單的，代碼如下，然后all并沒有并發(fā)控制能力，一瞬間會將全部請求發(fā)出，從而造成前面提到的瀏覽器卡頓問題。

這里get函數(shù)我們使用setTimeout+隨機時間來模擬請求，其返回promise實例。

function?gets(ids,?max)?{

??return?Promise.all(ids.map(id?=>?get(id)))

}

function?get(id)?{

??return?new?Promise((resolve)?=>?{

????setTimeout(()?=>?{?resolve({?id?})?},?Math.ceil(Math.random()?*?5))

??});

}

方法2 分批并發(fā)

你可能會想到一種分批發(fā)送的辦法，將請求按max數(shù)量分成N個組，每組并行發(fā)送，這需要結合遞歸和Promise.all，示例代碼如下：

function?gets(ids,?max)?{

??let?index?=?0;

??const?result?=?[];

??function?nextBatch()?{

????const?batch?=?ids.slice(index,?index?+?max);

????index?+=?max;

????return?Promise.all(batch.map(get)).then((res)?=>?{

??????result.push(...res);

??????if?(index?

????????return?nextBatch();

??????}

??????return?result;

????});

??}

??return?nextBatch();

}

這種方法的優(yōu)勢在于實現(xiàn)相對簡單，容易理解。但是它的缺點是，每一批請求中的最慢的請求會決定整個批次的完成時間，這可能會導致一些批次的其他請求早早完成后需要等待，從而降低整體的并發(fā)效率。

這種方法在業(yè)務中是不太能接受的，面試中的話，也只能勉強及格。

方法3 限制并發(fā)

一個更高效的思路是使用異步并發(fā)控制，而不是簡單的批處理。這種方法可以在任何時刻都保持最大數(shù)量的并發(fā)請求，而不需要等待整個批次完成。這需要我們維護一個請求池，在每個請求完成時，將下一個請求添加到請求池中，示例代碼如下：

gets函數(shù)返回一個promise，在請求全部完成后，promise變?yōu)閒ulfilled狀態(tài)；內(nèi)部采用遞歸，每個請求成功和失敗后，發(fā)送下一個請求；在最下面先發(fā)送max個請求到請求池中。

function?gets(ids,?max)?{

??return?new?Promise((resolve)?=>?{

????const?res?=?[];

????let?loadcount?=?0;

????let?curIndex?=?0;

????function?load(id,?index)?{

??????return?get(id).then(

????????(data)?=>?{

??????????loadcount++;

??????????if?(loadcount?===?ids.length)?{

????????????res[index]?=?data;

????????????resolve(res);

??????????}?else?{

????????????curIndex++;

????????????load(ids[curIndex]);

??????????}

????????},

????????(err)?=>?{

??????????res[index]?=?err;

??????????loadcount++;

??????????curIndex++;

??????????load(ids[curIndex]);

????????}

??????);

????}

????for?(let?i?=?0;?i?

??????curIndex?=?i;

??????load(ids[i],?i);

????}

??});

}

2.callback

在Promise之前，js中的異步都是基于回調(diào)函數(shù)的，比如 jQuery 的 ajax，Node.js 中的 http 模塊等。

茴字有多種寫法，下面我們挑戰(zhàn)一下使用callback來解決這個問題。下面我們先把get函數(shù)改造一下，基于回調(diào)函數(shù)的get如下所示：

function?get(id,?success,?error)?{

??setTimeout(()?=>?success({?id?}),?Math.ceil(Math.random()?*?5))

}

gets函數(shù)的接口也要改成回調(diào)函數(shù)，如下所示：

function?gets(ids,?max,?success,?error)?{}

回調(diào)函數(shù)也是基于上面的思路，把上面的代碼稍加改動即可，將其中的Promise換成callback，示例如下：

還記得前面讓你想其他思路嗎，還有一種結合遞歸和異步函數(shù)的方法，在Promise下會比這種方法更簡單，但其實還是這個思路更好，Promise和callback都可以使用。

function?gets(ids,?max,?success,?error)?{

??const?res?=?[];

??let?loadcount?=?0;

??let?curIndex?=?0;

??function?load(id,?index)?{

????return?get(

??????id,

??????(data)?=>?{

????????loadcount++;

????????if?(loadcount?===?ids.length)?{

??????????res[index]?=?data;

??????????success(res);

????????}?else?{

??????????curIndex++;

??????????load(ids[curIndex]);

????????}

??????},

??????(err)?=>?{

????????res[index]?=?err;

????????loadcount++;

????????curIndex++;

????????load(ids[curIndex]);

??????}

????);

??}

??for?(let?i?=?0;?i?

????curIndex?=?i;

????load(ids[i],?i);

??}

}

?

3.RxJS?

看看RxJS，絕大部分人都不太了解RxJS，RxJS號稱異步編程的lodash，對于這個問題，其代碼實現(xiàn)會非常簡單。

?

RxJS 是一個用于處理異步數(shù)據(jù)流的 JavaScript 庫，它通過「可觀察對象」（Observable）來代表隨時間推移發(fā)出值的數(shù)據(jù)流。你可以使用一系列操作符（如?map、filter、merge?等）來處理這些數(shù)據(jù)流，并通過「訂閱」（subscribe）來觀察并執(zhí)行相關操作。RxJS 使得處理復雜的異步邏輯變得簡單而優(yōu)雅，特別適合于實現(xiàn)并發(fā)控制等場景。

上面是RxJS的簡介，相信看完了還是不理解，RxJS其實是比較難學的，建議大家閱讀其他擴展資料。

下面先用RxJS改造我們的get函數(shù)，改造完如下所示，這需要用到Observable和observer，這些都是RxJS的概念，即便不知道其含義，看代碼和Promise是比較相似的。

import { Observable } from 'RxJS';

function get(id) {

return new Observable((observer) => {

setTimeout(() => {

observer.next({ id });

observer.complete();

}, Math.ceil(Math.random() * 5));

});

}

下面我們參考Promise中的思路，依次看看在RxJS中如何實現(xiàn)。

方法1 全部并發(fā)

在RxJS中和Promise.all類似的功能是forkJoin，這種方法最簡單，代碼如下所示，和Promise.all類似，這并不滿足我們的需求。

import?{?forkJoin?}?from?'RxJS';

function?gets(ids)?{

??const?observables?=?ids.map(get);

??return?forkJoin(observables);

}

方法2 分批并發(fā)

下面來看下如何實現(xiàn)分批并發(fā)，在Promise中我們使用遞歸+Promise.all來實現(xiàn)的。

在RxJS中，我們使用concatMap操作符來確保這些組是依次處理的，而不是同時處理。在處理每個組時，我們使用forkJoin來并行處理組內(nèi)的所有請求。最后，我們使用reduce操作符來將所有組的結果合并成一個一維數(shù)組。

如果不理解RxJS，我們單純看代碼，可以看到RxJS代碼的表現(xiàn)性更強，通過語義化的操作符串聯(lián)，就完成了Promise中很多命令式的代碼。

import?{?from,?forkJoin?}?from?'RxJS';

import?{?concatMap,?reduce?}?from?'RxJS/operators';

function?gets(ids,?max)?{

??//?將ids按max分組

??const?groups?=?[];

??for?(let?i?=?0;?i?

????groups.push(ids.slice(i,?i?+?max));

??}

??//?使用concatMap控制組之間的串行執(zhí)行，并在每一組內(nèi)使用forkJoin實現(xiàn)并行請求

??//?使用reduce來收集和合并所有組的結果

??return?from(groups).pipe(

????concatMap((group)?=>?forkJoin(group.map(get))),

????reduce((acc,?results)?=>?acc.concat(results),?[])

??);

}

方法3 限制并發(fā)

最后我們來看看RxJS如何實現(xiàn)限制并發(fā)，在這個實現(xiàn)中，我們使用mergeMap來控制并發(fā)，并使用一個Map對象來存儲每個請求的結果，其中鍵是ID，值是請求結果。這樣，我們可以在所有請求完成后，按照原始ID數(shù)組的順序從Map中提取結果。

示例代碼如下，控制并發(fā)是RxJS支持的功能，實現(xiàn)就是一個參數(shù)，非常簡單

function?gets(ids,?max)?{

??return?from(ids).pipe(

????mergeMap((id)?=>?get(id).pipe(

??????map(result?=>?({?id,?result?}))

????),?max),

????reduce((acc,?{?id,?result?})?=>?acc.set(id,?result),?new?Map()),

????map(resMap?=>?ids.map(id?=>?resMap.get(id)))

??);

}

?總結

我們探討了使用Promise，callback和RxJS的方式實現(xiàn)并發(fā)限制，每種方式中又介紹了三種代碼思路，包括全部并發(fā)、分批并發(fā)以及限制并發(fā)。每種方法都有其適用場景和優(yōu)缺點：

「全部并發(fā)」適用于需要將請求分批次處理的場景，簡單易懂，但可能不是最高效的方法。 「分批并發(fā)」在保持一定并發(fā)度的同時，避免同時發(fā)出過多的請求，適用于需要控制資源消耗的場景。 「限制并發(fā)」則結合了并發(fā)的高效性和結果順序的一致性，適用于對結果順序有要求的并發(fā)請求處理。

通過選擇合適的方法，我們可以在保證性能的同時，滿足不同場景下對并發(fā)控制的需求。

再次給大家安利RxJS，RxJS作為一個強大的響應式編程庫，為我們提供了靈活而強大的工具來處理這些復雜的異步邏輯。

文章的最后，我想引申下請求層的概念，在實際項目中，請求層的設計和實現(xiàn)對整個應用的性能和穩(wěn)定性至關重要。一個健壯的請求層不僅能夠處理基本的數(shù)據(jù)請求和響應，還能夠應對各種復雜的網(wǎng)絡環(huán)境和業(yè)務需求。以下是請求層可以處理的一些常見問題：

「失敗和錯誤處理」：優(yōu)雅地處理請求失敗和服務器返回的錯誤，提升用戶體驗。 「失敗重試」：在請求失敗時自動重試，增加請求的成功率。 「接口降級」：在服務不可用時，提供備選方案，保證應用的基本功能。 「模擬接口」：在后端服務尚未開發(fā)完成時，模擬接口響應，加速前端開發(fā)。 「模擬列表接口」：模擬分頁、排序等列表操作，方便前端調(diào)試和測試。 「接口聚合和競態(tài)」：合并多個接口請求，減少網(wǎng)絡開銷；處理接口請求的競態(tài)問題，確保數(shù)據(jù)的一致性。 「邏輯聚合」：將多個資源的創(chuàng)建和更新等操作聚合成一個請求，簡化前端邏輯。 「控制并發(fā)數(shù)量」：限制同時進行的請求數(shù)量，避免過度消耗資源。 「前端分頁」：在前端進行數(shù)據(jù)分頁，減輕后端壓力。 「超時設置」：為每個請求設置超時時間，防止長時間等待。

通過在請求層中實現(xiàn)這些功能，我們可以使得前端應用更加穩(wěn)定和可靠，同時也提升了用戶的體驗。因此，「加強請求層的建設」是每個前端項目都應該重視的一個方面。

柚子快報邀請碼778899分享：前端并發(fā)控制

http://yzkb.51969.com/

相關文章