快捷導(dǎo)航

go Worker Pool Pattern的實(shí)現(xiàn)

更新時間：2026年06月25日 08:58:01 作者：geovindu

本文展示了一個用Go語言實(shí)現(xiàn)的WorkerPool工作池模式,通過協(xié)程池處理珠寶訂單的完整生產(chǎn)流程,文中通過示例代碼介紹的非常詳細(xì),對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧

項(xiàng)目結(jié)構(gòu)：

/*
# 版權(quán)所有  2026 ?涂聚文有限公司? ?
# 許可信息查看：言語成了邀功盡責(zé)的功臣，還需要行爲(wèi)每日來值班嗎
# 描述：Worker Pool Pattern 工作池模式
# Author    : geovindu,Geovin Du 涂聚文.
# IDE       : goLang 2024.3.6 go 26.2
# os        : windows 10
# database  : mysql 9.0 sql server 2019, postgreSQL 17.0  Oracle 21c Neo4j
# Datetime  : 2026/6/21 17:56
# User      :  geovindu
# Product   : GoLand
# Project   : godesginpattern
# File      : logger.go
*/
package utils
 
import (
    "godesginpattern/workerpool/config"
    "io"
    "log"
    "os"
)
 
var Logger *log.Logger
 
func InitLogger() {
    // 打開日志文件
    file, err := os.OpenFile(config.LogFileName, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0666)
    if err != nil {
        log.Fatal("日志文件創(chuàng)建失敗:", err)
    }
 
    // 同時輸出到：控制臺 + 文件
    multiWriter := io.MultiWriter(os.Stdout, file)
 
    // 初始化日志
    Logger = log.New(multiWriter, "", config.LogFlag)
}
 
 
 
 
/*
# 版權(quán)所有  2026 ?涂聚文有限公司? ?
# 許可信息查看：言語成了邀功盡責(zé)的功臣，還需要行爲(wèi)每日來值班嗎
# 描述：Worker Pool Pattern 工作池模式
# Author    : geovindu,Geovin Du 涂聚文.
# IDE       : goLang 2024.3.6 go 26.2
# os        : windows 10
# database  : mysql 9.0 sql server 2019, postgreSQL 17.0  Oracle 21c Neo4j
# Datetime  : 2026/6/21 18:12
# User      :  geovindu
# Product   : GoLand
# Project   : godesginpattern
# File      : retry.go
*/
package utils
 
import (
    "godesginpattern/workerpool/config"
    "time"
)
 
// Retry 執(zhí)行函數(shù)并自動重試
func Retry(task func() error) error {
    var err error
    for i := 0; i < config.MaxRetryTimes; i++ {
        err = task()
        if err == nil {
            return nil
        }
        Logger.Printf("任務(wù)失敗，第 %d 次重試，錯誤: %v", i+1, err)
        time.Sleep(500 * time.Millisecond)
    }
    return err
}
 
 
 
/*
# 版權(quán)所有  2026 ?涂聚文有限公司? ?
# 許可信息查看：言語成了邀功盡責(zé)的功臣，還需要行爲(wèi)每日來值班嗎
# 描述：Worker Pool Pattern 工作池模式
# Author    : geovindu,Geovin Du 涂聚文.
# IDE       : goLang 2024.3.6 go 26.2
# os        : windows 10
# database  : mysql 9.0 sql server 2019, postgreSQL 17.0  Oracle 21c Neo4j
# Datetime  : 2026/6/21 18:12
# User      :  geovindu
# Product   : GoLand
# Project   : godesginpattern
# File      : monitor.go
*/
package utils
 
import "sync/atomic"
 
type Monitor struct {
    Running  int64
    Waiting  int64
    Finished int64
    Failed   int64
}
 
func (m *Monitor) IncRunning()  { atomic.AddInt64(&m.Running, 1) }
func (m *Monitor) DecRunning()  { atomic.AddInt64(&m.Running, -1) }
func (m *Monitor) IncWaiting()  { atomic.AddInt64(&m.Waiting, 1) }
func (m *Monitor) DecWaiting()  { atomic.AddInt64(&m.Waiting, -1) }
func (m *Monitor) IncFinished() { atomic.AddInt64(&m.Finished, 1) }
func (m *Monitor) IncFailed()   { atomic.AddInt64(&m.Failed, 1) }
 
func (m *Monitor) Get() (running, waiting, finished, failed int64) {
    return atomic.LoadInt64(&m.Running),
        atomic.LoadInt64(&m.Waiting),
        atomic.LoadInt64(&m.Finished),
        atomic.LoadInt64(&m.Failed)
}
 
 
/*
# 版權(quán)所有  2026 ?涂聚文有限公司? ?
# 許可信息查看：言語成了邀功盡責(zé)的功臣，還需要行爲(wèi)每日來值班嗎
# 描述：Worker Pool Pattern 工作池模式
# Author    : geovindu,Geovin Du 涂聚文.
# IDE       : goLang 2024.3.6 go 26.2
# os        : windows 10
# database  : mysql 9.0 sql server 2019, postgreSQL 17.0  Oracle 21c Neo4j
# Datetime  : 2026/6/21 17:58
# User      :  geovindu
# Product   : GoLand
# Project   : godesginpattern
# File      : settings.go
*/
package config
 
import "log"
 
const (
    WorkerCount     = 3
    QueueMaxSize    = 100
    MinTaskDelay    = 0.3
    MaxTaskDelay    = 0.8
    MaxRetryTimes   = 3
    LogFileName     = "jewelry.log"
    DBPath          = "jewelry.db"
    HTTPPort        = ":8080"
    MonitorInterval = 2
)
 
const LogFlag = log.LstdFlags | log.Lmicroseconds
 
 
 
/*
# 版權(quán)所有  2026 ?涂聚文有限公司? ?
# 許可信息查看：言語成了邀功盡責(zé)的功臣，還需要行爲(wèi)每日來值班嗎
# 描述：Worker Pool Pattern 工作池模式
# Author    : geovindu,Geovin Du 涂聚文.
# IDE       : goLang 2024.3.6 go 26.2
# os        : windows 10
# database  : mysql 9.0 sql server 2019, postgreSQL 17.0  Oracle 21c Neo4j
# Datetime  : 2026/6/21 18:01
# User      :  geovindu
# Product   : GoLand
# Project   : godesginpattern
# File      : task.go
*/
package tasks
 
import (
    "godesginpattern/workerpool/config"
    "math/rand"
    "time"
)
 
func RawMaterialCheck(orderID string) error {
    simulateDelay()
    return nil
}
 
func JewelryProcess(orderID string) error {
    simulateDelay()
    return nil
}
 
func FinishedGoodsCheck(orderID string) error {
    simulateDelay()
    return nil
}
 
func InventoryRecord(orderID string) error {
    simulateDelay()
    return nil
}
 
func OrderDelivery(orderID string) error {
    simulateDelay()
    return nil
}
 
var FullProcessTasks = []func(string) error{
    RawMaterialCheck,
    JewelryProcess,
    FinishedGoodsCheck,
    InventoryRecord,
    OrderDelivery,
}
 
func simulateDelay() {
    rand.Seed(time.Now().UnixNano())
    delay := config.MinTaskDelay + rand.Float64()*(config.MaxTaskDelay-config.MinTaskDelay)
    time.Sleep(time.Duration(delay*1000) * time.Millisecond)
}
 
 
 
/*
# 版權(quán)所有  2026 ?涂聚文有限公司? ?
# 許可信息查看：言語成了邀功盡責(zé)的功臣，還需要行爲(wèi)每日來值班嗎
# 描述：Worker Pool Pattern 工作池模式
# Author    : geovindu,Geovin Du 涂聚文.
# IDE       : goLang 2024.3.6 go 26.2
# os        : windows 10
# database  : mysql 9.0 sql server 2019, postgreSQL 17.0  Oracle 21c Neo4j
# Datetime  : 2026/6/21 18:01
# User      :  geovindu
# Product   : GoLand
# Project   : godesginpattern
# File      : worker_pool.go
*/
package core
 
import (
    "godesginpattern/workerpool/config"
    "godesginpattern/workerpool/utils"
    "os"
    "os/signal"
    "sync"
    "syscall"
    "time"
)
 
type Task struct {
    OrderID string
    Func    func(string) error
}
 
type WorkerPool struct {
    workerCount int
    taskChan    chan *Task
    wg          sync.WaitGroup
    monitor     *utils.Monitor
    quit        chan os.Signal
}
 
func NewWorkerPool(workerCnt int, queueSize int) *WorkerPool {
    wp := &WorkerPool{
        workerCount: workerCnt,
        taskChan:    make(chan *Task, queueSize),
        monitor:     &utils.Monitor{},
        quit:        make(chan os.Signal, 1),
    }
    signal.Notify(wp.quit, syscall.SIGINT, syscall.SIGTERM)
    return wp
}
 
func (wp *WorkerPool) worker(id int) {
    utils.Logger.Printf("Worker %d 已啟動", id)
    for {
        select {
        case task, ok := <-wp.taskChan:
            if !ok {
                utils.Logger.Printf("Worker %d 安全退出", id)
                return
            }
 
            wp.monitor.DecWaiting()
            wp.monitor.IncRunning()
 
            utils.Logger.Printf("Worker %d 開始任務(wù): %s", id, task.OrderID)
            err := utils.Retry(func() error {
                return task.Func(task.OrderID)
            })
 
            if err != nil {
                utils.Logger.Printf("Worker %d 任務(wù)失敗: %s, 錯誤: %v", id, task.OrderID, err)
            } else {
                utils.Logger.Printf("Worker %d 完成任務(wù): %s", id, task.OrderID)
            }
 
            wp.monitor.DecRunning()
            wp.monitor.IncFinished()
            wp.wg.Done()
 
        case <-wp.quit:
            utils.Logger.Printf("Worker %d 收到關(guān)閉信號，退出", id)
            return
        }
    }
}
 
func (wp *WorkerPool) Start() {
    utils.Logger.Println("工作池啟動")
    for i := 1; i <= wp.workerCount; i++ {
        go wp.worker(i)
    }
    go wp.monitorLoop()
}
 
func (wp *WorkerPool) Submit(task *Task) {
    wp.wg.Add(1)
    wp.monitor.IncWaiting()
    wp.taskChan <- task
}
 
func (wp *WorkerPool) monitorLoop() {
    ticker := time.NewTicker(time.Duration(config.MonitorInterval) * time.Second)
    defer ticker.Stop()
    for {
        select {
        case <-ticker.C:
            r, w, f, failed := wp.monitor.Get()
            utils.Logger.Printf("[監(jiān)控] 運(yùn)行:%d | 等待:%d | 完成:%d | 失敗:%d", r, w, f, failed)
            // 運(yùn)行、等待都為0，代表所有任務(wù)全部完成
            if r == 0 && w == 0 {
                utils.Logger.Println("所有任務(wù)處理完成，自動停止監(jiān)控輸出")
                return
            }
        case <-wp.quit:
            utils.Logger.Println("監(jiān)控循環(huán)收到關(guān)閉信號，停止監(jiān)控輸出")
            return
        }
    }
}
 
func (wp *WorkerPool) Wait() {
    // 阻塞等待 Ctrl+C / kill 信號
    <-wp.quit
    utils.Logger.Println("優(yōu)雅關(guān)閉中...")
    // 關(guān)閉任務(wù)通道，worker 不再接收新任務(wù)
    close(wp.taskChan)
    // 等待正在執(zhí)行的任務(wù)全部處理完畢
    wp.wg.Wait()
    utils.Logger.Println("? 所有任務(wù)執(zhí)行完畢，服務(wù)安全退出")
}

調(diào)用：

/*
# 版權(quán)所有  2026 ?涂聚文有限公司? ?
# 許可信息查看：言語成了邀功盡責(zé)的功臣，還需要行爲(wèi)每日來值班嗎
# 描述：Worker Pool Pattern 工作池模式
# Author    : geovindu,Geovin Du 涂聚文.
# IDE       : goLang 2024.3.6 go 26.2
# os        : windows 10
# database  : mysql 9.0 sql server 2019, postgreSQL 17.0  Oracle 21c Neo4j
# Datetime  : 2026/6/21 18:02
# User      :  geovindu
# Product   : GoLand
# Project   : godesginpattern
# File      : workerpoolbll.go
*/
package bll
 
import (
    "fmt"
    "godesginpattern/workerpool/config"
    "godesginpattern/workerpool/core"
    "godesginpattern/workerpool/tasks"
    "godesginpattern/workerpool/utils"
)
 
func WorkerPoolMain() {
    utils.InitLogger()
    logger := utils.Logger
 
    logger.Println("================================================")
    logger.Println("   珠寶企業(yè)級生產(chǎn)系統(tǒng)（Go 企業(yè)級 Worker Pool）")
    logger.Println("================================================")
 
    pool := core.NewWorkerPool(config.WorkerCount, config.QueueMaxSize)
    pool.Start()
 
    totalOrder := 10
    logger.Printf("開始提交 %d 個珠寶訂單\n", totalOrder)
 
    for i := 1; i <= totalOrder; i++ {
        orderID := fmt.Sprintf("訂單-%03d", i)
        for _, fn := range tasks.FullProcessTasks {
            pool.Submit(&core.Task{
                OrderID: orderID,
                Func:    fn,
            })
        }
    }
 
    logger.Println("? 所有訂單已提交，按 Ctrl+C 優(yōu)雅關(guān)閉")
    pool.Wait()
}

輸出：