使用channel实现协程池
通过 Channel 实现 Goroutine Pool,缺点是会造成协程的频繁开辟和注销，但好在简单灵活通用。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
|
package mainimport ( "fmt" "io/ioutil" "net/http" "sync")// Pool goroutine Pooltype 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 里面取数据进行消费，这样可以避免协程的创建与注销。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
|
package mainimport ( "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实现协程池内容请搜索服务器之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持服务器之家！
原文链接：https://www.cnblogs.com/niuben/p/15006362.html
