快捷導(dǎo)航

JavaScript+TypeScript實(shí)現(xiàn)并發(fā)隊(duì)列的示例

更新時(shí)間：2024年08月23日 11:32:54 作者：知半愚無(wú)

本文主要介紹了JavaScript+TypeScript實(shí)現(xiàn)并發(fā)隊(duì)列的示例,文中通過(guò)示例代碼介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友們下面隨著小編來(lái)一起學(xué)習(xí)學(xué)習(xí)吧

1. 前言

本文使用了 TypeScript 和 JavaScript，可能有的讀者并沒(méi)有學(xué)過(guò) TypeScript，擔(dān)心看不懂。其實(shí)我認(rèn)為有了 TypeScript 你應(yīng)該更容易看懂，因?yàn)?TypeScript 僅僅是繁瑣了一點(diǎn)，因?yàn)樗皇墙o變量加上了類(lèi)型，但是它能增加代碼的可讀性和可維護(hù)性，所以你應(yīng)該能快速理解。

安裝 TypeScript 見(jiàn)文末。

生活中許多同時(shí)發(fā)生的事情，比如：你在打代碼，他在打代碼，她也在打代碼，而我在看你們打代碼。這不是并發(fā)而是并行。

并發(fā)和并行的最大區(qū)別就是多件事情是交給了一個(gè)人做還是多個(gè)人做。如果是交給了一個(gè)人做就是并發(fā)，交給了多個(gè)人做就是并行。而這里要說(shuō)的是并發(fā)執(zhí)行，并使用 TypeScript 和 JavaScript 來(lái)實(shí)現(xiàn)一個(gè)并發(fā)隊(duì)列。

在生活中我們能處處看到并發(fā)隊(duì)列，與本文要說(shuō)的并發(fā)隊(duì)列非常像。比如說(shuō)排隊(duì)，在一個(gè)售票窗口，只能一個(gè)一個(gè)的進(jìn)行，后面的人只能先等待前面的人買(mǎi)完票了，處理完手續(xù)后才能進(jìn)行買(mǎi)票。本文要講的并發(fā)隊(duì)列原理與這個(gè)非常像。

2. 核心代碼解析

先不展示全部的代碼，講清楚核心的邏輯后，其他的代碼也就是起個(gè)輔助的作用，也就沒(méi)有難理解的地方了。

核心代碼我將其分為以下幾個(gè)部分，從易到難進(jìn)行講解：

使用示例
添加任務(wù)
運(yùn)行任務(wù)
執(zhí)行一個(gè)任務(wù)
判斷是否執(zhí)行結(jié)束

2.1. 執(zhí)行示例

可以看到下面定義并添加了了兩個(gè)任務(wù)，均在兩秒后輸出一段話(huà)到控制臺(tái)，但是我們?cè)趧?chuàng)建并發(fā)隊(duì)列時(shí)指定最大并發(fā)數(shù)為 1，所以一次只能執(zhí)行一個(gè)任務(wù)，并且該任務(wù)隊(duì)列的執(zhí)行順序是先添加的先執(zhí)行。

// 所有的任務(wù)執(zhí)行完畢后的回調(diào)函數(shù)
let callback = (result: any) => {
    console.log(result);
};

let concurrencyTask = new ConcurrencyTask(1, callback);

// 添加任務(wù)
concurrencyTask.addTask((resolve, reject) => {
    setTimeout(() => {
        console.log("2 秒后得到執(zhí)行"); // 2 秒后輸出
        resolve();
    }, 2000);
});

concurrencyTask.addTask((resolve, reject) => {
    setTimeout(() => {
        console.log("4 秒后得到執(zhí)行"); // 4 秒后輸出
        resolve();
    }, 2000);
});

concurrencyTask.run(false);

2.2. 添加任務(wù)

下面是添加任務(wù)的代碼，添加的任務(wù)要求是一個(gè)函數(shù)，并在執(zhí)行時(shí)會(huì)接收到三個(gè)參數(shù)：resolve，rejecct，args。這三個(gè)參數(shù)分別為 Promise 的 resolve 和 reject，而 args 是函數(shù)執(zhí)行需要的可變參數(shù)，如果在任務(wù)隊(duì)列執(zhí)行過(guò)程中添加任務(wù)則不允許加入。

type Task = (resolve: Function, reject: Function, ...args: Array<any>) => any;

/**
 * 添加任務(wù)到任務(wù)隊(duì)列, 不會(huì)執(zhí)行
 * @param task 任務(wù)
 * @return 是否添加成功, 如果任務(wù)處于執(zhí)行階段返回 false
 */
public addTask(task: Task): boolean {
    if (!this.getRunning()) {
        this.taskList.push(task);
        return true;
    }
    return false;
}

2.3. 運(yùn)行任務(wù)

canAbort 參數(shù)表示隊(duì)列執(zhí)行過(guò)程中是否可中斷，在執(zhí)行的任務(wù)中調(diào)用 reject 函數(shù)即可中斷任務(wù)的執(zhí)行，中斷后任務(wù)隊(duì)列將進(jìn)行重置，清空已執(zhí)行的和未執(zhí)行的任務(wù)以及重置其他數(shù)據(jù)。

下面的代碼的意思是執(zhí)行指定最大并發(fā)數(shù)的數(shù)量的任務(wù)，如果最大并發(fā)數(shù)大于任務(wù)總數(shù)量，則以任務(wù)總數(shù)量為最大并發(fā)數(shù)來(lái)執(zhí)行。

/**
 * 開(kāi)始運(yùn)行任務(wù)
 * @param canAbort 是否可中斷
 * @param args 任務(wù)執(zhí)行參數(shù)
 */
public run(canAbort: boolean = false, ...args: Array<any>): void {
    this.canAbort = canAbort;
    this.setRunning(true);
    let length = this.taskList.length;
    let maxConcurrency = Math.min(this.getMaxConcurrency(), length);
    for (let index = 0; index < maxConcurrency; index++) {
        this.executeSingleTask(args);
    }
}

2.4. 執(zhí)行一個(gè)任務(wù)

由于任務(wù)執(zhí)行具有異步性，所以我們使用 Promise 來(lái)包裹任務(wù)，并把 resolve, reject 傳遞給任務(wù)函數(shù)，讓它來(lái)決定任務(wù)何時(shí)結(jié)束。

當(dāng)一個(gè)任務(wù)調(diào)用了 resolve 函數(shù)時(shí)，將會(huì)判斷任務(wù)是否全部得到執(zhí)行，即執(zhí)行 judgeExecuteEnd 函數(shù)，如果任務(wù)調(diào)用 reject 函數(shù)，將會(huì)判斷是否可以中斷任務(wù)的執(zhí)行，并重置任務(wù)隊(duì)列，當(dāng)然不想重置任務(wù)隊(duì)列可以在源代碼上進(jìn)行修改，這里我就不改了。

然后每個(gè)任務(wù)的 promise 會(huì)保存在 taskPromiseList 變量中，它是一個(gè) Promise 類(lèi)型的數(shù)組。

/**
 * 執(zhí)行單個(gè)任務(wù)
 * @param args 函數(shù)執(zhí)行參數(shù)
 */
private executeSingleTask(...args: Array<any>): void {
    let promise = new Promise<void>((resolve, reject) => {
        let result = this.taskList[this.taskIndex++](resolve, reject, args);
        this.handleResult.push(result);
    });
    promise.then(() => {
        this.judgeExecuteEnd(args);
    }).catch((error) => {
        // 如果可以中斷任務(wù)的執(zhí)行, 則重置任務(wù)隊(duì)列
        if (this.canAbort) {
            this.reset();
            return;
        }
        console.error(error);
    });
    this.taskPromiseList.push(promise);
}

2.5. 判斷是否執(zhí)行結(jié)束

下面的代碼中 taskIndex 是當(dāng)前任務(wù)的索引，runOver 為是否執(zhí)行結(jié)束的標(biāo)志。

這里我們判斷 taskPromiseList 中的 promise 是否全部完成

/**
 * 判斷是否執(zhí)行結(jié)束
 * @param args 函數(shù)執(zhí)行所需參數(shù)
 */
private judgeExecuteEnd(args: Array<any>): void {
    // 如果全部任務(wù)都得到執(zhí)行, 并且執(zhí)行沒(méi)有結(jié)束
    // 設(shè)置 runOver 的原因是最后幾個(gè)并發(fā)執(zhí)行的任務(wù)在執(zhí)行完畢后都會(huì)
    // 觸發(fā)該函數(shù), 而 runOverCallback 函數(shù)應(yīng)只執(zhí)行一次
    if (this.taskIndex >= this.taskList.length && !this.runOver) {
        this.runOver = true;
        let result = this.handleResult;
        Promise.all(this.taskPromiseList).then(() => {
            this.runOverCallback && this.runOverCallback(result);
        }).catch((error) => {
            // 如果不允許中斷，則會(huì)執(zhí)行任務(wù)全部完成回調(diào)
            if(!this.canAbort) {
                this.runOverCallback && this.runOverCallback(result);
            }
            console.error(error);
        });
        this.reset();
        return;
    }
    // 如果沒(méi)有執(zhí)行結(jié)束，就執(zhí)行下一個(gè)任務(wù)
    this.executeSingleTask(args);
}

3. 源代碼展示

下面的代碼直接復(fù)制到 ts 文件中是不會(huì)有任何的效果的，因?yàn)闉g覽器不能解析 ts 代碼，我們需要使用 ts 編譯器將其編譯為 js 代碼后，再引用 js 文件即可。安裝 TypeScript 見(jiàn)文末。

/* 
  功能描述: 并發(fā)隊(duì)列
  創(chuàng)建時(shí)間: 2023年 12月 17日
 */

type Task = (resolve: Function, reject: Function, ...args: Array<any>) => any;
type ResultCallback = (result: Array<any>) => any;

/**
 * 并發(fā)任務(wù)隊(duì)列
 */
class ConcurrencyTask {

    /**
     * 任務(wù)集合
     */
    private taskList: Array<Task>;

    /**
     * 處理結(jié)果
     */
    private handleResult: Array<any>;

    /**
     * 是否正在執(zhí)行任務(wù)
     */
    private running: boolean;

    /**
     * 最大并發(fā)數(shù)
     */
    private maxConcurrency: number;

    /**
     * 默認(rèn)的最大并發(fā)數(shù)
     */
    private static DEFAULT_MAX_CONCURRENCY: number = 2;

    /**
     * 當(dāng)前任務(wù)索引
     */
    private taskIndex: number;

    /**
     * 用 promise 包裹任務(wù)
     */
    private taskPromiseList: Array<Promise<void>>;

    /**
     * 是否可中斷
     */
    private canAbort: boolean;

    /**
     * 執(zhí)行結(jié)束
     */
    private runOver: boolean;

    /**
     * 任務(wù)全部執(zhí)行完畢時(shí)的回調(diào)函數(shù)
     */
    private runOverCallback: ResultCallback;

    /**
     * 創(chuàng)建并發(fā)任務(wù)隊(duì)列
     * @param maxConcurrency 最大并發(fā)數(shù)
     * @param runOverCallback 任務(wù)全部執(zhí)行完畢后的回調(diào)
     */
    public constructor(maxConcurrency: number = ConcurrencyTask.DEFAULT_MAX_CONCURRENCY, runOverCallback: ResultCallback) {
        this.setRunOverCallback(runOverCallback);
        this.setMaxConcurrency(maxConcurrency);
        this.initial();
    }

    private initial(): void {
        this.canAbort = false;
        this.reset();
    }

    /**
     * 添加任務(wù)到任務(wù)隊(duì)列, 不會(huì)執(zhí)行
     * @param task 任務(wù)
     * @return 是否添加成功, 如果任務(wù)處于執(zhí)行階段返回 false
     */
    public addTask(task: Task): boolean {
        if (!this.getRunning()) {
            this.taskList.push(task);
            return true;
        }
        return false;
    }

    /**
     * 開(kāi)始運(yùn)行任務(wù)
     * @param canAbort 是否可中斷
     * @param args 任務(wù)執(zhí)行參數(shù)
     */
    public run(canAbort: boolean = false, ...args: Array<any>): void {
        this.canAbort = canAbort;
        this.setRunning(true);
        let length = this.taskList.length;
        let maxConcurrency = Math.min(this.getMaxConcurrency(), length);
        for (let index = 0; index < maxConcurrency; index++) {
            this.executeSingleTask(args);
        }
    }

    /**
     * 執(zhí)行單個(gè)任務(wù)
     * @param args 函數(shù)執(zhí)行參數(shù)
     */
    private executeSingleTask(...args: Array<any>): void {
        let promise = new Promise<void>((resolve, reject) => {
            let result = this.taskList[this.taskIndex++](resolve, reject, args);
            this.handleResult.push(result);
        });
        promise.then(() => {
            this.judgeExecuteEnd(args);
        }).catch((error) => {
            // 如果可以中斷任務(wù)的執(zhí)行, 則重置任務(wù)隊(duì)列
            if (this.canAbort) {
                this.reset();
                return;
            }
            console.error(error);
        });
        this.taskPromiseList.push(promise);
    }

    /**
     * 判斷是否執(zhí)行結(jié)束
     * @param args 函數(shù)執(zhí)行所需參數(shù)
     */
    private judgeExecuteEnd(args: Array<any>): void {
        // 如果全部任務(wù)都得到執(zhí)行, 并且執(zhí)行沒(méi)有結(jié)束
        // 設(shè)置 runOver 的原因是最后幾個(gè)并發(fā)執(zhí)行的任務(wù)在執(zhí)行完畢后都會(huì)
        // 觸發(fā)該函數(shù), 而 runOverCallback 函數(shù)應(yīng)只執(zhí)行一次
        if (this.taskIndex >= this.taskList.length && !this.runOver) {
            this.runOver = true;
            let result = this.handleResult;
            Promise.all(this.taskPromiseList).then(() => {
                this.runOverCallback && this.runOverCallback(result);
            }).catch((error) => {
                if(!this.canAbort) {
                    this.runOverCallback && this.runOverCallback(result);
                }
                console.error(error);
            });
            this.reset();
            return;
        }
        this.executeSingleTask(args);
    }

    private reset(): void {
        this.taskList = [];
        this.taskIndex = 0;
        this.taskPromiseList = [];
        this.running = false;
        this.handleResult = [];
    }

    private setRunning(running: boolean): void {
        this.running = running;
    }

    public getRunning(): boolean {
        return this.running;
    }

    /**
     * 設(shè)置任務(wù)全部執(zhí)行完畢后的回調(diào)函數(shù), 如果隊(duì)列正在執(zhí)行則返回 false
     * @param runOverCallback 回調(diào)函數(shù)
     */
    public setRunOverCallback(runOverCallback: ResultCallback): boolean {
        if(!this.getRunning()) {
            this.runOverCallback = runOverCallback;
            return true;
        }
        return false;
    }

    /**
     * 設(shè)置最大并發(fā)數(shù), 如果正在執(zhí)行返回 false
     * @param maxConcurrency 最大并發(fā)數(shù), 小于等于 0 時(shí)使用默認(rèn)值
     */
    public setMaxConcurrency(maxConcurrency: number): boolean {
        if(maxConcurrency <= 0) {
            this.maxConcurrency = ConcurrencyTask.DEFAULT_MAX_CONCURRENCY;
        }
        if (!this.getRunning()) {
            this.maxConcurrency = maxConcurrency;
            return true;
        }
        return false;
    }

    public getMaxConcurrency(): number {
        return this.maxConcurrency;
    }
}

4. 安裝 TypeScript

由于 TypeScript 是運(yùn)行在 Node.js 上的，所以我們還需要安裝 Node.js，安裝 Node.js 可前往 Node.Js 中文網(wǎng)。

這里僅提供 windows 上的 TypeScript 的安裝方式。

首先以管理員的方式進(jìn)入 cmd（win + R，輸入 cmd，然后 ctrl + shift + enter 即可）。

使用以下的命令全局安裝：

npm i -g typescript

之后在任意目錄下創(chuàng)建一個(gè) ts 文件，然后在該文件夾下打開(kāi) cmd，執(zhí)行 tsc xx.ts 就會(huì)得到一個(gè)編譯后的 ja 文件。

到此這篇關(guān)于JavaScript+TypeScript實(shí)現(xiàn)并發(fā)隊(duì)列的示例的文章就介紹到這了,更多相關(guān)JavaScript TypeScript并發(fā)隊(duì)列內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: