使用channel實(shí)現(xiàn)協(xié)程池
通過(guò) Channel 實(shí)現(xiàn) Goroutine Pool,缺點(diǎn)是會(huì)造成協(xié)程的頻繁開(kāi)辟和注銷(xiāo),但好在簡(jiǎn)單靈活通用。
package main
import (
"fmt"
"io/ioutil"
"net/http"
"sync"
)
// Pool goroutine Pool
type Pool struct {
queue chan int
wg *sync.WaitGroup
}
// New 新建一個(gè)協(xié)程池
func New(size int) *Pool {
if size = 0 {
size = 1
}
return Pool{
queue: make(chan int, size),
wg: sync.WaitGroup{},
}
}
// Add 新增一個(gè)執(zhí)行
func (p *Pool) Add(delta int) {
// delta為正數(shù)就添加
for i := 0; i delta; i++ {
p.queue - 1
}
// delta為負(fù)數(shù)就減少
for i := 0; i > delta; i-- {
-p.queue
}
p.wg.Add(delta)
}
// Done 執(zhí)行完成減一
func (p *Pool) Done() {
-p.queue
p.wg.Done()
}
func (p *Pool) Wait() {
p.wg.Wait()
}
func main() {
// 這里限制100個(gè)并發(fā)
pool := New(100) // sync.WaitGroup{}
//假設(shè)需要發(fā)送1000萬(wàn)個(gè)http請(qǐng)求,然后我并發(fā)100個(gè)協(xié)程取完成這件事
for i := 0; i 10000000; i++ {
pool.Add(1) //發(fā)現(xiàn)已存在100個(gè)人正在發(fā)了,那么就會(huì)卡住,直到有人完成了宣布自己退出協(xié)程了
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()
}
消費(fèi)者模式實(shí)現(xiàn)協(xié)程池
頻繁對(duì)協(xié)程開(kāi)辟與剔除,如果對(duì)性能有著很高的要求,建議優(yōu)化成固定數(shù)目的協(xié)程取 channel 里面取數(shù)據(jù)進(jìn)行消費(fèi),這樣可以避免協(xié)程的創(chuàng)建與注銷(xiāo)。
package main
import (
"fmt"
"strconv"
"sync"
)
// 任務(wù)對(duì)象
type task struct {
Production
Consumer
}
// 設(shè)置消費(fèi)者數(shù)目,也就是work pool大小
func (t *task) setConsumerPoolSize(poolSize int) {
t.Production.Jobs = make(chan *Job, poolSize*10)
t.Consumer.WorkPoolNum = poolSize
}
// 任務(wù)數(shù)據(jù)對(duì)象
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
}
// 異步開(kāi)啟多個(gè)work去處理任務(wù),但是所有work執(zhí)行完畢才會(huì)退出程序
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() {
// 實(shí)現(xiàn)一個(gè)用于處理數(shù)據(jù)的閉包,實(shí)現(xiàn)業(yè)務(wù)代碼
consumerHandler := func(jobs chan *Job) (b bool) {
for jobs := range jobs {
fmt.Println(jobs)
}
return
}
// new一個(gè)任務(wù)處理對(duì)象
t := NewTask(consumerHandler)
t.setConsumerPoolSize(500) // 500個(gè)協(xié)程同時(shí)消費(fèi)
// 根據(jù)自己的業(yè)務(wù)去生成數(shù)據(jù)通過(guò)AddData方法添加數(shù)據(jù)到生成channel,這里是100萬(wàn)條數(shù)據(jù)
go func() {
for i := 0; i 1000000; i++ {
job := new(Job)
iStr := strconv.Itoa(i)
job.Data = "定義任務(wù)數(shù)據(jù)格式" + iStr
t.AddData(job)
}
}()
// 消費(fèi)者消費(fèi)數(shù)據(jù)
t.Consumer.disposeData(t.Production.Jobs)
}
到此這篇關(guān)于GO實(shí)現(xiàn)協(xié)程池管理的方法的文章就介紹到這了,更多相關(guān)go實(shí)現(xiàn)協(xié)程池內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家!
您可能感興趣的文章:- golang協(xié)程池模擬實(shí)現(xiàn)群發(fā)郵件功能
- golang協(xié)程池設(shè)計(jì)詳解
- golang 40行代碼實(shí)現(xiàn)通用協(xié)程池