使用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實現協程池內容請搜索腳本之家以前的文章或繼續瀏覽下面的相關文章希望大家以后多多支持腳本之家！
您可能感興趣的文章:- golang協程池模擬實現群發郵件功能
- golang協程池設計詳解
- golang 40行代碼實現通用協程池
