深入浅出Golang的协程池设计

 1package main
 2
 3import "fmt"
 4import "time"
 5
 6func go_worker(name string) {
 7        for i := 0; i < 10; i++ {
 8                fmt.Println("我是一个go协程, 我的名字是 ", name, "----")
 9                time.Sleep(1 * time.Second)
10        }
11        fmt.Println(name, " 执行完毕!")
12}
13
14func main() {
15        go go_worker("小黑")  //创建一个goroutine协程去执行 go_worker("小黑")
16        go go_worker("小白")  //创建一个goroutine协程去执行 go_worker("小白")
17
18        //防止main函数执行完毕,程序退出
19        for {
20                time.Sleep(1 * time.Second)
21        }
22}

 1package main
 2
 3import "fmt"
 4
 5func worker(c chan int) {
 6        //从channel中去读数据
 7        num := <-c
 8        fmt.Println("foo recv channel ", num)
 9}
10
11func main() {
12        //创建一个channel
13        c := make(chan int)
14
15        go worker(c)
16
17        //main协程 向一个channel中写数据
18        c <- 1
19
20        fmt.Println("send 1 -> channel over")
21}

  1package main
  2
  3import (
  4        "fmt"
  5        "time"
  6)
  7
  8/* 有关Task任务相关定义及操作 */
  9//定义任务Task类型,每一个任务Task都可以抽象成一个函数
 10type Task struct {
 11        f func() error //一个无参的函数类型
 12}
 13
 14//通过NewTask来创建一个Task
 15func NewTask(f func() error) *Task {
 16        t := Task{
 17                f: f,
 18        }
 19
 20        return &t
 21}
 22
 23//执行Task任务的方法
 24func (t *Task) Execute() {
 25        t.f() //调用任务所绑定的函数
 26}
 27
 28/* 有关协程池的定义及操作 */
 29//定义池类型
 30type Pool struct {
 31        //对外接收Task的入口
 32        EntryChannel chan *Task
 33
 34        //协程池最大worker数量,限定Goroutine的个数
 35        worker_num int
 36
 37        //协程池内部的任务就绪队列
 38        JobsChannel chan *Task
 39}
 40
 41//创建一个协程池
 42func NewPool(cap int) *Pool {
 43        p := Pool{
 44                EntryChannel: make(chan *Task),
 45                worker_num:   cap,
 46                JobsChannel:  make(chan *Task),
 47        }
 48
 49        return &p
 50}
 51
 52//协程池创建一个worker并且开始工作
 53func (p *Pool) worker(work_ID int) {
 54        //worker不断的从JobsChannel内部任务队列中拿任务
 55        for task := range p.JobsChannel {
 56                //如果拿到任务,则执行task任务
 57                task.Execute()
 58                fmt.Println("worker ID ", work_ID, " 执行完毕任务")
 59        }
 60}
 61
 62//让协程池Pool开始工作
 63func (p *Pool) Run() {
 64        //1,首先根据协程池的worker数量限定,开启固定数量的Worker,
 65        //  每一个Worker用一个Goroutine承载
 66        for i := 0; i < p.worker_num; i++ {
 67                go p.worker(i)
 68        }
 69
 70        //2, 从EntryChannel协程池入口取外界传递过来的任务
 71        //   并且将任务送进JobsChannel中
 72        for task := range p.EntryChannel {
 73                p.JobsChannel <- task
 74        }
 75
 76        //3, 执行完毕需要关闭JobsChannel
 77        close(p.JobsChannel)
 78
 79        //4, 执行完毕需要关闭EntryChannel
 80        close(p.EntryChannel)
 81}
 82
 83//主函数
 84func main() {
 85        //创建一个Task
 86        t := NewTask(func() error {
 87                fmt.Println(time.Now())
 88                return nil
 89        })
 90
 91        //创建一个协程池,最大开启3个协程worker
 92        p := NewPool(3)
 93
 94        //开一个协程 不断的向 Pool 输送打印一条时间的task任务
 95        go func() {
 96                for {
 97                        p.EntryChannel <- t
 98                }
 99        }()
100
101        //启动协程池p
102        p.Run()
103
104}