答案是使用Go語言通過channel和goroutine實(shí)現(xiàn)Worker Pool模式來控制並發(fā)數(shù)量並高效處理批量任務(wù)。 1. 定義任務(wù)Job和結(jié)果Result結(jié)構(gòu)體；2. 創(chuàng)建緩衝的任務(wù)channel和結(jié)果channel；3. 啟動(dòng)固定數(shù)量的worker goroutine，每個(gè)worker從jobs channel讀取任務(wù)並寫入results channel，完成後調(diào)用wg.Done()；4. 主goroutine發(fā)送所有任務(wù)後關(guān)閉jobs channel；5. 使用sync.WaitGroup等待所有worker完成，並在goroutine中關(guān)閉results channel；6. 主goroutine從results中收集結(jié)果，統(tǒng)計(jì)成功與失敗數(shù)量。該模式通過限制goroutine數(shù)量防止資源耗盡，適用於批量任務(wù)處理，並可通過加入context超時(shí)、重試機(jī)制等進(jìn)行擴(kuò)展優(yōu)化，最終程序正確輸出處理結(jié)果並安全退出。

在Go 中， Worker Pool（工作池）是一種常見的並發(fā)模式，用於高效地處理大量任務(wù)，同時(shí)限制同時(shí)運(yùn)行的goroutine 數(shù)量，避免資源耗盡。它非常適合處理批量任務(wù)，比如文件處理、網(wǎng)絡(luò)請(qǐng)求、數(shù)據(jù)清洗等。

下面通過一個(gè)具體的例子來演示如何用Go 實(shí)現(xiàn)一個(gè)簡(jiǎn)單的Worker Pool。

?? 場(chǎng)景：處理一批任務(wù)（比如下載URL）

假設(shè)我們有100 個(gè)任務(wù)（比如HTTP 請(qǐng)求），但我們只想同時(shí)運(yùn)行10 個(gè)goroutine 來處理這些任務(wù)，防止系統(tǒng)過載。

? 核心結(jié)構(gòu)

• 一個(gè)任務(wù)隊(duì)列（ jobs channel）
• 多個(gè)worker（goroutine）
• 一個(gè)結(jié)果隊(duì)列（ results channel）
• 等待所有任務(wù)完成（ sync.WaitGroup ）

? 完整示例代碼

package main

import (
    "fmt"
    "math/rand"
    "sync"
    "time"
)

// 任務(wù)結(jié)構(gòu)type Job struct {
    ID int
}

// 結(jié)果結(jié)構(gòu)type Result struct {
    JobID int
    Success bool
    Message string
}

func worker(id int, jobs <-chan Job, results chan<- Result, wg *sync.WaitGroup) {
    defer wg.Done()
    for job := range jobs {
        fmt.Printf("Worker %d started job %d\n", id, job.ID)

        // 模擬耗時(shí)操作（比如網(wǎng)絡(luò)請(qǐng)求）
        time.Sleep(time.Duration(rand.Intn(1000)) * time.Millisecond)

        success := rand.Float32() > 0.3 // 70% 成功率var msg string
        if success {
            msg = "processed successfully"
        } else {
            msg = "failed"
        }

        results <- Result{
            JobID: job.ID,
            Success: success,
            Message: msg,
        }
        fmt.Printf("Worker %d finished job %d\n", id, job.ID)
    }
}

func main() {
    const numJobs = 100
    const numWorkers = 10

    jobs := make(chan Job, numJobs)
    results := make(chan Result, numJobs)

    var wg sync.WaitGroup

    // 1. 啟動(dòng)workers
    for i := 1; i <= numWorkers; i {
        wg.Add(1)
        go worker(i, jobs, results, &wg)
    }

    // 2. 發(fā)送任務(wù)for i := 1; i <= numJobs; i {
        jobs <- Job{ID: i}
    }
    close(jobs) // 關(guān)閉jobs channel，表示沒有更多任務(wù)// 3. 在另一個(gè)goroutine 中等待所有worker 完成，然後關(guān)閉results
    go func() {
        wg.Wait()
        close(results)
    }()

    // 4. 收集結(jié)果successful := 0
    failed := 0
    for result := range results {
        if result.Success {
            successful  
        } else {
            failed  
        }
        // 可選：打印或記錄結(jié)果// fmt.Printf("Job %d: %s\n", result.JobID, result.Message)
    }

    fmt.Printf("Processing complete. Success: %d, Failed: %d\n", successful, failed)
}

? 關(guān)鍵點(diǎn)說明

? jobs和results使用buffered channel

• jobs := make(chan Job, numJobs) ：緩衝通道，避免發(fā)送阻塞
• 也可以不緩衝，但在發(fā)送任務(wù)時(shí)要確保有worker 在接收

? close(jobs)後， range jobs會(huì)自動(dòng)退出

• 所有worker 的for job := range jobs在channel 關(guān)閉後會(huì)自然結(jié)束

? 使用sync.WaitGroup等待所有worker 退出

• 每個(gè)worker 執(zhí)行完wg.Done()
• 主goroutine 等待完成後關(guān)閉results ，防止range results死鎖

? results也需要關(guān)閉

• 否則for range results會(huì)永遠(yuǎn)阻塞

? 可擴(kuò)展優(yōu)化

• 錯(cuò)誤重試機(jī)制：在worker 中加入重試邏輯
• 超時(shí)控制：使用context.WithTimeout
• 動(dòng)態(tài)worker 數(shù)量：根據(jù)負(fù)載調(diào)整
• 優(yōu)先級(jí)任務(wù)隊(duì)列：使用多個(gè)channel 或優(yōu)先隊(duì)列

? 適用場(chǎng)景

• 批量處理文件或數(shù)據(jù)庫記錄
• 並發(fā)抓取網(wǎng)頁
• 郵件發(fā)送隊(duì)列
• 圖片/視頻轉(zhuǎn)碼
• 日誌處理

基本上就這些。 Go 的channel goroutine 讓worker pool 實(shí)現(xiàn)非常簡(jiǎn)潔、直觀。關(guān)鍵是理解channel 關(guān)閉和WaitGroup 配合的使用，避免死鎖或洩漏。

以上是以身作則的工作池的詳細(xì)內(nèi)容。更多資訊請(qǐng)關(guān)注PHP中文網(wǎng)其他相關(guān)文章！