Go并发编程（三）协程池

我不是女神ヾ 2023-01-18 06:51 268阅读 0赞

### 文章目录 ###

*  Go并发编程（三）协程池
 *   *  为什么需要协程池
     *  实现
     *   *  数据结构定义
         *  新增任务&执行任务
         *  goroutine异常处理
         *  关闭协程池
         *  使用

# Go并发编程（三）协程池 #

本文参考如下博客实现了一个简易的协程池

*  [100 行写一个 go 的协程池 (任务池)][100 _ go _]

## 为什么需要协程池 ##

goroutine 太多仍会导致调度性能下降、GC 频繁、内存暴涨, 引发一系列问题。在面临这样的场景时, 限制 goroutine 的数量、重用 goroutine

## 实现 ##

实现的基本思路是采用生产者-消费者模型，用来执行任务的goroutine作为消费者，操作任务队列的goroutine是生产者，任务队列使用的是go中的buffer channel

![在这里插入图片描述][20210421004451118.png]

### 数据结构定义 ###

任务定义：

// 任务定义
    type Task struct {
        
    	Handler func(v ...interface{
        }) // 任务处理函数
    	Params []interface{
        } 		   // 处理函数参数列表
    }

协程池定义：

// 任务池定义
    type TaskPool struct {
        
    	Capacity int64 		   		   // 任务池容量
    	RunningGoroutine int64 		   // 运行中的goroutine数量
    	TaskQueue chan *Task   		   // 任务队列
    	Status int64 		     	   // 任务池状态
    	sync.Mutex
    	PanicHandler func(interface{
        }) // goroutine异常处理机制
    }

协程池状态常量定义

// 协程池状态
    const(
    	RUNNING = iota
    	STOP
    )

全局异常定义：

// 池容量非法异常
    var ErrInvalidPoolCap = errors.New("task pool capacity invaild")
    var ErrPoolAlreadyClosed = errors.New("pool is already go")

### 新增任务&执行任务 ###

新增任务本质就是做goroutine数量检查，小于协程池容量则新启协程，超过就复用原有协程，协程的回收依赖于GC，任务是直接丢进管道，等待消费的goroutine执行

// 新增任务
    func (p *TaskPool) Put(t *Task) error{
        
    	p.Lock()
    	defer p.Unlock()
    
    	if p.Status == STOP{
        
    		return ErrPoolAlreadyClosed
    	}
    
    	// 如果协程池未满则新启协程
    	if p.RunningGoroutine < p.Capacity{
        
    		// 协程池未满，则产生协程
    		p.run()
    	}
    	// 任务入队
    	p.TaskQueue <- t
    
    	return nil
    }

执行任务其实就是监听channel消费具体的任务，这里采用的是带缓冲区的channel，所以消费生产是非阻塞的

// 从任务队列中取出任务执行
    func (pool *TaskPool)run()  {
        
    	// 新增运行中的goroutine
    	incRunning(pool)
    	go func() {
        
    		// 执行完成后运行中的goroutine--
    		defer func() {
        
    			decRunning(pool)
    			// goroutine panic
    			if r := recover();r != nil{
        
    				if pool.PanicHandler != nil{
        
    					pool.PanicHandler(r);
    				} else {
         // 默认处理
    					log.Printf("Worker panic: %s\n", r)
    				}
    			}
    			pool.checkRunningWork()
    		}()
    		// 具体goroutine执行策略
    		for{
        
    			select {
        
    			case task,ok := <- pool.TaskQueue:{
        
    				if !ok{
        
    					// 任务从管道消费失败
    					return
    				}
    				// 执行任务
    				task.Handler(task.Params)
    			}
    			}
    		}
    	}()
    }

### goroutine异常处理 ###

如果某一个goroutine抛出panic就会导致整个程序崩溃退出，为了保证程序安全执行，需要对panic进行recover，进行异常处理，异常处理函数用户自定义

defer func() {
        
    			decRunning(pool)
    			// goroutine panic
    			if r := recover();r != nil{
        
    				if pool.PanicHandler != nil{
        
    					pool.PanicHandler(r);
    				} else {
         // 默认处理
    					log.Printf("Worker panic: %s\n", r)
    				}
    			}
    			pool.checkRunningWork()
    		}()

### 关闭协程池 ###

关闭协程池需要做两个步骤：

1.  关闭任务进入队列的入口
2.  执行完任务队列中剩余的任务

// 安全关闭协程池
    func (p *TaskPool) CloseTask() error{
        
    	p.Lock()
    	defer p.Unlock()
    
    	if p.Status == STOP{
        
    		return ErrPoolAlreadyClosed
    	}
    
    	atomic.CompareAndSwapInt64(&p.Status,RUNNING,STOP)
    
    	// 清空任务队列
    	for len(p.TaskQueue) > 0 {
         // 阻塞等待所有任务被 worker 消费
    		time.Sleep(1e6) // 防止等待任务清空 cpu 负载突然变大, 这里小睡一下
    	}
    
    	return nil
    }

### 使用 ###

func TestMyPool()  {
        
    	pool,err := InitTaskPool(10)
    	if err != nil{
        
    		panic(err)
    	}
    	for i := 0;i < 20;i++{
        
    		time.Sleep(1e6)
    		pool.Put(&Task{
        Handler: func(v ...interface{
        }) {
        
    			fmt.Print("i = ",i," ")
    		},Params: []interface{
        }{
        i}})
    		fmt.Println("pool running goroutine size: ",pool.GetPoolRunningGSize())
    	}
    
    }

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTkyMjI4OQ_size_16_color_FFFFFF_t_70]

[100 _ go _]: https://segmentfault.com/a/1190000021468353
[20210421004451118.png]: /images/20221021/2d2ec9edd0514e848cc6c11a9a6e28ca.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTkyMjI4OQ_size_16_color_FFFFFF_t_70]: /images/20221021/315db75f972048a487c340a867e20669.png