waitgroup+channel控制goroutine并发数量

60 阅读 0 评论 40 点赞

我是靠谱客的博主勤劳冥王星，最近开发中收集的这篇文章主要介绍waitgroup+channel控制goroutine并发数量，觉得挺不错的，现在分享给大家，希望可以做个参考。

概述

版本一：

package main 
import (
      "fmt"
      "runtime"
      "sync"
)
var wg = sync.WaitGroup{}
// 任务业务流程
func business(ch chan bool, i int) {
  fmt.Println("go func", i, " goroutine count = ", runtime.NumGoroutine)
  <-ch
  wg.Done()
}

func main() {
  // 模拟用户需求的业务数量
  task_cnt := 10
  ch := make(chan bool, 3)
  for i := 0; i < taskk_cnt; i++ {
    wg.Add(1)
    // 如果channel满了，就会阻塞
    ch <- true  
    // 开启一个新协程
    go business(ch, i)
  }
  wg.Wait()
}

版本二：

package main 
import (
      "fmt"
      "runtime"
      "sync"
)
var wg = sync.WaitGroup{}
// 每个go的worker都要执行的一个工作流程
func business(ch chan int){
    // 消费一个任务
    for t := range ch {
        fmt.Println(" go task = ", t, ", goroutine count = ", runtime.NumGoroutine())
      wg.Done()
  }
}

// 发送一个任务（任务的输入，任务的生产）
func sendTask(task int, ch chan int) {
    wg.Add(1)
    ch <- task
}
func main() {
    // 无buffer的channel
    ch := make(chan int)
    // 1 启动goroutine工作池（go的数量是固定的）充当任务task的消费
    goCnt := 3
    for i := 0; i < goCnt; i++ {
      // 启动goroutine的worker
      go business(ch)
    }
    // 2模拟用户需求业务的数量，不断的给工作池发送task
    taskCnt := math.MaxInt64
    for t := 0; t < taskCnt; t++ {
        // 发送任务
        sendTask(t, ch)
    }
    wg.Wait()
}

版本三：

package gpool

import (
    "sync"
)

type pool struct {
    queue chan int
    wg    *sync.WaitGroup
}

func New(size int) *pool {
    if size <= 0 {
        size = 1
    }
    return &pool{
        queue: make(chan int, size),
        wg:    &sync.WaitGroup{},
    }
}

func (p *pool) Add(delta int) {
    for i := 0; i < delta; i++ {
        p.queue <- 1
    }
    for i := 0; i > delta; i-- {
        <-p.queue
    }
    p.wg.Add(delta)
}

func (p *pool) Done() {
    <-p.queue
    p.wg.Done()
}

func (p *pool) Wait() {
    p.wg.Wait()
}

测试代码：

package gpool_test

import (
    "runtime"
    "testing"
    "time"
    "gpool"
)

func Test_Example(t *testing.T) {
    pool := gpool.New(100)
    println(runtime.NumGoroutine())
    for i := 0; i < 1000; i++ {
        pool.Add(1)
        go func() {
            time.Sleep(time.Second)
            println(runtime.NumGoroutine())
            pool.Done()
        }()
    }
    pool.Wait()
    println(runtime.NumGoroutine())
}