網站首頁 編程語言 正文
Goroutine
Goroutine 是 Golang 提供的一種輕量級線程,我們通常稱之為「協程」,相比較線程,創建一個協程的成本是很低的。所以你會經常看到 Golang 開發的應用出現上千個協程并發的場景。
Goroutine 的優勢:
- 與線程相比,Goroutines 成本很低。
它們的堆棧大小只有幾 kb,堆棧可以根據應用程序的需要增長和縮小,context switch 也很快,而在線程的情況下,堆棧大小必須指定并固定。
- Goroutine 被多路復用到更少數量的 OS 線程。
一個包含數千個 Goroutine 的程序中可能只有一個線程。如果該線程中的任何 Goroutine 阻塞等待用戶輸入,則創建另一個 OS 線程并將剩余的 Goroutine 移動到新的 OS 線程。所有這些都由運行時處理,作為開發者無需耗費心力關心,這也使得我們有很干凈的 API 來支持并發。
- Goroutines 使用 channel 進行通信。
channel 的設計有效防止了在使用 Goroutine 訪問共享內存時發生競爭條件(race conditions) 。channel 可以被認為是 Goroutine 進行通信的管道。
下文中我們會以「協程」來代指 Goroutine。
協程池
在高并發場景下,我們可能會啟動大量的協程來處理業務邏輯。協程池是一種利用池化技術,復用對象,減少內存分配的頻率以及協程創建開銷,從而提高協程執行效率的技術。
最近抽空了解了字節官方開源的 gopkg 庫提供的 gopool
協程池實現,感覺還是很高質量的,代碼也非常簡潔清晰,而且 Kitex
底層也在使用 gopool
來管理協程,這里我們梳理一下設計和實現。
gopool
Repository:https://github.com/bytedance/gopkg/tree/develop/util/gopool
gopool
?is a high-performance goroutine pool which aims to reuse goroutines and limit the number of goroutines. It is an alternative to the?go
?keyword.
了解官方 README 就會發現gopool
的用法其實非常簡單,將曾經我們經常使用的?go func(){...}
?替換為?gopool.Go(func(){...})
即可。
此時 gopool
將會使用默認的配置來管理你啟動的協程,你也可以選擇針對業務場景配置池子大小,以及擴容上限。
old:
go func() { // do your job }()
new:
import ( "github.com/bytedance/gopkg/util/gopool" ) gopool.Go(func(){ /// do your job })
核心實現
下面我們來看看gopool
是怎樣實現協程池管理的。
Pool
Pool
是一個定義了協程池能力的接口。
type Pool interface { // 池子的名稱 Name() string // 設置池子內Goroutine的容量 SetCap(cap int32) // 執行 f 函數 Go(f func()) // 帶 ctx,執行 f 函數 CtxGo(ctx context.Context, f func()) // 設置發生panic時調用的函數 SetPanicHandler(f func(context.Context, interface{})) }
gopool
提供了這個接口的默認實現(即下面即將介紹的pool
),當我們直接調用 gopool.CtxGo 時依賴的就是這個。
這樣的設計模式在 Kitex
中也經常出現,所有的依賴均設計為接口,便于隨后擴展,底層提供一個默認的實現暴露出去,這樣對調用方也很友好。
type pool struct { // 池子名稱 name string // 池子的容量, 即最大并發工作的 goroutine 的數量 cap int32 // 池子配置 config *Config // task 鏈表 taskHead *task taskTail *task taskLock sync.Mutex taskCount int32 // 記錄當前正在運行的 worker 的數量 workerCount int32 // 當 worker 出現panic時被調用 panicHandler func(context.Context, interface{}) } // NewPool 創建一個新的協程池,初始化名稱,容量,配置 func NewPool(name string, cap int32, config *Config) Pool { p := &pool{ name: name, cap: cap, config: config, } return p }
調用 NewPool
獲取了以 Pool
的形式返回的 pool
結構體。
Task
type task struct { ctx context.Context f func() next *task }
task
是一個鏈表結構,可以把它理解為一個待執行的任務,它包含了當前節點需要執行的函數f
, 以及指向下一個task
的指針。
綜合前一節 pool
的定義,我們可以看到,一個協程池 pool
對應了一組task
。
pool
維護了指向鏈表的頭尾的兩個指針:taskHead
和 taskTail
,以及鏈表的長度taskCount
和對應的鎖 taskLock
。
Worker
type worker struct { pool *pool }
一個 worker
就是邏輯上的一個執行器,它唯一對應到一個協程池 pool
。當一個worker
被喚起,將會開啟一個goroutine
,不斷地從 pool
中的 task
鏈表獲取任務并執行。
func (w *worker) run() { go func() { for { // 聲明即將執行的 task var t *task // 操作 pool 中的 task 鏈表,加鎖 w.pool.taskLock.Lock() if w.pool.taskHead != nil { // 拿到 taskHead 準備執行 t = w.pool.taskHead // 更新鏈表的 head 以及數量 w.pool.taskHead = w.pool.taskHead.next atomic.AddInt32(&w.pool.taskCount, -1) } // 如果前一步拿到的 taskHead 為空,說明無任務需要執行,清理后返回 if t == nil { w.close() w.pool.taskLock.Unlock() w.Recycle() return } w.pool.taskLock.Unlock() // 執行任務,針對 panic 會recover,并調用配置的 handler func() { defer func() { if r := recover(); r != nil { msg := fmt.Sprintf("GOPOOL: panic in pool: %s: %v: %s", w.pool.name, r, debug.Stack()) logger.CtxErrorf(t.ctx, msg) if w.pool.panicHandler != nil { w.pool.panicHandler(t.ctx, r) } } }() t.f() }() t.Recycle() } }() }
整體來看
看到這里,其實就能把整個流程串起來了。我們來看看對外的接口 CtxGo(context.Context, f func())
到底做了什么?
func Go(f func()) { CtxGo(context.Background(), f) } func CtxGo(ctx context.Context, f func()) { defaultPool.CtxGo(ctx, f) } func (p *pool) CtxGo(ctx context.Context, f func()) { // 創建一個 task 對象,將 ctx 和待執行的函數賦值 t := taskPool.Get().(*task) t.ctx = ctx t.f = f // 將 task 插入 pool 的鏈表的尾部,更新鏈表數量 p.taskLock.Lock() if p.taskHead == nil { p.taskHead = t p.taskTail = t } else { p.taskTail.next = t p.taskTail = t } p.taskLock.Unlock() atomic.AddInt32(&p.taskCount, 1) // 以下兩個條件滿足時,創建新的 worker 并喚起執行: // 1. task的數量超過了配置的限制 // 2. 當前運行的worker數量小于上限(或無worker運行) if (atomic.LoadInt32(&p.taskCount) >= p.config.ScaleThreshold && p.WorkerCount() < atomic.LoadInt32(&p.cap)) || p.WorkerCount() == 0 { // worker數量+1 p.incWorkerCount() // 創建一個新的worker,并把當前 pool 賦值 w := workerPool.Get().(*worker) w.pool = p // 喚起worker執行 w.run() } }
相信看了代碼注釋,大家就能理解發生了什么。
gopool
會自行維護一個 defaultPool
,這是一個默認的 pool
結構體,在引入包的時候就進行初始化。當我們直接調用 gopool.CtxGo()
時,本質上是調用了 defaultPool
的同名方法
func init() { defaultPool = NewPool("gopool.DefaultPool", 10000, NewConfig()) } const ( defaultScalaThreshold = 1 ) // Config is used to config pool. type Config struct { // 控制擴容的門檻,一旦待執行的 task 超過此值,且 worker 數量未達到上限,就開始啟動新的 worker ScaleThreshold int32 } // NewConfig creates a default Config. func NewConfig() *Config { c := &Config{ ScaleThreshold: defaultScalaThreshold, } return c }
defaultPool
的名稱為 gopool.DefaultPool
,池子容量一萬,擴容下限為 1。
當我們調用 CtxGo
時,gopool
就會更新維護的任務鏈表,并且判斷是否需要擴容 worker
:
- 若此時已經有很多
worker
啟動(底層一個worker
對應一個goroutine
),不需要擴容,就直接返回。 - 若判斷需要擴容,就創建一個新的
worker
,并調用worker.run()
方法啟動,各個worker
會異步地檢查pool
里面的任務鏈表是否還有待執行的任務,如果有就執行。
三個角色的定位
-
task
是一個待執行的任務節點,同時還包含了指向下一個任務的指針,鏈表結構; -
worker
是一個實際執行任務的執行器,它會異步啟動一個goroutine
執行協程池里面未執行的task
; -
pool
是一個邏輯上的協程池,對應了一個task
鏈表,同時負責維護task
狀態的更新,以及在需要的時候創建新的worker
。
使用 sync.Pool 進行性能優化
其實到這個地方,gopool
已經是一個代碼簡潔清晰的協程池庫了,但是性能上顯然有改進空間,所以gopool
的作者應用了多次 sync.Pool
來池化對象的創建,復用woker和task對象。
這里建議大家直接看源碼,其實在上面的代碼中已經有所涉及。
- task 池化
var taskPool sync.Pool func init() { taskPool.New = newTask } func newTask() interface{} { return &task{} } func (t *task) Recycle() { t.zero() taskPool.Put(t) }
- worker 池化
var workerPool sync.Pool func init() { workerPool.New = newWorker } func newWorker() interface{} { return &worker{} } func (w *worker) Recycle() { w.zero() workerPool.Put(w) }
原文鏈接:https://juejin.cn/post/7086443265309818894
相關推薦
- 2023-07-02 Python+streamlit實現輕松創建人事系統_python
- 2022-05-21 python實現會員信息管理系統(List)_python
- 2023-03-16 Redis設置key的過期時間_Redis
- 2022-08-02 深入了解Golang的map增量擴容_Golang
- 2022-10-31 React中使用antd組件的方法_React
- 2022-04-01 Error!!在k8s中查看pod日志報錯
- 2022-06-16 golang?validator庫參數校驗實用技巧干貨_Golang
- 2022-09-09 python中對開區間和閉區間的理解_python
- 最近更新
-
- window11 系統安裝 yarn
- 超詳細win安裝深度學習環境2025年最新版(
- Linux 中運行的top命令 怎么退出?
- MySQL 中decimal 的用法? 存儲小
- get 、set 、toString 方法的使
- @Resource和 @Autowired注解
- Java基礎操作-- 運算符,流程控制 Flo
- 1. Int 和Integer 的區別,Jav
- spring @retryable不生效的一種
- Spring Security之認證信息的處理
- Spring Security之認證過濾器
- Spring Security概述快速入門
- Spring Security之配置體系
- 【SpringBoot】SpringCache
- Spring Security之基于方法配置權
- redisson分布式鎖中waittime的設
- maven:解決release錯誤:Artif
- restTemplate使用總結
- Spring Security之安全異常處理
- MybatisPlus優雅實現加密?
- Spring ioc容器與Bean的生命周期。
- 【探索SpringCloud】服務發現-Nac
- Spring Security之基于HttpR
- Redis 底層數據結構-簡單動態字符串(SD
- arthas操作spring被代理目標對象命令
- Spring中的單例模式應用詳解
- 聊聊消息隊列,發送消息的4種方式
- bootspring第三方資源配置管理
- GIT同步修改后的遠程分支