GO實現協程池管理的方法
使用channel實現協程池
通過 Channel 實現 Goroutine Pool,缺點是會造成協程的頻繁開辟和註銷,但好在簡單靈活通用。
package main import ( "fmt" "io/ioutil" "net/http" "sync" ) // Pool goroutine Pool type Pool struct { queue chan int wg *sync.WaitGroup } // New 新建一個協程池 func New(size int) *Pool { if size <= 0 { size = 1 } return &Pool{ queue: make(chan int, size), wg: &sync.WaitGroup{}, } } // Add 新增一個執行 func (p *Pool) Add(delta int) { // delta為正數就添加 for i := 0; i < delta; i++ { p.queue <- 1 } // delta為負數就減少 for i := 0; i > delta; i-- { <-p.queue } p.wg.Add(delta) } // Done 執行完成減一 func (p *Pool) Done() { <-p.queue p.wg.Done() } func (p *Pool) Wait() { p.wg.Wait() } func main() { // 這裡限制100個並發 pool := New(100) // sync.WaitGroup{} //假設需要發送1000萬個http請求,然後我並發100個協程取完成這件事 for i := 0; i < 10000000; i++ { pool.Add(1) //發現已存在100個人正在發瞭,那麼就會卡住,直到有人完成瞭宣佈自己退出協程瞭 go func(i int) { resp, err := http.Get("https://www.baidu.com") if err != nil { fmt.Println(i, err) } else { defer resp.Body.Close() result, _ := ioutil.ReadAll(resp.Body) fmt.Println(i, string(result)) } pool.Done() }(i) } pool.Wait() }
消費者模式實現協程池
頻繁對協程開辟與剔除,如果對性能有著很高的要求,建議優化成固定數目的協程取 channel 裡面取數據進行消費,這樣可以避免協程的創建與註銷。
package main import ( "fmt" "strconv" "sync" ) // 任務對象 type task struct { Production Consumer } // 設置消費者數目,也就是work pool大小 func (t *task) setConsumerPoolSize(poolSize int) { t.Production.Jobs = make(chan *Job, poolSize*10) t.Consumer.WorkPoolNum = poolSize } // 任務數據對象 type Job struct { Data string } func NewTask(handler func(jobs chan *Job) (b bool)) (t *task) { t = &task{ Production: Production{Jobs: make(chan *Job, 100)}, Consumer: Consumer{WorkPoolNum: 10, Handler: handler}, } return } type Production struct { Jobs chan *Job } func (c Production) AddData(data *Job) { c.Jobs <- data } type Consumer struct { WorkPoolNum int Handler func(chan *Job) (b bool) Wg sync.WaitGroup } // 異步開啟多個work去處理任務,但是所有work執行完畢才會退出程序 func (c *Consumer) disposeData(data chan *Job) { for i := 0; i <= c.WorkPoolNum; i++ { c.Wg.Add(1) go func() { defer func() { c.Wg.Done() }() c.Handler(data) }() } c.Wg.Wait() } func main() { // 實現一個用於處理數據的閉包,實現業務代碼 consumerHandler := func(jobs chan *Job) (b bool) { for jobs := range jobs { fmt.Println(jobs) } return } // new一個任務處理對象 t := NewTask(consumerHandler) t.setConsumerPoolSize(500) // 500個協程同時消費 // 根據自己的業務去生成數據通過AddData方法添加數據到生成channel,這裡是100萬條數據 go func() { for i := 0; i < 1000000; i++ { job := new(Job) iStr := strconv.Itoa(i) job.Data = "定義任務數據格式" + iStr t.AddData(job) } }() // 消費者消費數據 t.Consumer.disposeData(t.Production.Jobs) }
到此這篇關於GO實現協程池管理的方法的文章就介紹到這瞭,更多相關go實現協程池內容請搜索WalkonNet以前的文章或繼續瀏覽下面的相關文章希望大傢以後多多支持WalkonNet!
推薦閱讀:
- Go語言如何輕松編寫高效可靠的並發程序
- Golang中的sync包的WaitGroup操作
- Go語言學習之WaitGroup用法詳解
- golang 並發編程之生產者消費者詳解
- 示例剖析golang中的CSP並發模型