利用有缓冲通道写满时阻塞的特点来控制协程的并发数量

利用有缓冲通道写满时阻塞的特点来控制协程的并发数量

type LineGraber struct {
	//最多允许n个并发同时执行
	ConcurrentNum int `json:"concurrent_num"`
}

func (grab LineGraber) Work() {

// "work start"

//最少5个协程一起工作
concurrentNum := grab.ConcurrentNum
if concurrentNum <= 0 {
    concurrentNum = 5
}
// sem通道容量限制了并发的协程数
sem := make(chan struct{}, concurrentNum)

// 准备待处理的任务列表  
taskLst := grab.buildTaskLst()

// 接口任务处理结果的通道
chanRespv := make(chan string, len(taskLst))

total := len(taskLst)
// log info fmt.Sprintf("tasksize:%d", total) 

var wg sync.WaitGroup
for idx, key := range taskLst {
    wg.Add(1)
    go func(taskv string, i int) {
   // 确保 wg.Done() (与add对应), 
   //     <-sem, 从semchan中取出(与下面 sem<- struct{}{}对应)
        defer func() {
            wg.Done()
            <-sem
        }()
        sem <- struct{}{}
        // 真实业务逻辑部分 
        // 方法中要将结果写入到chanRespv 
        // log info i, taskv 
        grab.processTask(taskv, oneproxy, chanRespv)

    }(key, idx)
}

wg.Wait()
close(chanRespv)

// range 可以从已经关闭的chan 中读取原有数据, 并退出  
for respv := range chanRespv {
    // log info read from chanResp
    var respResult RespResult
    err := json.Unmarshal([]byte(respv), &respResult)
    if err != nil {
        // log error decode异常 
        continue
    }
    // process ... 

}

// log info ("work done")

}

// 以上方法中在 wg.Wait之后才开始接收 chanRespv ,其实还可以优化\尽早提前处理chanRespv(可以及时读取到结果,避免所有任务完成后才读取到结果的情况), 只需新启协程在wg.Wait 之前接收处理 . 
// 要增加另外一个通道, 控制完成并退出的信号

// 在wg.wait 之前 
chanSignal := make(chan interface{})
go func() {
    for {
        select {
        case result, ok := <-limiter.resultChan:
            common.ZLogger.Info("done", zap.Any("result", result))
            if !ok {
                common.ZLogger.Info("result通道关闭,退出当前goroutine")
      // 通知主线程,可以继续向下执行了 
                chanSignal<- 1
                return
            }
        }
    }
}()

// 在程序最后添加 
<-chanSignal
	

// 业务逻辑处理  
// 注意chan 类型是只可写入 chan <-  string  
func (grab LineGraber) processTask(key string, proxy string, outchan chan<- string) {
	aimUrl := fmt.Sprintf(grap_base_url, key)
	respv, err := doget... 
	if err != nil {
		// log error ...
	}
	//log info ... 
	outchan <- respv
}