Go goroutine 池

本文将介绍如何在 Go 中实现一个简易的 goroutine 池。

一、为什么需要 goroutine 池？

相比于操作系统线程而言，goroutine 具有资源占用小、管理开销小、调度开销小等特点。因此，Go 应用通常为每个新建立的连接、每个请求创建一个新的 goroutine。

但是，goroutine 的开销虽然小，但也不是完全没有开销，大量的 goroutine 也会带来同样大量的开销。

二、goroutine 池的实现

goroutine 池的实现主要分为三个部分：

  • 池的创建和销毁
  • goroutine 的管理
  • 任务的提交和调度
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
type Pool struct {
	active chan struct{}  // goroutine计数channel；其大小对应池的goroutine数；增删goroutine时，应该对其写读数据
	tasks  chan Task      // 任务channel
	wg     sync.WaitGroup // pool销毁时，通过它等待所有goroutine退出
	quit   chan struct{}  // 退出信号channel
}

type Task func()

// New 创建goroutine池
func New(capacity int) *Pool {
	p := &Pool{
		tasks:  make(chan Task),
		quit:   make(chan struct{}),
		active: make(chan struct{}, capacity),
	}

	go p.run()

	return p
}

// Schedule 提交任务
func (p *Pool) Schedule(t Task) error {
	select {
	// 若读取到退出信号，报错
	case <-p.quit:
		return errors.New("pool stopped")
	// 将任务放入任务channel中
	case p.tasks <- t:
		return nil
	}
}

// run 运行goroutine池
func (p *Pool) run() {
	for {
		select {
		// 若读取到退出信号，停止轮询
		case <-p.quit:
			return
		// 若active channel可写，创建新线程
		case p.active <- struct{}{}:
			p.newWorker()
		}
	}
}

// newWorker 创建新的goroutine
func (p *Pool) newWorker() {
	p.wg.Add(1)
	go func() {
		defer func() {
			if err := recover(); err != nil {
				fmt.Printf("worker panic and exit: [%s]", err)
			}
			<-p.active
			p.wg.Done()
		}()

		for {
			select {
			// 若读取到退出信号，停止轮询
			case <-p.quit:
				return
			// 读取任务channel并执行
			case t := <-p.tasks:
				t()
			}
		}
	}()
}

参考