Goroutine 的优势和缺陷#

Goroutine 是 Go 语言内部实现的高并发协程模型，并且可以通过 Channel 进行快捷便利的通信。

但是也有一些缺陷：一，不支持限制 Goroutine 的数量。二，在运行完毕之后不会立即被GC（Garbage Collection）回收。

👉More About Goroutine

Go Pool#

Diagram

Pool 中主要包含三种部件：

Worker - 实际执行函数
JobsChan - 将 job 提交给 Worker
TasksChan - 用户将任务提交到 TasksChan，Pool 再将任务提交到 JobsChan。

另外需要定义 Task 结构，至少应该包含需要执行的函数。一个样例如下:

fn - 需要执行的函数
submitTime - 提交时间
User - 提交用户
taskId - 任务ID
…

通过Pool，我们可以限制goroutine的数量，并且可以复用goroutine。在一些并发量极大，单个goroutine执行任务并不繁重的情况下，可以节约很多内存。

因为虽然goroutine的初始内存只有2kb，但是如果并发量达到十万、百万级别，内存的消耗也是不容小觑的。当任务并不繁重时，一些goroutine可能被初始化之后，只执行了几行代码就被搁置了，但没有及时的GC。大量的died goroutine不必要地占用了大量内存资源。池化后，可以达到线程复用，从而节约内存。

Implementation#

1
// The implementaion of a simple goroutines pool with fixed capacity
2
// By dicer 2/25/2021
3

4
package main
5

6
import (
7
   "fmt"
8
   "time"
9
)
10

11
// Define Task
12
type Task struct{
13
   fn func() error
14
   submitTime time.Time
15
   // Alternative property
16
   // #user
17
   // #taskID
18
}
19

20
// Return a new task with a function argFn
21
func NewTask(argFn func() error) *Task {
22
   t := Task{
23
      fn:         argFn,
24
      submitTime: time.Now(),
25
   }
26
   return &t
27
}
28

29
// Execute the fn of a Task
30
func (t *Task) Execute() {
31
   t.fn()
32
}
33

34
// Define Pool
35
type Pool struct {
36
   TasksChan chan *Task
37
   JobsChan chan *Task
38
   Cap uint
39
}
40

41
// Return a new pool with fixed capacity
42
func NewPool(cap uint) *Pool {
43
   p := Pool{
44
      TasksChan: make(chan *Task, cap),
45
      JobsChan: make(chan *Task, cap),
46
      Cap: cap,
47
   }
48
   return &p
49
}
50

51
// Generate a worker entity used as a coroutine
52
func (p *Pool) Worker(WorkerID int) {
53
   // take a task from JobsChan
54
   for t := range p.JobsChan {
55
      // execute this task
56
      t.Execute()
57
      // print info
58
      fmt.Printf("Worker %d has finished a task submited at %v.\n", WorkerID, t.submitTime)
59
   }
60
}
61

62
// Run the pool with cap_size worker
63
func (p *Pool) Run() {
64

65
   for i := 0; uint(i) < p.Cap; i++ {
66
      go p.Worker(i)
67
   }
68

69
   for  {
70
      p.JobsChan <- <- p.TasksChan
71
   }
72
   fmt.Println("Debug")
73
}
74

75
// Test the Pool
76
func main() {
77
   t := NewTask(func() error {
78
      _, err := fmt.Println(time.Now())
79
      return err
80
   })
81

82
   p := NewPool(5)
83
   go p.Run()
84

85
   for {
86
      p.TasksChan <- t
87
   }
88
}