手把手教你Golang的协程池设计

前言

 

现在很多公司都在陆续的搭建golang的语言栈，大家有没有想过为什么会出现这种情况?一是因为go比较适合做中间件，还有一个原因就是go的并发支持比较好，也就是咱们平时所谓的高并发，并发支持离不开协程，当然协程也不是乱用的，需要管理起来，管理协程的方式就是协程池，所以协程池也并没有那么神秘，今天咱们就来一步一步的揭开协程池的面纱，如果你没有接触过go的协程这块的话也没有关系，我会尽量写的详细。

goroutine(协程)

 

先来看一个简单的例子

1. func go_worker(name string) { 
2.    for i := 0; i < 5; i++ { 
3.     fmt.Println("我的名字是", name) 
4.     time.Sleep(1 * time.Second) 
5.    } 
6.    fmt.Println(name, "执行完毕") 
7. } 
8. func main() { 
9.     go_worker("123") 
10.     go_worker("456") 
11.    for i := 0; i < 5; i++ { 
12.     fmt.Println("我是main") 
13.     time.Sleep(1 * time.Second) 
14.    } 
15. } 

咱们在执行这段代码的时候，当然是按照顺序执行

go_worker("123")->go_worker("456")->我是main执行

输出结果如下

1. 我的名字是 123 
2. 我的名字是 123 
3. 我的名字是 123 
4. 我的名字是 123 
5. 我的名字是 123 
6. 123 执行完毕 
7. 我的名字是 456 
8. 我的名字是 456 
9. 我的名字是 456 
10. 我的名字是 456 
11. 我的名字是 456 
12. 456 执行完毕 
13. 我是main 
14. 我是main 
15. 我是main 
16. 我是main 
17. 我是main 

这样的执行是并行的，也就是说必须得等一个任务执行结束，下一个任务才会开始，如果某个任务比较慢的话，整个程序的效率是可想而知的，但是在go语言中，支持协程，所以我们可以把上面的代码改造一下

1. func go_worker(name string) { 
2.    for i := 0; i < 5; i++ { 
3.     fmt.Println("我的名字是", name) 
4.     time.Sleep(1 * time.Second) 
5.    } 
6.    fmt.Println(name, "执行完毕") 
7. } 
8. func main() { 
9.    go go_worker("123")  //协程 
10.    go go_worker("456")  //协程 
11.    for i := 0; i < 5; i++ { 
12.     fmt.Println("我是main") 
13.     time.Sleep(1 * time.Second) 
14.    } 
15. } 

我们在不同的go_worker前面加上了一个go，这样所有任务就异步的串行了起来，输出结果如下

1. 我是main 
2. 我的名字是 456 
3. 我的名字是 123 
4. 我的名字是 123 
5. 我是main 
6. 我的名字是 456 
7. 我是main 
8. 我的名字是 456 
9. 我的名字是 123 
10. 我是main 
11. 我的名字是 456 
12. 我的名字是 123 
13. 我的名字是 456 
14. 我的名字是 123 
15. 我是main 

大家可以看到这样的话就是各自任务执行各自的事情，互相不影响，效率也得到了很大的提升，这就是goroutine

channel(管道)

 

有了协程之后就会带来一个新的问题，协程之间是如何通信的?于是就引出了管道这个概念，管道其实很简单，无非就是往里放数据，往外取数据而已

1. func worker(c chan int) { 
2.    num := <-c  //读取管道中的数据，并输出 
3.    fmt.Println("接收到参数c:", num) 
4. } 
5. func main() { 
6.    //channel的创建,需要执行管道数据的类型，我们这里是int 
7.    c := make(chan int) 
8.    //开辟一个协程 去执行worker函数 
9.    go worker(c) 
10.    c <- 2  //往管道中写入2 
11.    fmt.Println("main") 
12. } 

我们可以看到上述例子，在main函数中，我们定义了一个管道，为int类型，而且往里面写入了一个2，然后在worker中读取管道c，就能获取到2

协程会引发的问题

 

既然golang中开启协程这么方便，那么会不会存在什么坑呢?

我们可以看上图，实际业务中，不同的业务都开启不同的goroutine来执行，但是在cpu微观层面上来讲，是串行的一个指令一个指令去执行的，只是执行的非常快而已，如果指令来的太多，cpu的切换也会变多，在切换的过程中就需要消耗性能，所以协程池的主要作用就是管理goroutine，限定goroutine的个数

协程池的实现

 

• 首先不同的任务，请求过来，直接往entryChannel中写入，entryChannel再和jobsChannel建立通信
• 然后我们固定开启三个协程(不一定是三个，只是用三个举例子)，固定的从jobsChannel中读取数据，来进行任务处理。
• 其实本质上，channel就是一道桥梁，做一个中转的作用，之所以要设计一个jobsChannel和entryChannel，是为了解耦，entryChannel可以完全用做入口，jobsChannel可以做更深入的比如任务优先级，或者加锁，解锁等处理

代码实现

 

原理清楚了，接下来我们来具体看代码实现

首先我们来处理任务 task，task无非就是业务中的各种任务，需要能实力化，并且执行，代码如下

1. //定义任务Task类型,每一个任务Task都可以抽象成一个函数 
2. type Task struct{ 
3.    f func() error //一个task中必须包含一个具体的业务 
4. } 
5.  
6.  
7. //通过NewTask来创建一个Task 
8. func NewTask(arg_f func() error) *Task{ 
9.    t := Task{ 
10.     f:arg_f, 
11.    } 
12.    return &t 
13. } 
14.  
15.  
16. //Task也需要一个执行业务的方法 
17. func (t *Task) Execute(){ 
18.    t.f()//调用任务中已经绑定好的业务方法 
19. } 

接下来我们来定义协程池

1. //定义池类型 
2. type Pool struct{ 
3.    EntryChannel chan *Task 
4.    WorkerNum int 
5.    JobsChanel chan *Task 
6. } 
7. //创建一个协程池 
8. func NewPool(cap int) *Pool{ 
9.    p := Pool{ 
10.     EntryChannel: make(chan *Task), 
11.     JobsChanel: make(chan *Task), 
12.     WorkerNum: cap, 
13.    } 
14.    return &p 
15. } 

协程池需要创建worker，然后不断的从JobsChannel内部任务队列中拿任务开始工作

1. //协程池创建worker并开始工作 
2. func (p *Pool) worker(workerId int){ 
3.     //worker不断的从JobsChannel内部任务队列中拿任务 
4.     for task := range p.JobsChanel{ 
5.      task.Execute() 
6.      fmt.Println("workerId",workerId,"执行任务成功") 
7.     } 
8. } 
9. EntryChannel获取Task任务 
10. func (p *Pool) ReceiveTask(t *Task){ 
11.    p.EntryChannel <- t 
12. } 
13. //让协程池开始工作 
14. func (p *Pool) Run(){ 
15.    //1:首先根据协程池的worker数量限定，开启固定数量的worker 
16.    for i:=0; i<p.WorkerNum; i++{ 
17.     go p.worker(i) 
18.    } 
19.    //2:从EntryChannel协程出入口取外部传递过来的任务 
20.    //并将任务送进JobsChannel中 
21.    for task := range p.EntryChannel{ 
22.     p.JobsChanel <- task 
23.    } 
24.    //3:执行完毕需要关闭JobsChannel和EntryChannel 
25.    close(p.JobsChanel) 
26.    close(p.EntryChannel) 
27. } 

然后我们看在main函数中

1. //创建一个task 
2.    t:= NewTask(func() error{ 
3.     fmt.Println(time.Now()) 
4.     return nil 
5.    }) 
6.  
7.    //创建一个协程池，最大开启5个协程worker 
8.    p:= NewPool(3) 
9.    //开启一个协程，不断的向Pool输送打印一条时间的task任务 
10.    go func(){ 
11.     for { 
12.      p.ReceiveTask(t)//把任务推向EntryChannel 
13.     } 
14.    }() 
15.    //启动协程池p 
16.    p.Run() 

基于上述方法，咱们一个简单的协程池设计就完成了，当然在实际生产环境中这样做还是不够的，不过这些方法能手写出来，那对golang是相当熟悉了，

原文链接：https://mp.weixin.qq.com/s/nKztiA9eErBdbIpL_0WAFg