詳解JavaScript如何實現(xiàn)異步并發(fā)任務控制器

更新時間：2023年05月22日 09:55:10 作者：熊的貓

實現(xiàn)一個控制并發(fā)數(shù)的任務隊列?、實現(xiàn)一個異步并發(fā)任務控制器，這已經是非常經典的手寫題目了，因為其中涉及?異步?和?并發(fā)?的內容，所以本文就來講講到底如何實現(xiàn)呢

前言

“ 實現(xiàn)一個控制并發(fā)數(shù)的任務隊列、實現(xiàn)一個異步并發(fā)任務控制器” 等，已經是非常經典的手寫題目了，因為其中涉及異步和 并發(fā) 的內容，在正式開始實現(xiàn)之前我們先來簡單了解一下它們的概念，畢竟只有知道為什么才能更好的實現(xiàn)，而不是單純的去記憶。

異步 & 并發(fā)

異步

單線程的 JavaScript

我們都知道 默認情況 下 JavaScript 是 單線程 的，又或者說 JavaScript 只在一個線程上運行。

【注意】JavaScript 雖然只在一個線程上運行，但不表示 JavaScript 引擎只有一個線程，實際上，JavaScript 引擎有多個線程，單個腳本只能在一個線程上運行（即 主線程），其他線程都是在后臺配合

而 單線程 就意味著，所有任務需要排隊，前一個任務結束，才會執(zhí)行后一個任務，如果前一個任務耗時很長，后一個任務就不得不一直等著。

JavaScript 異步的產生

如果排隊是因為計算量大，CPU 處理不過來，這時候也算合理，但很多時候 CPU 是空閑的，是因為 IO 設備（輸入/輸出設備）很慢（比如 Ajax 操作從網絡讀取數(shù)據(jù)），CPU 不得不等著結果返回，才能繼續(xù)往下執(zhí)行。

JavaScript 語言的設計者意識到，這時主線程完全可以不管 IO 設備，掛起處于等待中的任務，先運行排在后面的任務，等到 IO 設備返回了結果，再回過頭，把掛起的任務繼續(xù)執(zhí)行下去。

在 JavaScript 為了更好的處理異步問題，我們通常都會選擇使用 Promise 或 async/await。

并發(fā)

早期計算機的 CPU 是 單核的，一個 CPU 在 同一時間 只能執(zhí)行 一個進程/線程，當系統(tǒng)中有 多個進程/線程 等待執(zhí)行時，CPU 只能執(zhí)行完一個再執(zhí)行下一個。

而所謂的 并發(fā)，指在同一時刻只能有一條 進程指令 執(zhí)行，但多個 進程指令 被快速的 交替執(zhí)行，那么在宏觀上看就是多個進程同時執(zhí)行的效果，但在微觀上并不是同時執(zhí)行的，只是把時間分成若干段，使多個進程快速交替的執(zhí)行。

實現(xiàn)異步并發(fā)任務控制器

通過上述內容我們已經知道了異步和 并發(fā) 的基本概念，現(xiàn)在開始具體實現(xiàn)吧！

題目如下：

假設現(xiàn)在要發(fā)送多個請求，但要實現(xiàn)并發(fā)控制，即可以通過一個 limit 控制并發(fā)數(shù)，當任務數(shù)量超過對應的 limit 限制的并發(fā)數(shù)時，后續(xù)的任務需要延遲到 正在執(zhí)行中 的任務執(zhí)行完后 再執(zhí)行，并且需要支持動態(tài)添加 額外的異步任務，同時當 最后一個任務 執(zhí)行完成，需要執(zhí)行對應的 callback 函數(shù)。

生成任務集合

// 生成用于測試的任務集合
const tasks = new Array(10).fill(0).map((v, i) => {
    return function task() {
        return new Promise((resolve) => {
            setTimeout(() => {
                resolve(i + 1)
            }, i * 1000);
        })
    }
})

方式一：并發(fā)控制函數(shù) concurrencyControl

核心思路

通過循環(huán)執(zhí)行當前隊列頭部的任務

當前隊列頭部任務執(zhí)行完畢

若是最后一個任務，則執(zhí)行 callback
否則，繼續(xù)執(zhí)行 下一個隊頭任務

// 并發(fā)控制函數(shù)
function concurrencyControl(tasks, limit, callback) {
    const queue = tasks.slice()  // 當前執(zhí)行的任務隊列
    let count = 0 // 已完成的任務數(shù)量
    const runTask = () => {
        while (limit) {
            limit--
            if (queue.length) {
                const task = queue.shift() // 取出當前隊頭任務
                task().then(() => {
                    limit++
                    count++
                    if (count === tasks.length) { // 最后一個任務
                        callback() // 執(zhí)行回調函數(shù)
                    }else{
                        runTask() // 繼續(xù)執(zhí)行下一個任務
                    }
                })
            }
        }
    }
    return runTask
}
// 測試代碼
const sendRequest = concurrencyControl(tasks, 3, (taskId) => {
    console.log(`task ${taskId} finish！`)
})
sendRequest()

不同時間的任務：

相同時間的任務：

方式二：并發(fā)控制器 ConcurrencyControl

方式一 雖然能夠簡單的完成自動化的并發(fā)控制，但是不支持 動態(tài)添加任務 的要求，這就意味著要 保持狀態(tài) 了，并且如果當前執(zhí)行的 promise 任務狀態(tài)為 rejected 時就無法執(zhí)行完全部的任務，因為 task().then 對應的 onreject 的回調沒有被提供，下面我們就可以通過一個 ConcurrencyControl 類來實現(xiàn)。

核心思路

將原本使用到的變量，轉換成對應的實例屬性
新增 addTask() 方法用于動態(tài)添加任務，并且在其內部自動啟動任務執(zhí)行
task().then 替換為 task().finally，目的是當對應的 promise 任務為 reject 狀態(tài)時仍能夠執(zhí)行

class ConcurrencyControl {
    constructor(tasks, limit, callback) {
        this.queue = tasks.slice() // 當前執(zhí)行的任務隊列
        this.tasks = tasks // 原始任務集合
        this.count = 0 // 已完成的任務數(shù)量
        this.limit = limit
        this.callback = callback
    }
    runTask() {
        while (this.limit) {
            this.limit--
            if (this.queue.length) {
                const task = this.queue.shift() // 取出隊頭任務
                task().finally(() => {
                    this.limit++
                    this.count++
                    if (this.count === this.tasks.length) { // 最后一個任務
                        this.callback() // 執(zhí)行回調函數(shù)
                    } else {
                        this.runTask() // 繼續(xù)執(zhí)行下一個任務
                    }
                })
            }
        }
    }
    addTask(task) {
        // 同步添加任務
        this.queue.push(task)
        this.tasks.push(task)
        // 當直接調用 addTask 也可直接執(zhí)行
        this.runTask()
    }
}
// 測試代碼
const Control = new ConcurrencyControl(tasks, 3, () => {
    console.log(`task all finish！`)
})
// 執(zhí)行隊列任務
Control.runTask()
// 添加新任務
Control.addTask(function task() {
    return new Promise((resolve) => {
        setTimeout(() => {
            console.log(`task ${Control.tasks.length} finish！`)
            resolve(Control.tasks.length)
        }, Control.tasks.length * 200);
    })
})

方式三：優(yōu)化并發(fā)控制器 ConcurrencyControl

核心思路

優(yōu)化掉 this.count 計數(shù)，通過 this.queue.size 來代替
優(yōu)化掉 this.addTask() 方法中的 this.queue.push(task)，通過 this.tasks 的變化來自動影響 this.queue 隊列
優(yōu)化掉 this.limit ++/--，通過 this.queue.size < this.limit 來替換

class ConcurrencyControl {
    constructor(tasks, limit, callback) {
        this.tasks = tasks.slice() // 淺拷貝，避免修改原數(shù)據(jù)
        this.queue = new Set() // 任務隊列
        this.limit = limit // 最大并發(fā)數(shù)
        this.callback = callback // 回調
    }
    runTask() {
        // 邊界判斷
        if(this.tasks.length == 0) return
        // 當任務隊列有剩余，繼續(xù)添加任務 
        while (this.queue.size < this.limit) {
            const task = this.tasks.shift() // 取出隊頭任務
            this.queue.add(task) // 往隊列中添加當前執(zhí)行的任務
            task()
                .finally(() => {
                    this.queue.delete(task) // 當前任務執(zhí)行完畢，從隊列中刪除改任務
                    if (this.queue.size == 0) {
                        this.callback() // 執(zhí)行回調函數(shù)
                    } else {
                        this.runTask() // 繼續(xù)執(zhí)行下一個任務
                    }
                })
        }
    }
    addTask(task) {
        // 同步添加任務
        this.tasks.push(task)
        // 當直接調用 addTask 也可直接執(zhí)行
        this.runTask()
    }
}
// 測試代碼
const Control = new ConcurrencyControl(tasks, 3, () => {
    console.log(`task all finish！`)
})
Control.runTask() // 執(zhí)行隊列任務
Control.addTask(function task() { // 添加新任務
    return new Promise((resolve) => {
        setTimeout(() => {
            console.log(`task 9999 finish！`)
            resolve(999)
        }, 100);
    })
})