Go語言協(xié)程池的實現(xiàn)示例

更新時間：2025年09月26日 09:52:15 作者：fbbqt

本文主要介紹了Go語言協(xié)程池的原理與實現(xiàn),通過限制goroutine數(shù)量避免性能下降,及通過AddTask和Run方法管理任務分配,確保并發(fā)控制,提升系統(tǒng)穩(wěn)定性,感興趣的可以了解一下

Go語言雖然有著高效的GMP調度模型，理論上支持成千上萬的goroutine，但是goroutine過多，對調度，gc以及系統(tǒng)內存都會造成壓力，這樣會使我們的服務性能不升反降。常用做法可以用池化技術，構造一個協(xié)程池，把進程中的協(xié)程控制在一定的數(shù)量，防止系統(tǒng)中goroutine過多，影響服務性能。

協(xié)程池模型

協(xié)程池簡單理解就是有一個池子一樣的東西，里面裝有固定數(shù)量的goroutine，當有一個任務到來的時候，會將這個任務交給池子里的一個空閑的goroutine去處理，如果池子里沒有空閑的goroutine了，任務就會阻塞等待。所以協(xié)程池有三個角色Worker，Task，Pool。

屬性定義

Worker：用于執(zhí)行任務的goroutine
Task: 具體的任務
Pool: 池子

下面看一下各個角色的定義：

Task定義

Task有一個函數(shù)成員，表示這個task具體的執(zhí)行邏輯：

type Task struct {
    f func() error  // 具體的執(zhí)行邏輯
}

Pool定義

Pool有兩個成員，Capacity表示池子里的worker的數(shù)量，即工作的goroutine的數(shù)量，JobCh表示任務隊列用于存放任務，goroutine從這個JobCh獲取任務執(zhí)行任務邏輯：

type Pool struct {
    RunningWorkers int64     // 運行著的worker數(shù)量
    Capacity       int64     // 協(xié)程池worker容量---goroutine數(shù)量
    JobCh          chan *Task // 用于worker取任務
    sync.Mutex
}

Worker 定義

// p為Pool對象指針
for task := range p.JobCh {
    do ...      
}

執(zhí)行任務單元，簡單理解就是干活的goroutine，這個worker其實只做一件事情，就是不斷的從任務隊列里面取任務執(zhí)行，而worker的數(shù)量就是協(xié)程池里協(xié)程的數(shù)量，由Pool的參數(shù)指定。

方法定義

NewTask用于創(chuàng)建一個任務，參數(shù)是一個函數(shù)，返回值是一個Task類型。

func NewTask(funcArg func() error) *Task

NewPool返回一個協(xié)程數(shù)量固定為Capacity協(xié)程池對象指針，其任務隊列的長度為taskNum。

func NewPool(Capacity int, taskNum int) *Pool

接下來主要介紹協(xié)程池的各個方法：

AddTask方法是往協(xié)程池添加任務，如果當前運行著的worker數(shù)量小于協(xié)程池worker容量，則立即啟動一個協(xié)程worker來處理任務，否則將任務添加到任務隊列。

func (p *Pool) AddTask(task *Task)

Run方法將協(xié)程池跑起來，啟動一個worker來處理任務。

func (p *Pool) Run()

協(xié)程池處理任務流程圖：

協(xié)程池實現(xiàn)：

package main

import (
    "fmt"
    "sync"
    "sync/atomic"
    "time"
)

type Task struct {
    f func() error // 具體的任務邏輯
}

func NewTask(funcArg func() error) *Task {
    return &Task{
       f: funcArg,
    }
}

type Pool struct {
    RunningWorkers int64      // 運行著的worker數(shù)量
    Capacity       int64      // 協(xié)程池worker容量
    JobCh          chan *Task // 用于worker取任務
    sync.Mutex
}

func NewPool(capacity int64, taskNum int) *Pool {
    return &Pool{
       Capacity: capacity,
       JobCh:    make(chan *Task, taskNum),
    }
}

func (p *Pool) GetCap() int64 {
    return p.Capacity
}

func (p *Pool) incRunning() { // runningWorkers + 1
    atomic.AddInt64(&p.RunningWorkers, 1)
}

func (p *Pool) decRunning() { // runningWorkers - 1
    atomic.AddInt64(&p.RunningWorkers, -1)
}

func (p *Pool) GetRunningWorkers() int64 {
    return atomic.LoadInt64(&p.RunningWorkers)
}

func (p *Pool) run() {
    p.incRunning()
    go func() {
       defer func() {
          p.decRunning()
       }()
       for task := range p.JobCh {
          task.f()
       }
    }()
}

// AddTask 往協(xié)程池添加任務
func (p *Pool) AddTask(task *Task) {
    // 加鎖防止啟動多個 worker
    p.Lock()
    defer p.Unlock()

    if p.GetRunningWorkers() < p.GetCap() { // 如果任務池滿, 則不再創(chuàng)建 worker
       // 創(chuàng)建啟動一個 worker
       p.run()
    }

    // 將任務推入隊列, 等待消費
    p.JobCh <- task
}

func main() {
    // 創(chuàng)建任務池
    pool := NewPool(3, 10)

    for i := 0; i < 20; i++ {
       // 任務放入池中
       pool.AddTask(NewTask(func() error {
          fmt.Printf("I am Task\n")
          return nil
       }))
    }

    time.Sleep(1e9) // 等待執(zhí)行
}

運行結果：

I am Task
I am Task
I am Task
I am Task
I am Task
I am Task
I am Task
I am Task
I am Task
I am Task
I am Task
I am Task
I am Task
I am Task
I am Task
I am Task
I am Task
I am Task
I am Task
I am Task