快捷導(dǎo)航

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

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

實(shí)現(xiàn)一個(gè)控制并發(fā)數(shù)的任務(wù)隊(duì)列?、實(shí)現(xiàn)一個(gè)異步并發(fā)任務(wù)控制器，這已經(jīng)是非常經(jīng)典的手寫題目了，因?yàn)槠渲猩婕?異步?和?并發(fā)?的內(nèi)容，所以本文就來講講到底如何實(shí)現(xiàn)呢

前言

“ 實(shí)現(xiàn)一個(gè)控制并發(fā)數(shù)的任務(wù)隊(duì)列、實(shí)現(xiàn)一個(gè)異步并發(fā)任務(wù)控制器” 等，已經(jīng)是非常經(jīng)典的手寫題目了，因?yàn)槠渲猩婕?nbsp;異步和 并發(fā) 的內(nèi)容，在正式開始實(shí)現(xiàn)之前我們先來簡單了解一下它們的概念，畢竟只有知道為什么才能更好的實(shí)現(xiàn)，而不是單純的去記憶。

異步 & 并發(fā)

異步

單線程的 JavaScript

我們都知道 默認(rèn)情況 下 JavaScript 是 單線程 的，又或者說 JavaScript 只在一個(gè)線程上運(yùn)行。

【注意】JavaScript 雖然只在一個(gè)線程上運(yùn)行，但不表示 JavaScript 引擎只有一個(gè)線程，實(shí)際上，JavaScript 引擎有多個(gè)線程，單個(gè)腳本只能在一個(gè)線程上運(yùn)行（即 主線程），其他線程都是在后臺(tái)配合

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

JavaScript 異步的產(chǎn)生

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

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

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

并發(fā)

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

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

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

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

題目如下：

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

生成任務(wù)集合

// 生成用于測試的任務(wù)集合
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í)行當(dāng)前隊(duì)列頭部的任務(wù)

當(dāng)前隊(duì)列頭部任務(wù)執(zhí)行完畢

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

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

不同時(shí)間的任務(wù)：

相同時(shí)間的任務(wù)：

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

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

核心思路

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

class ConcurrencyControl {
    constructor(tasks, limit, callback) {
        this.queue = tasks.slice() // 當(dāng)前執(zhí)行的任務(wù)隊(duì)列
        this.tasks = tasks // 原始任務(wù)集合
        this.count = 0 // 已完成的任務(wù)數(shù)量
        this.limit = limit
        this.callback = callback
    }
    runTask() {
        while (this.limit) {
            this.limit--
            if (this.queue.length) {
                const task = this.queue.shift() // 取出隊(duì)頭任務(wù)
                task().finally(() => {
                    this.limit++
                    this.count++
                    if (this.count === this.tasks.length) { // 最后一個(gè)任務(wù)
                        this.callback() // 執(zhí)行回調(diào)函數(shù)
                    } else {
                        this.runTask() // 繼續(xù)執(zhí)行下一個(gè)任務(wù)
                    }
                })
            }
        }
    }
    addTask(task) {
        // 同步添加任務(wù)
        this.queue.push(task)
        this.tasks.push(task)
        // 當(dāng)直接調(diào)用 addTask 也可直接執(zhí)行
        this.runTask()
    }
}
// 測試代碼
const Control = new ConcurrencyControl(tasks, 3, () => {
    console.log(`task all finish！`)
})
// 執(zhí)行隊(duì)列任務(wù)
Control.runTask()
// 添加新任務(wù)
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 計(jì)數(shù)，通過 this.queue.size 來代替
優(yōu)化掉 this.addTask() 方法中的 this.queue.push(task)，通過 this.tasks 的變化來自動(dòng)影響 this.queue 隊(duì)列
優(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() // 任務(wù)隊(duì)列
        this.limit = limit // 最大并發(fā)數(shù)
        this.callback = callback // 回調(diào)
    }
    runTask() {
        // 邊界判斷
        if(this.tasks.length == 0) return
        // 當(dāng)任務(wù)隊(duì)列有剩余，繼續(xù)添加任務(wù) 
        while (this.queue.size < this.limit) {
            const task = this.tasks.shift() // 取出隊(duì)頭任務(wù)
            this.queue.add(task) // 往隊(duì)列中添加當(dāng)前執(zhí)行的任務(wù)
            task()
                .finally(() => {
                    this.queue.delete(task) // 當(dāng)前任務(wù)執(zhí)行完畢，從隊(duì)列中刪除改任務(wù)
                    if (this.queue.size == 0) {
                        this.callback() // 執(zhí)行回調(diào)函數(shù)
                    } else {
                        this.runTask() // 繼續(xù)執(zhí)行下一個(gè)任務(wù)
                    }
                })
        }
    }
    addTask(task) {
        // 同步添加任務(wù)
        this.tasks.push(task)
        // 當(dāng)直接調(diào)用 addTask 也可直接執(zhí)行
        this.runTask()
    }
}
// 測試代碼
const Control = new ConcurrencyControl(tasks, 3, () => {
    console.log(`task all finish！`)
})
Control.runTask() // 執(zhí)行隊(duì)列任務(wù)
Control.addTask(function task() { // 添加新任務(wù)
    return new Promise((resolve) => {
        setTimeout(() => {
            console.log(`task 9999 finish！`)
            resolve(999)
        }, 100);
    })
})