多协程如何使用channel优雅的收集结果

小锟哥哥

发布于 2022-05-10 08:17:32

7520

发布于 2022-05-10 08:17:32

文章被收录于专栏：GoLang全栈

前言

Go语言里面最具特色的就是他的协程和 channel ，有了它们以后我们可以非常方便的处理多线程的问题。

但是随之而来的问题就是，有些时候我们需要同时执行多个协程，然后再根据其结果再进行处理，这时候收集多个协程的值就非常关键。

这篇文章我们一起来实现从一个小白到优雅的处理这个问题的方式。

小白的演进

先来一个基础核心：

//模拟耗时操作
func job(number int) int {
 time.Sleep(time.Millisecond * 500)
 return number
}
func main() {
 fmt.Println(job(1))
}

我们用 job 方法来模拟耗时的方法，现在需要执行多次，在不使用协程的情况变成了这样：

//模拟耗时操作
func job(number int) int {
 time.Sleep(time.Millisecond * 500)
 return number
}
func main() {
 start := time.Now()
 num := 5
 for i:=0; i< num; i++ {
  fmt.Println(job(i))
 }
 end := time.Since(start)
 fmt.Println("总共耗时：",end.String())
}

我们在这里加入执行时间统计，执行结果可能是这样的：

0
1
2
3
4
总共耗时：2.512076777s

现在我们加入协程，我们这里直接使用 sync.WaitGroup 来管理协程。

//模拟耗时操作
func job(number int) int {
 time.Sleep(time.Millisecond * 500)
 return number
}
func main() {
 start := time.Now()
 num := 5
 wg := sync.WaitGroup{}
 for i:=0; i< num; i++ {
  wg.Add(1)
  go func(input int) {
   defer wg.Done()
   fmt.Println(job(input))
  }(i)
 }
 wg.Wait()
 end := time.Since(start)
 fmt.Println("总共耗时：",end.String())
}

执行结果大概是这样的：

2
1
3
0
4
总共耗时：502.8991ms

时间明显缩短了。

但是这里我们是直接在协程里面把结果打印出来，并未收集到 channel 里面，下面我们收集起来。

大部分人可能会这样做：

//模拟耗时操作
func job(number int) int {
 time.Sleep(time.Millisecond * 500)
 return number
}
func main() {
 start := time.Now()
 num := 5
 resCha := make(chan int) //用于收集结果的channel
 for i:=0; i< num; i++ {
  go func(input int) {
   resCha <- job(input)
  }(i)
 }
 cnt := 0
 for item := range resCha {
  cnt++
  fmt.Println("收到的结果：",item)
  
  // 这里进行判断是否遍历完了，依次来考虑是否需要关闭channel跳出循环。
  // 如果不关闭channel会报死锁
  if cnt == num {
   close(resCha)
   break
  }
 }
 end := time.Since(start)
 fmt.Println("总共耗时：",end.String())
}

优雅的处理

有没有比较好的方法，可以在取数据时不关注 channel 里面的协程是否处理完了呀？

答案肯定是有的。

改良下，优雅的收集结果，代码如下：

//模拟耗时操作
func job(number int) int {
 time.Sleep(time.Millisecond * 500)
 return number
}
func main() {
 start := time.Now()
 num := 5
 resCha := make(chan int) //收集结果的channel
 wg := sync.WaitGroup{}
 for i:=0; i< num; i++ {
  wg.Add(1)
  go func(input int) {
   defer wg.Done()
   resCha <- job(input)
  }(i)
 }

 // 再开一个协程等到执行完毕
 go func() {
  // 这个方法执行完毕后关闭通道
  defer close(resCha)
  wg.Wait()
 }()

 for item := range resCha {
  fmt.Println("收到的结果：",item)
 }

 end := time.Since(start)
 fmt.Println("总共耗时：",end.String())
}

其原理也很简单：

开一个协程跑 wg.Waite 就好了，因为它能确保被挂在 wg 里面的所有协程都执行完毕了才会被执行，有点想监工的感觉。

如果熟悉 JavaScript 的同学，就会发现这个和里面的 Promise.all() 很像。

这样你就可以不用管协程他啥时候执行完毕了，你只管写你的业务逻辑就好了。

本文参与腾讯云自媒体同步曝光计划，分享自微信公众号。

原始发表：2021-08-23，如有侵权请联系 cloudcommunity@tencent.com 删除

javascript