如果你熟悉 Java、Python 等編程語言，那么你一定聽說或者使用過進(jìn)程池或線程池。因?yàn)檫M(jìn)程和線程不是越多越好，過多的進(jìn)程或線程可能造成資源浪費(fèi)和性能下降。所以池化技術(shù)在這些主流編程語言中非常流行，可以有效控制并發(fā)場(chǎng)景下資源使用量。

而 Go 語言則沒有提供多進(jìn)程和多線程的支持，僅提供了協(xié)程（goroutine）的概念。在 Go 中開啟 goroutine 的成本非常低，以至于我們?cè)诮^大多數(shù)情況下開啟 goroutine 時(shí)根本無需考慮協(xié)程數(shù)量，所以也就很少有人提及 Go 的協(xié)程池化技術(shù)。不過使用場(chǎng)景少，不代表完全沒用。通過協(xié)程池我們可以來掌控資源使用量，降低協(xié)程泄漏風(fēng)險(xiǎn)。

gammazero/workerpool 就是用來實(shí)現(xiàn)協(xié)程池的 Go 包，本文我們一起來學(xué)習(xí)一下其使用方法，并深入其源碼來探究下如何實(shí)現(xiàn)一個(gè) Go 協(xié)程池。

使用示例

workerpool 直譯過來是工作池，在 Go 中就是指代協(xié)程池。workerpool 的用法非常簡(jiǎn)單，示例代碼如下：

package main

import (
	"fmt"
	"time"

	"github.com/gammazero/workerpool"
)

func main() {
	wp := workerpool.New(2)
	requests := []string{"alpha", "beta", "gamma", "delta", "epsilon"}

	for _, r := range requests {
		wp.Submit(func() {
			fmt.Printf("%s: Handling request: %s\n", time.Now().Format(time.RFC3339), r)
			time.Sleep(1 * time.Second)
		})
	}

	wp.StopWait()
}

workerpool.New(2) 表示我們創(chuàng)建了一個(gè)容量為 2 的協(xié)程池，即同一時(shí)刻最多只會(huì)有 2 個(gè) goroutine 正在執(zhí)行。wp.Submit() 用來提交一個(gè)任務(wù)，任務(wù)類型為無參數(shù)和返回值的函數(shù) func()，這里我們?cè)?for 循環(huán)中提交了 5 個(gè)任務(wù)。調(diào)用 wp.StopWait() 可以等待所有已提交的任務(wù)執(zhí)行完成。

執(zhí)行示例代碼，得到輸出如下：

$ go run main.go
2025-05-08T23:40:16+08:00: Handling request: alpha
2025-05-08T23:40:16+08:00: Handling request: beta
2025-05-08T23:40:17+08:00: Handling request: gamma
2025-05-08T23:40:17+08:00: Handling request: delta
2025-05-08T23:40:18+08:00: Handling request: epsilon

不過這里的輸出內(nèi)容并不是一下子全部輸出完成的，而是兩行兩行的輸出。

根據(jù)打印的時(shí)間可以發(fā)現(xiàn)，是先輸出：

2025-05-08T23:40:16+08:00: Handling request: alpha
2025-05-08T23:40:16+08:00: Handling request: beta

接著等待 1s 再輸出：

2025-05-08T23:40:17+08:00: Handling request: gamma
2025-05-08T23:40:17+08:00: Handling request: delta

再次等待 1s 最后輸出：

2025-05-08T23:40:18+08:00: Handling request: epsilon

這個(gè)輸出結(jié)果符合預(yù)期，也就是說同一時(shí)刻最多只會(huì)有 2 個(gè) goroutine 正在執(zhí)行。

源碼解讀

workerpool 用法非常簡(jiǎn)單，接下來我們一起看看其實(shí)現(xiàn)原理。

下圖是 workerpool 源碼中實(shí)現(xiàn)的全部功能：

WorkerPool 是一個(gè)結(jié)構(gòu)體，源碼中圍繞這個(gè)結(jié)構(gòu)體定義了很多函數(shù)或方法。這些函數(shù)或方法你不必死記硬背，先有一個(gè)宏觀上的認(rèn)知，接下來我將帶你深入學(xué)習(xí)其中的核心方法。

WorkerPool 結(jié)構(gòu)體完整定義如下：

// WorkerPool 是 Go 協(xié)程的集合池，用于確保同時(shí)處理請(qǐng)求的協(xié)程數(shù)量嚴(yán)格受控于預(yù)設(shè)的上限值
type WorkerPool struct {
	maxWorkers   int                 // 最大工作協(xié)程數(shù)
	taskQueue    chan func()         // 任務(wù)提交隊(duì)列
	workerQueue  chan func()         // 工作協(xié)程消費(fèi)隊(duì)列
	stoppedChan  chan struct{}       // 停止完成通知通道
	stopSignal   chan struct{}       // 停止信號(hào)通道
	waitingQueue deque.Deque[func()] // 等待隊(duì)列（雙端隊(duì)列）
	stopLock     sync.Mutex          // 停止操作互斥鎖
	stopOnce     sync.Once           // 控制只停止一次
	stopped      bool                // 是否已經(jīng)停止
	waiting      int32               // 等待隊(duì)列中任務(wù)計(jì)數(shù)
	wait         bool                // 協(xié)程池退出時(shí)是否等待已入隊(duì)任務(wù)執(zhí)行完成
}

這里屬性很多，其中有 3 個(gè)屬性是需要我們重點(diǎn)關(guān)注的，taskQueue、workerQueue 以及 waitingQueue，這三者分別代表任務(wù)提交隊(duì)列、工作隊(duì)列和等待隊(duì)列。稍后我將通過一個(gè)流程圖來講解任務(wù)在這 3 個(gè)隊(duì)列中的傳遞流程，現(xiàn)在我們一起來看一下 WorkerPool 的構(gòu)造函數(shù)：

// New 創(chuàng)建并啟動(dòng)協(xié)程池
// maxWorkers 參數(shù)指定可以并發(fā)執(zhí)行任務(wù)的最大工作協(xié)程數(shù)。
func New(maxWorkers int) *WorkerPool {
	// 至少有一個(gè) worker
	if maxWorkers < 1 {
		maxWorkers = 1
	}

	// 實(shí)例化協(xié)程池對(duì)象
	pool := &WorkerPool{
		maxWorkers:  maxWorkers,
		taskQueue:   make(chan func()),
		workerQueue: make(chan func()),
		stopSignal:  make(chan struct{}),
		stoppedChan: make(chan struct{}),
	}

	// 啟動(dòng)任務(wù)調(diào)度器
	go pool.dispatch()

	return pool
}

New 函數(shù)創(chuàng)建一個(gè)指定容量的協(xié)程池對(duì)象 WorkerPool，我們已經(jīng)在使用示例中見過其用法了。這里邏輯還是比較簡(jiǎn)單的，僅接收一個(gè)參數(shù)，并初始化了幾個(gè)必要的屬性。

值得注意的是，這里通過開啟新的 goroutine 的方式啟動(dòng)了 dispatch() 方法，這個(gè)方法是協(xié)程池最核心的邏輯，用來實(shí)現(xiàn)任務(wù)的調(diào)度執(zhí)行。

為此，我畫了一張流程圖，來分析 WorkerPool 最核心的任務(wù)派發(fā)流程：

圖中涉及兩個(gè)方法，其中 Submit() 方法用于提交一個(gè)任務(wù)到協(xié)程池，dispatch 方法則用于派發(fā)任務(wù)到 goroutine 中去執(zhí)行。dispatch 方法內(nèi)部有一個(gè)無限循環(huán)，實(shí)現(xiàn)任務(wù)實(shí)時(shí)派發(fā)執(zhí)行。這個(gè) for 無限循環(huán)中控制著任務(wù)在 3 個(gè)隊(duì)列中的流轉(zhuǎn)和工作協(xié)程數(shù)量。

只要通過 Submit() 方法提交任務(wù)，就一定會(huì)進(jìn)入任務(wù)提交隊(duì)列 taskQueue 中，而 taskQueue 是一個(gè)通過 make(chan func()) 初始化的無緩沖的 channel，所以任務(wù)不會(huì)在里面停留，要么通過鏈路 ② 下發(fā)到等待隊(duì)列 waitingQueue 中，要么通過鏈路 ④ 下發(fā)到工作隊(duì)列 workerQueue 中。最終具體會(huì)下發(fā)到哪里，是 dispatch 方法中的 for 循環(huán)邏輯來決定的。

dispatch 的 for 循環(huán)中會(huì)處理任務(wù)分發(fā)，核心邏輯有兩個(gè)部分，包含兩種處理模式：

隊(duì)列優(yōu)先模式：在 for 循環(huán)中，會(huì)優(yōu)先判斷等待隊(duì)列 waitingQueue 是否為空，如果不為空，則進(jìn)入隊(duì)列優(yōu)先模式。

• 此時(shí)會(huì)優(yōu)先從等待隊(duì)列對(duì)頭取出任務(wù)，然后交給工作隊(duì)列 workerQueue，協(xié)程池中的工作協(xié)程（worker）就會(huì)不停的從 workerQueue 中拿到任務(wù)并執(zhí)行。
• 如果此時(shí)剛好還有新的任務(wù)被提交，則新提交的任務(wù)自動(dòng)進(jìn)入等待隊(duì)列尾部。
• 任務(wù)從提交到執(zhí)行的流程是 ① ② ③。

直通模式：等待隊(duì)列完全清空后，程序自動(dòng)切換到直通模式。

• 此時(shí)等待隊(duì)列 workerQueue 已經(jīng)清空，如果有新任務(wù)提交進(jìn)來，可以直接交給工作隊(duì)列 workerQueue，讓工作協(xié)程（worker）來執(zhí)行。
• 如果此時(shí)工作協(xié)程數(shù)量達(dá)到了協(xié)程池的上限，則將任務(wù)提交到等待隊(duì)列 waitingQueue 中。
• 任務(wù)從提交到執(zhí)行的流程是 ① ④。

以上就是協(xié)程池 dispatch 方法的核心調(diào)度流程。

接下來，我將對(duì) WorkerPool 核心代碼進(jìn)行一一解讀，以此來從微觀上更加細(xì)致的理解協(xié)程池的設(shè)計(jì)。

Submit 方法實(shí)現(xiàn)如下：

// Submit 將任務(wù)函數(shù)提交到工作池隊(duì)列等待執(zhí)行，不會(huì)等待任務(wù)執(zhí)行完成
func (p *WorkerPool) Submit(task func()) {
	if task != nil {
		p.taskQueue <- task
	}
}

這個(gè)方法非常簡(jiǎn)單，就是將任務(wù)提交到 taskQueue 隊(duì)列中。

接下來我們看下是最核心也是最復(fù)雜的方法 dispatch 是如何實(shí)現(xiàn)的：

// 任務(wù)派發(fā)，循環(huán)的將下一個(gè)排隊(duì)中的任務(wù)發(fā)送給可用的工作協(xié)程（worker）執(zhí)行
func (p *WorkerPool) dispatch() {
	defer close(p.stoppedChan)            // 保證調(diào)度器退出時(shí)關(guān)閉停止通知通道
	timeout := time.NewTimer(idleTimeout) // 創(chuàng)建 2 秒周期的空閑檢測(cè)定時(shí)器
	var workerCount int                   // 當(dāng)前活躍 worker 計(jì)數(shù)器
	var idle bool                         // 空閑狀態(tài)標(biāo)識(shí)
	var wg sync.WaitGroup                 // 用于等待所有 worker 完成

Loop:
	for { // 主循環(huán)處理任務(wù)分發(fā)

		// 隊(duì)列優(yōu)先模式：優(yōu)先檢測(cè)等待隊(duì)列
		if p.waitingQueue.Len() != 0 {
			if !p.processWaitingQueue() {
				break Loop // 等待隊(duì)列為空，退出循環(huán)
			}
			continue
		}

		// 直通模式：開始處理提交上來的新任務(wù)
		select {
		case task, ok := <-p.taskQueue: // 接收到新任務(wù)
			if !ok { // 協(xié)程池停止時(shí)會(huì)關(guān)閉任務(wù)通道，如果 !ok 說明協(xié)程池已停止，退出循環(huán)
				break Loop
			}

			select {
			case p.workerQueue <- task: // 嘗試派發(fā)任務(wù)
			default: // 沒有空閑的 worker，無法立即派發(fā)任務(wù)
				if workerCount < p.maxWorkers { // 如果協(xié)程池中的活躍協(xié)程數(shù)量小于最大值，那么創(chuàng)建一個(gè)新的協(xié)程（worker）來執(zhí)行任務(wù)
					wg.Add(1)
					go worker(task, p.workerQueue, &wg) // 創(chuàng)建新的 worker 執(zhí)行任務(wù)
					workerCount++                       // worker 記數(shù)加 1
				} else { // 已達(dá)協(xié)程池容量上限
					p.waitingQueue.PushBack(task)                              // 將任務(wù)提交到等待隊(duì)列
					atomic.StoreInt32(&p.waiting, int32(p.waitingQueue.Len())) // 原子更新等待計(jì)數(shù)
				}
			}
			idle = false // 標(biāo)記為非空閑
		case <-timeout.C: // 空閑超時(shí)處理
			// 在一個(gè)空閑超時(shí)周期內(nèi)，存在空閑的 workers，那么停止一個(gè) worker
			if idle && workerCount > 0 {
				if p.killIdleWorker() { // 回收一個(gè) worker
					workerCount-- // worker 計(jì)數(shù)減 1
				}
			}
			idle = true                // 標(biāo)記為空閑
			timeout.Reset(idleTimeout) // 復(fù)用定時(shí)器
		}
	}

	if p.wait { // 調(diào)用了 StopWait() 方法，需要運(yùn)行等待隊(duì)列中的任務(wù)，直至隊(duì)列清空
		p.runQueuedTasks()
	}

	// 終止所有 worker
	for workerCount > 0 {
		p.workerQueue <- nil // 發(fā)送終止信號(hào)給 worker
		workerCount--        // worker 計(jì)數(shù)減 1，直至為 0 退出循環(huán)
	}
	wg.Wait() // 阻塞等待所有 worker 完成

	timeout.Stop() // 停止定時(shí)器
}

這個(gè)方法代碼量稍微有點(diǎn)多，不過結(jié)合我上面畫的流程圖，其實(shí)也好理解。我在代碼注釋中也標(biāo)出了兩種任務(wù)處理模式：等待隊(duì)列優(yōu)先模式和直通模式。

我們先看等待隊(duì)列優(yōu)先模式：

// 隊(duì)列優(yōu)先模式：優(yōu)先檢測(cè)等待隊(duì)列
if p.waitingQueue.Len() != 0 {
    if !p.processWaitingQueue() {
        break Loop // 協(xié)程池已經(jīng)停止
    }
    continue // 隊(duì)列不為空則繼續(xù)下一輪循環(huán)
}

如果等待隊(duì)列不為空，則優(yōu)先處理等待隊(duì)列。p.processWaitingQueue 方法實(shí)現(xiàn)如下：

// 處理等待隊(duì)列
func (p *WorkerPool) processWaitingQueue() bool {
	select {
	case task, ok := <-p.taskQueue: // 接收到新任務(wù)
		if !ok { // 協(xié)程池停止時(shí)會(huì)關(guān)閉任務(wù)通道，如果 !ok 說明協(xié)程池已停止，返回 false，不再繼續(xù)處理
			return false
		}
		p.waitingQueue.PushBack(task) // 將新任務(wù)加入等待隊(duì)列隊(duì)尾
	case p.workerQueue <- p.waitingQueue.Front(): // 從等待隊(duì)列隊(duì)頭獲取任務(wù)并放入工作隊(duì)列
		p.waitingQueue.PopFront() // 任務(wù)已經(jīng)開始處理，所以要從等待隊(duì)列中移除任務(wù)
	}
	atomic.StoreInt32(&p.waiting, int32(p.waitingQueue.Len())) // 原子修改等待隊(duì)列中任務(wù)計(jì)數(shù)
	return true
}

這個(gè)方法中有兩個(gè) case 需要處理：

• 接收到新任務(wù)，直接加入到等待隊(duì)列 waitingQueue 的隊(duì)尾。
• 從等待隊(duì)列 waitingQueue 的隊(duì)頭獲取任務(wù)并放入工作隊(duì)列 workerQueue。

這與前文流程圖中的講解吻合。

任務(wù)交給工作隊(duì)列 workerQueue 以后，誰來處理 workerQueue 中的任務(wù)呢？我們接著往下看直通模式的代碼。

直通模式的代碼中同樣使用 select 多路復(fù)用，將邏輯分成了兩個(gè) case 來處理：

// 直通模式：開始處理提交上來的新任務(wù)
select {
case task, ok := <-p.taskQueue: // 接收到新任務(wù)
    ...
case <-timeout.C: // 空閑超時(shí)處理
    ...
}

兩個(gè) case 分別實(shí)現(xiàn)任務(wù)執(zhí)行和空閑超時(shí)處理。

我們先來看處理任務(wù)的 case：

case task, ok := <-p.taskQueue: // 接收到新任務(wù)
    if !ok { // 協(xié)程池停止時(shí)會(huì)關(guān)閉任務(wù)通道，如果 !ok 說明協(xié)程池已停止，退出循環(huán)
        break Loop
    }

    select {
    case p.workerQueue <- task: // 嘗試派發(fā)任務(wù)
    default: // 沒有空閑的 worker，無法立即派發(fā)任務(wù)
        if workerCount < p.maxWorkers { // 如果協(xié)程池中的活躍協(xié)程數(shù)量小于最大值，那么創(chuàng)建一個(gè)新的協(xié)程（worker）來執(zhí)行任務(wù)
            wg.Add(1)
            go worker(task, p.workerQueue, &wg) // 創(chuàng)建新的 worker 執(zhí)行任務(wù)
            workerCount++                       // worker 記數(shù)加 1
        } else { // 已達(dá)協(xié)程池容量上限
            p.waitingQueue.PushBack(task)                              // 將任務(wù)提交到等待隊(duì)列
            atomic.StoreInt32(&p.waiting, int32(p.waitingQueue.Len())) // 原子更新等待計(jì)數(shù)
        }
    }
    idle = false // 標(biāo)記為非空閑

直通模式下，有新的任務(wù)提交進(jìn)來，首先會(huì)嘗試直接將其加入工作隊(duì)列 workerQueue 中，如果任務(wù)下發(fā)失敗，則說明當(dāng)前時(shí)刻沒有空閑的工作協(xié)程（worker），無法立即派發(fā)任務(wù)。那么繼續(xù)比較當(dāng)前正在執(zhí)行的工作協(xié)程數(shù)量（workerCount）和協(xié)程池大小（maxWorkers），如果協(xié)程池中的活躍協(xié)程數(shù)量小于最大值，那么創(chuàng)建一個(gè)新的協(xié)程（worker）來執(zhí)行任務(wù)。否則，說明正在執(zhí)行的工作協(xié)程數(shù)量已達(dá)協(xié)程池容量上限，那么將任務(wù)提交到等待隊(duì)列 waitingQueue 中。那么下一次 for 循環(huán)執(zhí)行的時(shí)候，檢測(cè)到 waitingQueue 中有任務(wù)，就會(huì)優(yōu)先處理 waitingQueue。這也就實(shí)現(xiàn)了兩種模式的切換。

我們?cè)賮砜聪鹿ぷ鲄f(xié)程 worker 是如何執(zhí)行任務(wù)的：

// 工作協(xié)程，執(zhí)行任務(wù)并在收到 nil 信號(hào)時(shí)停止
func worker(task func(), workerQueue chan func(), wg *sync.WaitGroup) {
	for task != nil { // 循環(huán)執(zhí)行任務(wù)，直至接收到終止信號(hào) nil
		task()               // 執(zhí)行任務(wù)
		task = <-workerQueue // 接收新任務(wù)
	}
	wg.Done() // 標(biāo)記 worker 完成
}

可以發(fā)現(xiàn)，這里使用 for 循環(huán)來不停的執(zhí)行提交過來的任務(wù)，直至從 workerQueue 中接收到終止信號(hào) nil。那么這個(gè)終止信號(hào)是何時(shí)下發(fā)的呢？往下看你馬上能找到答案。

接下來我們看一下直通模式的另外一個(gè) case 邏輯：

case <-timeout.C: // 空閑超時(shí)處理
    // 在一個(gè)空閑超時(shí)周期內(nèi)，存在空閑的 workers，那么停止一個(gè) worker
    if idle && workerCount > 0 {
        if p.killIdleWorker() { // 回收一個(gè) worker
            workerCount-- // worker 計(jì)數(shù)減 1
        }
    }
    idle = true                // 標(biāo)記為空閑
    timeout.Reset(idleTimeout) // 復(fù)用定時(shí)器

這里使用定時(shí)器來管理超過特定時(shí)間，未收到任務(wù)，需要關(guān)閉空閑的工作協(xié)程（worker）。

關(guān)閉 worker 的方法是 p.killIdleWorker：

// 停止一個(gè)空閑 worker
func (p *WorkerPool) killIdleWorker() bool {
	select {
	case p.workerQueue <- nil: // 發(fā)送終止信號(hào)給工作協(xié)程（worker）
		// Sent kill signal to worker.
		return true
	default:
		// No ready workers. All, if any, workers are busy.
		return false
	}
}

這里正是通過給 workerQueue 發(fā)送 nil 來作為終止信號(hào)，以此來實(shí)現(xiàn)通知 worker 退出的。

看完了 Submit 和 dispatch 方法源碼，你現(xiàn)在是否對(duì)協(xié)程池有了更深入的認(rèn)知呢？你可以再回顧一下我在前文中畫的任務(wù)調(diào)度流程圖，加深印象。

workerpool 的源碼就講解到這里，其他方法實(shí)現(xiàn)其實(shí)都比較簡(jiǎn)單，就交給你自己去探索了。你可以參考我的中文注釋版源碼：github.com/jianghushin…。

總結(jié)

協(xié)程池作為 Go 中不那么常用的技術(shù)，依然有其存在的價(jià)值，本文介紹的 workerpool 項(xiàng)目是一個(gè)協(xié)程池的實(shí)現(xiàn)。

workerpool 用法非常簡(jiǎn)單，僅需要通過 workerpool.New(n) 函數(shù)既可創(chuàng)建一個(gè)大小為 n 的協(xié)程池，之后通過 wp.Submit(task) 既可以提交任務(wù)到協(xié)程池。

workerpool 內(nèi)部提供了 3 個(gè)隊(duì)列來對(duì)任務(wù)進(jìn)行派發(fā)調(diào)度，任務(wù)提交隊(duì)列 taskQueue 和工作隊(duì)列 workQueue 都是使用 channel 實(shí)現(xiàn)的，并且無緩沖，真正帶有緩沖效果的隊(duì)列是等待隊(duì)列 WaitingQueue，這個(gè)是真正的隊(duì)列實(shí)現(xiàn)，采用雙端隊(duì)列，而非 channel，并且不限制隊(duì)列長(zhǎng)度。也就是說，無論排隊(duì)多少個(gè)任務(wù)，workerpool 都不會(huì)阻止新任務(wù)的提交。所以，我們?cè)趧?chuàng)建協(xié)程池時(shí)需要設(shè)置一個(gè)合理的大小限制，以防止等待隊(duì)列無限增長(zhǎng)，任務(wù)很長(zhǎng)一段時(shí)間都得不到執(zhí)行。

此外，workerpool 內(nèi)部雖然會(huì)維護(hù)一個(gè)協(xié)程池，但超過一定空閑時(shí)間沒有任務(wù)提交過來，工作協(xié)程是會(huì)關(guān)閉的，之后新任務(wù)進(jìn)來再次啟動(dòng)新的協(xié)程，因?yàn)閱?dòng)新協(xié)程開銷小，所以沒長(zhǎng)久駐留協(xié)程。

到此這篇關(guān)于淺析在Go語言中如何實(shí)現(xiàn)協(xié)程池的文章就介紹到這了,更多相關(guān)Go協(xié)程池內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

最新評(píng)論