用go实现一个协程池，大概用什么实现

题目序号：773

题目来源：网易

频次：1

答案：Carpe-Wang

定义一个task 的结构体 标示具体要执行的任务格式

type Job func([]interface{})type taskWork struct {
        Run       Job    
        startBool bool    
        params    []interface{}
}

定义一个worker 池，控制协程相关信息

type WorkPool struct {
    taskPool  chan taskWork
    workNum   int
    maxNum    int
    stopTopic bool
    //考虑后期 作为冗余队列使用
    taskQue chan taskWork
}

实现协程池相关启动，停止，扩容策略，缩减策略，备用队列启用等 逻辑

//得到一个线程池并返回 句柄
func (p *WorkPool) InitPool() {
    *p = WorkPool{workNum: workerNumDefault,
        maxNum: workerNumMax, stopTopic: false,
        taskPool: make(chan taskWork, workerNumDefault*2), taskQue: nil}
 
    (p).start()
    go (p).workerRemoveConf()
}
 
//开始work
func (p *WorkPool) start() {
    for i := 0; i < workerNumDefault; i++ {
        p.workInit(i)
        fmt.Println("start pool task:", i)
    }
}
 
//初始化 work池 后期应该考虑如何 自动 增减协程数，以达到最优
func (p *WorkPool) workInit(id int) {
    go func(idNum int) {
        //var i int = 0
        for {
            select {
            case task := <-p.taskPool:
                if task.startBool == true && task.Run != nil {
                    //fmt.Print("this is pool ", idNum, "---")
                    task.Run(task.params)
                }
                //单个结束任务
                if task.startBool == false {
                    //fmt.Print("this is pool -- ", idNum, "---")
                    return
                }
                //防止从channal 中读取数据超时
            case <-time.After(time.Millisecond * 1000):
                //fmt.Println("time out init")
                if p.stopTopic == true && len(p.taskPool) == 0 {
                    fmt.Println("topic=", p.stopTopic)
                    //work数递减
                    p.workNum--
                    return
                }
                //从备用队列读取数据
            case queTask := <-p.taskQue:
                if queTask.startBool == true && queTask.Run != nil {
                    //fmt.Print("this is que ", idNum, "---")
                    queTask.Run(queTask.params)
                }
            }
 
        }
    }(id)
 
}
 
//停止一个workPool
func (p *WorkPool) Stop() {
    p.stopTopic = true
}
 
//普通运行实例，非自动扩充
func (p *WorkPool) Run(funcJob Job, params ...interface{}) {
    p.taskPool <- taskWork{funcJob, true, params}
}
 
//用select 去做 实现 自动扩充 协程个数 启用备用队列等特性
func (p *WorkPool) RunAuto(funcJob Job, params ...interface{}) {
    task := taskWork{funcJob, true, params}
    select {
    //正常写入
    case p.taskPool <- task:
        //写入超时 说明队列满了 写入备用队列
    case <-time.After(time.Millisecond * 1000):
        p.taskQueInit()
        p.workerAddConf()
        //task 入备用队列
        p.taskQue <- task
    }
}
 
//自动初始化备用队列
func (p *WorkPool) taskQueInit() {
    //扩充队列
    if p.taskQue == nil {
        p.taskQue = make(chan taskWork, p.maxNum*2)
    }
}
 
//自动扩充协程 简单的自动扩充策略
func (p *WorkPool) workerAddConf() {
    //说明需要扩充进程  协程数量小于 1000 协程数量成倍增长
    if p.workNum < 1000 {
        p.workerAdd(p.workNum)
    } else if p.workNum < p.maxNum {
        tmpNum := p.maxNum - p.workNum
        tmpNum = tmpNum / 10
        if tmpNum == 0 {
            tmpNum = 1
        }
        p.workerAdd(1)
    }
}
 
//自动缩减协程 实现比较粗糙，可以考虑后续精细实现一些策略
func (p *WorkPool) workerRemoveConf() {
    for {
        select {
        case <-time.After(time.Millisecond * 1000 * 600):
            if p.workNum > workerNumDefault && len(p.taskPool) == 0 && len(p.taskQue) == 0 {
                rmNum := (p.workNum - workerNumDefault) / 5
                if rmNum == 0 {
                    rmNum = 1
                }
                p.workerRemove(rmNum)
            }
        }
    }
 
}
func (p *WorkPool) workerAdd(num int) {
    for i := 0; i < num; i++ {
        p.workNum++
        p.workInit(p.workNum)
    }
}
func (p *WorkPool) workerRemove(num int) {
    for i := 0; i < num; i++ {
        task := taskWork{startBool: false}
        p.taskPool <- task
        p.workNum--
    }
}