客戶(hù)端JavaScript的線程池設(shè)計(jì)詳解

更新時(shí)間：2022年01月24日 11:35:35 作者：Vanghua

這篇文章主要為大家介紹了客戶(hù)端JavaScript的線程池設(shè)計(jì)，具有一定的參考價(jià)值，感興趣的小伙伴們可以參考一下，希望能夠給你帶來(lái)幫助

1.介紹：

本打算在客戶(hù)端JavaScript進(jìn)行機(jī)器學(xué)習(xí)算法計(jì)算時(shí)應(yīng)用線程池來(lái)優(yōu)化，就像()演示的神經(jīng)網(wǎng)絡(luò)。但是由于各種原因不了了之了。本次遇到了一個(gè)新的問(wèn)題，客戶(hù)端的MD5運(yùn)算也是耗時(shí)操作，如果同時(shí)對(duì)多個(gè)字符串或文件進(jìn)行MD5加密就可以使用線程池來(lái)優(yōu)化。

2.準(zhǔn)備工作：

到npm官網(wǎng)搜索spark-md5，到其github倉(cāng)庫(kù)下載spark-md5.js。該js文件支持AMD，CommonJS和web工作線程的模塊系統(tǒng)，我們?cè)趯?shí)現(xiàn)線程池時(shí)，線程工作代碼交給web工作線程處理。

在這里插入圖片描述

3.測(cè)試spark-md5是否正常工作：

創(chuàng)建一個(gè)網(wǎng)頁(yè)，再創(chuàng)建一個(gè)worker.js用于保存工作線程的代碼。以下述代碼測(cè)試，如果成功輸出MD5編碼，那么準(zhǔn)備工作完成。

客戶(hù)端網(wǎng)頁(yè)代碼

<script>
    let worker = new Worker("worker.js")
    worker.postMessage("Danny")
    worker.onmessage = function({data}) {
        console.log(data)
        worker.terminate()
    }
</script>

工作線程代碼

self.importScripts("spark-md5.js")

self.onmessage = function({data}) {
    self.postMessage(self.SparkMD5.hash(data))
}

4.線程池設(shè)計(jì)

1. 目標(biāo)：本次線程池設(shè)計(jì)的目標(biāo)是初始創(chuàng)建n個(gè)初始線程，能夠滿(mǎn)足任意個(gè)線程請(qǐng)求，超出n的請(qǐng)求并不丟棄，而是等待到出現(xiàn)空閑線程后再分配之。

2. 基本設(shè)計(jì)思路：為了基本滿(mǎn)足上述目標(biāo)，至少要有一個(gè)線程分配功能，一個(gè)線程回收功能。

3. 線程分配功能設(shè)計(jì)：

線程池滿(mǎn)指的是線程池已經(jīng)沒(méi)有可用空閑線程
通知對(duì)象是一個(gè)不可逆狀態(tài)機(jī)，可以用Promise對(duì)象來(lái)實(shí)現(xiàn)
阻塞請(qǐng)求隊(duì)列存儲(chǔ)Promise對(duì)象的resolve方法即可
存儲(chǔ)線程池中的線程使用數(shù)組即可，數(shù)組每個(gè)元素是一個(gè)對(duì)象，包括線程和線程狀態(tài)
返回給用戶(hù)的可用線程還需要有線程在數(shù)組中的下標(biāo)，在線程釋放中會(huì)用到

在這里插入圖片描述

4. 線程釋放功能設(shè)計(jì)：

線程釋放功能需要接收一個(gè)參數(shù)，為線程的標(biāo)識(shí)，3中設(shè)計(jì)該標(biāo)識(shí)為數(shù)組下標(biāo)
當(dāng)線程釋放后，查看阻塞請(qǐng)求隊(duì)列是否為空，如果不為空，說(shuō)明有被阻塞的線程請(qǐng)求，此時(shí)令隊(duì)首元素出隊(duì)即可，執(zhí)行resolve()通知對(duì)象的狀態(tài)變更為Fulfilled

在這里插入圖片描述

5. 實(shí)現(xiàn)線程池：

class MD5Pool {
    // worker用于存儲(chǔ)線程
    worker = []
    // status是線程池狀態(tài)
    status = "Idle"
    // 阻塞請(qǐng)求隊(duì)列
    blockRequestQueue = []
    // size為用戶(hù)希望的線程池的容量
    constructor(size) {
        for(let i = 0; i < size; i ++)
            this.worker.push({
                worker: new Worker("worker.js"),
                status: "Idle"
            })
    }
    
    // 線程池狀態(tài)更新函數(shù)
    statusUpdate() {
        let sum = 0
        this.worker.forEach(({ status }) => {
            if(status === "Busy")
                sum ++
        })
        if(sum === this.worker.length)
            this.status = "Busy"
         else
            this.status = "Idle"
    }
    
    // 線程請(qǐng)求方法
    assign() {
        if(this.status !== "Busy") {
            // 此時(shí)線程池不滿(mǎn)，遍歷線程，尋找一個(gè)空閑線程
            for (let i = 0; i < this.worker.length; i++)
                if (this.worker[i].status === "Idle") {
                    // 該線程空閑，更新?tīng)顟B(tài)為忙碌
                    this.worker[i].status = "Busy"
                    // 更新線程池狀態(tài)，如果這是最后一個(gè)空閑線程，那么線程池狀態(tài)變?yōu)闈M(mǎn)
                    this.statusUpdate()
                    // 返回給用戶(hù)該線程，和該線程的標(biāo)識(shí)，標(biāo)識(shí)用數(shù)組下標(biāo)表示
                    return {
                        worker: this.worker[i].worker,
                        index: i
                    }
                }
        }
        else {
            // 此時(shí)線程池滿(mǎn)
            let resolve = null
            // 創(chuàng)建一個(gè)通知對(duì)象
            let promise = new Promise(res => {
                // 取得通知對(duì)象的狀態(tài)改變方法
                resolve = res
            })
            // 通知對(duì)象的狀態(tài)改變方法加入阻塞請(qǐng)求隊(duì)列
            this.blockRequestQueue.push(resolve)
            // 返回給請(qǐng)求者線程池已滿(mǎn)信息和通知對(duì)象
            return {
                info: "full",
                wait: promise
            }
        }
    }
    
    // 線程釋放方法，接收一個(gè)參數(shù)為線程標(biāo)識(shí)
    release(index) {
        this.worker[index].status = "Idle"
        // 阻塞請(qǐng)求隊(duì)列中的第一個(gè)請(qǐng)求出隊(duì)，隊(duì)列中存儲(chǔ)的是promise的resolve方法，此時(shí)執(zhí)行，通知請(qǐng)求者已經(jīng)有可用的線程了
        if(this.blockRequestQueue.length)
            // 阻塞請(qǐng)求隊(duì)列隊(duì)首出列，并執(zhí)行通知對(duì)象的狀態(tài)改變方法
            this.blockRequestQueue.shift()()
        // 更新線程池狀態(tài)，此時(shí)一定空閑
        this.status = "Idle"
    }
}

5.spark-md5對(duì)文件進(jìn)行md5編碼

說(shuō)明：

在3的測(cè)試中spark-md5只是對(duì)簡(jiǎn)單字符串進(jìn)行MD5編碼，并非需要大量運(yùn)算的耗時(shí)操作。spark-md5可以對(duì)文件進(jìn)行MD5編碼，耗時(shí)較多，實(shí)現(xiàn)如下。

注意：

spark-md5對(duì)文件編碼時(shí)必須要對(duì)文件進(jìn)行切片后再加密整合，否則不同文件可能會(huì)有相同編碼。詳情見(jiàn)github或npm。

// 在工作線程中引入spark-md5
self.importScripts("spark-md5.js")

let fd = new FileReader()
let spark = new self.SparkMD5.ArrayBuffer()

// 接收主線程發(fā)來(lái)的消息，是一個(gè)文件
self.onmessage = function(event) {
    // 獲取文件
    let chunk = event.data
    // spark-md5要求計(jì)算文件的MD5必須切片計(jì)算
    let chunks = fileSlice(chunk)
    // 計(jì)算MD5編碼
    load(chunks)
}

// 切片函數(shù)
function fileSlice(file) {
    let pos = 0
    let chunks = []
    // 將文件平均切成10分計(jì)算MD5
    const SLICE_SIZE = Math.ceil(file.size / 10)
    while(pos < file.size) {
        // slice可以自動(dòng)處理第二個(gè)參數(shù)越界
        chunks.push(file.slice(pos, pos + SLICE_SIZE))
        pos += SLICE_SIZE
    }
    return chunks
}

// MD5計(jì)算函數(shù)
async function load(chunks) {
    for(let i = 0; i < chunks.length; i ++) {
        fd.readAsArrayBuffer(chunks[i])
        // 在這里希望節(jié)約空間，因此復(fù)用了FileReader，而不是每次循環(huán)新創(chuàng)建一個(gè)FileReader。需要等到FileReader完成read后才可以進(jìn)行下一輪復(fù)用，因此用await阻塞。
        await new Promise(res => {
            fd.onload = function(event) {
                spark.append(event.target.result)
                if(i === chunks.length - 1) {
                    self.postMessage(spark.end())
                }
                res()
            }
        })
    }
}

6.大量文件進(jìn)行MD5加密并使用線程池優(yōu)化

下面的測(cè)試代碼就是對(duì)上文所述的拼接

網(wǎng)頁(yè)代碼

<input id="input" type="file" multiple onchange="handleChanged()"/>
<body>
    <script>
        class MD5Pool {
            worker = []
            status = "Idle"
            blockRequestQueue = []
            constructor(size) {
                for(let i = 0; i < size; i ++)
                    this.worker.push({
                        worker: new Worker("worker.js"),
                        status: "Idle"
                    })
            }

            statusUpdate() {
                let sum = 0
                this.worker.forEach(({ status }) => {
                    if(status === "Busy")
                        sum ++
                })
                if(sum === this.worker.length)
                    this.status = "Busy"
                 else
                    this.status = "Idle"
            }

            assign() {
                if(this.status !== "Busy") {
                    for (let i = 0; i < this.worker.length; i++)
                        if (this.worker[i].status === "Idle") {
                            this.worker[i].status = "Busy"
                            this.statusUpdate()
                            return {
                                worker: this.worker[i].worker,
                                index: i
                            }
                        }
                }
                else {
                    let resolve = null
                    let promise = new Promise(res => {
                        resolve = res
                    })
                    this.blockRequestQueue.push(resolve)
                    return {
                        info: "full",
                        wait: promise
                    }
                }
            }

            release(index) {
                this.worker[index].status = "Idle"
                // 阻塞請(qǐng)求隊(duì)列中的第一個(gè)請(qǐng)求出隊(duì)，隊(duì)列中存儲(chǔ)的是promise的resolve方法，此時(shí)執(zhí)行，通知請(qǐng)求者已經(jīng)有可用的線程了
                if(this.blockRequestQueue.length)
                    this.blockRequestQueue.shift()()
                this.status = "Idle"
            }
        }

        // input點(diǎn)擊事件處理函數(shù)
        function handleChanged() {
            let files = event.target.files
            // 創(chuàng)建一個(gè)大小為2的MD5計(jì)算線程池
            let pool = new MD5Pool(2)
            // 計(jì)算切片文件的MD5編碼
            Array.prototype.forEach.call(files, file => {
                getMD5(file, pool)
            })
        }

        // 獲取文件的MD5編碼的函數(shù)，第一個(gè)參數(shù)是文件，第二個(gè)參數(shù)是MD5線程池
        async function getMD5(chunk, pool) {
            let thread = pool.assign()
            // 如果info為full，那么說(shuō)明線程池線程已被全部占用，需要等待
            if(thread.info === "full") {
                // 獲取線程通知對(duì)象
                let wait = thread.wait
                // 等到wait兌現(xiàn)時(shí)說(shuō)明已經(jīng)有可用的線程了
                await wait
                thread = pool.assign()
                let { worker, index } = thread
                worker.postMessage(chunk)
                worker.onmessage = function (event) {
                    console.log(event.data)
                    pool.release(index)
                }
            } else {
                let { worker, index } = thread
                worker.postMessage(chunk)
                worker.onmessage = function (event) {
                    console.log(event.data)
                    pool.release(index)
                }
            }
        }
    </script>
</body>

工作線程代碼

self.importScripts("spark-md5.js")

let fd = new FileReader()
let spark = new self.SparkMD5.ArrayBuffer()

self.onmessage = function(event) {
    // 獲取文件
    let chunk = event.data
    // spark-md5要求計(jì)算文件的MD5必須切片計(jì)算
    let chunks = fileSlice(chunk)
    // 計(jì)算MD5編碼
    load(chunks)
}

// 切片函數(shù)
function fileSlice(file) {
    let pos = 0
    let chunks = []
    // 將文件平均切成10分計(jì)算MD5
    const SLICE_SIZE = Math.ceil(file.size / 10)
    while(pos < file.size) {
        // slice可以自動(dòng)處理第二個(gè)參數(shù)越界
        chunks.push(file.slice(pos, pos + SLICE_SIZE))
        pos += SLICE_SIZE
    }
    return chunks
}

// MD5計(jì)算函數(shù)
async function load(chunks) {
    for(let i = 0; i < chunks.length; i ++) {
        fd.readAsArrayBuffer(chunks[i])
        // 在這里希望節(jié)約空間，因此復(fù)用了FileReader，而不是每次循環(huán)新創(chuàng)建一個(gè)FileReader。需要等到FileReader完成read后才可以進(jìn)行下一輪復(fù)用，因此用await阻塞。
        await new Promise(res => {
            fd.onload = function(event) {
                spark.append(event.target.result)
                if(i === chunks.length - 1) {
                    self.postMessage(spark.end())
                }
                res()
            }
        })
    }
}

隨機(jī)選取18個(gè)文件進(jìn)行MD5編碼，結(jié)果如下

在這里插入圖片描述