讲透Go中的并发接收控制结构Select

select 与 switch

让我们来复习一下switch语句，在switch语句中，会逐个匹配case语句(可以是值也可以是表达式)，一个一个的判断过去，直到有符合的语句存在，执行匹配的语句内容后跳出switch。

坚守“ 做人真诚 · 做事靠谱 · 口碑至上 · 高效敬业 ”的价值观，专业网站建设服务10余年为成都食品包装袋小微创业公司专业提供企业网站建设营销网站建设商城网站建设手机网站建设小程序网站建设网站改版，从内容策划、视觉设计、底层架构、网页布局、功能开发迭代于一体的高端网站建设服务。

 
 
 
 
    
  
  
  
  
  1. func demo(number int){ 
    2.   
  
  
  
  2.     switch{ 
    3.   
  
  
  
  3.         case number >= 90: 
    4.   
  
  
  
  4.         fmt.Println("优秀") 
    5.   
  
  
  
  5.         default: 
    6.   
  
  
  
  6.         fmt.Println("太搓了") 
    7.   
  
  
  
  7.     } 
    8.   
  
  
  
  8. } 
    9.

而 select 用于处理通道，它的语法与 switch 非常类似。每个 case 语句里必须是一个 channel 操作。它既可以用于 channel 的数据接收，也可以用于 channel 的数据发送。

 
 
 
 
    
  
  
  
  
  1. func foo() { 
    2.   
  
  
  
  2.  chanInt := make(chan int) 
    3.   
  
  
  
  3.  defer close(chanInt) 
    4.   
  
  
  
  4.  go func() { 
    5.   
  
  
  
  5.   select { 
    6.   
  
  
  
  6.   case data, ok := <-chanInt: 
    7.   
  
  
  
  7.    if ok { 
    8.   
  
  
  
  8.     fmt.Println(data) 
    9.   
  
  
  
  9.    } 
    10.   
  
  
  
  10.   default: 
    11.   
  
  
  
  11.    fmt.Println("全部阻塞") 
    12.   
  
  
  
  12.   } 
    13.   
  
  
  
  13.  }() 
    14.   
  
  
  
  14.  chanInt <- 1 
    15.   
  
  
  
  15. } 
    16.

输出1

  • 这是一个简单的接收发送模型
  • 如果 select 的多个分支都满足条件，则会随机的选取其中一个满足条件的分支。
  • 第 6 行加上 ok 是因为上一节讲过，如果不加会导致通道关闭时收到零值
  • 回忆之前的知识，让接收和发送在不同的goroutine里，否则会死锁

这个程序存在什么问题?

假如发送太慢，所有case都处于阻塞状态，会直接执行default的内容。这里加一行sleep试试。

 
 
 
 
    
  
  
  
  
  1. func bar() { 
    2.   
  
  
  
  2.  chanInt := make(chan int) 
    3.   
  
  
  
  3.  defer close(chanInt) 
    4.   
  
  
  
  4.  go func() { 
    5.   
  
  
  
  5.   .... 
    6.   
  
  
  
  6.  }() 
    7.   
  
  
  
  7.  time.Sleep(time.Second) 
    8.   
  
  
  
  8.  chanInt <- 1 
    9.   
  
  
  
  9. } 
    10.
  • 倒数第二行加了sleep 1 秒，导致select语句提前结束
  • 猜测一下会输出全部阻塞吗?
 
 
 
 
    
  
  
  
  
  1. 全部阻塞 
    2.   
  
  
  
  2. fatal error: all goroutines are asleep - deadlock! 
    3.   
  
  
  
  3.  
    4.   
  
  
  
  4. goroutine 1 [chan send]: 
    5.   
  
  
  
  5. main.bar() 
    6.

是会输出全部阻塞的。

因为接收执行完了，退出了goroutine，而发送才刚刚执行到，没有与其匹配的接收，故死锁。

正确的做法是把接收套在循环里面。

 
 
 
 
    
  
  
  
  
  1. func baz() { 
    2.   
  
  
  
  2.  chanInt := make(chan int) 
    3.   
  
  
  
  3.  defer close(chanInt) 
    4.   
  
  
  
  4.  go func() { 
    5.   
  
  
  
  5.   for { 
    6.   
  
  
  
  6.    select { 
    7.   
  
  
  
  7.        ... 
    8.   
  
  
  
  8.    } 
    9.   
  
  
  
  9.   } 
    10.   
  
  
  
  10.  }() 
    11.   
  
  
  
  11.  chanInt <- 1 
    12.   
  
  
  
  12. } 
    13.
  • 不再死锁了
  • 假如程序不停止，会出现一个泄露的goroutine，永远的在for循环中无法跳出，此时引入下一节的内容

通知机制

Go 语言总是简单和灵活的，虽然没有针对提供专门的机制来处理退出，但我们可以自己组合

 
 
 
 
    
  
  
  
  
  1. func main() { 
    2.   
  
  
  
  2.  chanInt, done := make(chan int), make(chan struct{}) 
    3.   
  
  
  
  3.  defer close(chanInt) 
    4.   
  
  
  
  4.  defer close(done) 
    5.   
  
  
  
  5.  go func() { 
    6.   
  
  
  
  6.   for { 
    7.   
  
  
  
  7.    select { 
    8.   
  
  
  
  8.    case <-chanInt: 
    9.   
  
  
  
  9.    case <-done: 
    10.   
  
  
  
  10.     break 
    11.   
  
  
  
  11.    } 
    12.   
  
  
  
  12.   } 
    13.   
  
  
  
  13.  }() 
    14.   
  
  
  
  14.  done <- struct{}{} 
    15.   
  
  
  
  15. } 
    16.

没有给chanInt发送任何东西，按理说会阻塞，导致goroutine泄露

但可以使用额外的通道完成协程的退出控制

这种方式还可以做到周期性处理任务，下一节我们再详细讲解

case 的并发性

case是有并发属性的，比如两次输入，分别等待 1、2 秒，再进行两次读取，会花 3 秒时间吗?

 
 
 
 
    
  
  
  
  
  1. func main() { 
    2.   
  
  
  
  2.  c1,c2 := make(chan string), make(chan string) 
    3.   
  
  
  
  3.     close(c1) 
    4.   
  
  
  
  4.     close(c2) 
    5.   
  
  
  
  5.  go func() { 
    6.   
  
  
  
  6.   time.Sleep(time.Second * 1) 
    7.   
  
  
  
  7.   c1 <- "one" 
    8.   
  
  
  
  8.  }() 
    9.   
  
  
  
  9.  go func() { 
    10.   
  
  
  
  10.   time.Sleep(time.Second * 2) 
    11.   
  
  
  
  11.   c2 <- "two" 
    12.   
  
  
  
  12.  }() 
    13.   
  
  
  
  13.  start := time.Now() // 获取当前时间 
    14.   
  
  
  
  14.  for i := 0; i < 2; i++ { 
    15.   
  
  
  
  15.   select { 
    16.   
  
  
  
  16.   case <-c1: 
    17.   
  
  
  
  17.   case <-c2: 
    18.   
  
  
  
  18.   } 
    19.   
  
  
  
  19.  } 
    20.   
  
  
  
  20.  elapsed := time.Since(start) 
    21.   
  
  
  
  21.  // 这里没有用到3秒，为什么？ 
    22.   
  
  
  
  22.  fmt.Println("该函数执行完成耗时：", elapsed) 
    23.   
  
  
  
  23. } 
    24.

以上代码先初始化两个 channel c1 和 c2，然后开启两个 goroutine 分别往 c1 和 c2 写入数据，再通过 select 监听两个 channel，从中读取数据并输出。

运行结果如下：

 
 
 
 
    
  
  
  
  
  1. $ go run channel.go 
    2.   
  
  
  
  2. received one 
    3.   
  
  
  
  3. received two 
    4.   
  
  
  
  4. 该函数执行完成耗时：2.004695535s 
    5.

这充分说明case是并发的，但要注意此处的并发是 case 对channel阻塞做出的特殊处理。

case并发的原理

假如case后左边和右边跟了函数，会执行函数，我们来探索一下。

定义A、B函数，作用相同

 
 
 
 
    
  
  
  
  
  1. func A() int { 
    2.   
  
  
  
  2.  fmt.Println("start A") 
    3.   
  
  
  
  3.  time.Sleep(1 * time.Second) 
    4.   
  
  
  
  4.  fmt.Println("end A") 
    5.   
  
  
  
  5.  return 1 
    6.   
  
  
  
  6. } 
    7.

定义函数lee，请问该函数执行完成耗时多少呢?

 
 
 
 
    
  
  
  
  
  1. func lee() { 
    2.   
  
  
  
  2.  ch, done := make(chan int), make(chan struct{}) 
    3.   
  
  
  
  3.  defer close(ch) 
    4.   
  
  
  
  4.  go func() { 
    5.   
  
  
  
  5.   select { 
    6.   
  
  
  
  6.   case ch <- A(): 
    7.   
  
  
  
  7.   case ch <- B(): 
    8.   
  
  
  
  8.   case <-done: 
    9.   
  
  
  
  9.   } 
    10.   
  
  
  
  10.  }() 
    11.   
  
  
  
  11.  done <- struct{}{} 
    12.   
  
  
  
  12. } 
    13.

答案是 2 秒

 
 
 
 
    
  
  
  
  
  1. start A 
    2.   
  
  
  
  2. end A 
    3.   
  
  
  
  3. start B 
    4.   
  
  
  
  4. end B 
    5.   
  
  
  
  5. main.leespend time: 2.003504395s 
    6.
  • select 扫描是从左到右从上到下的，按这个顺序先求值，如果是函数会先执行函数。
  • 然后立马判断是否可以立即执行(这里是指 case 是否会因为执行而阻塞)。
  • 所以两个函数都会进入，而且是先进入 A 再进入 B，两个函数都会执行完，所以等待时间会累计。

如果都不会阻塞，此时就会使用一个伪随机的算法，去选中一个 case，只要选中了其他就被放弃了。

超时控制

我们来模拟一个更真实点的例子，让程序一段时间超时退出。

定义一个结构体

 
 
 
 
    
  
  
  
  
  1. type Worker struct { 
    2.   
  
  
  
  2.  stream  <-chan int //处理 
    3.   
  
  
  
  3.  timeout time.Duration //超时 
    4.   
  
  
  
  4.  done    chan struct{} //结束信号 
    5.   
  
  
  
  5. } 
    6.

定义初始化函数

 
 
 
 
    
  
  
  
  
  1. func NewWorker(stream <-chan int, timeout int) *Worker { 
    2.   
  
  
  
  2.  return &Worker{ 
    3.   
  
  
  
  3.   stream:  stream, 
    4.   
  
  
  
  4.   timeout: time.Duration(timeout) * time.Second, 
    5.   
  
  
  
  5.   done:    make(chan struct{}), 
    6.   
  
  
  
  6.  } 
    7.   
  
  
  
  7. } 
    8.

定义超时处理函数

 
 
 
 
    
  
  
  
  
  1. func (w *Worker) afterTimeStop() { 
    2.   
  
  
  
  2.  go func() { 
    3.   
  
  
  
  3.   time.Sleep(w.timeout) 
    4.   
  
  
  
  4.   w.done <- struct{}{} 
    5.   
  
  
  
  5.  }() 
    6.   
  
  
  
  6. } 
    7.
  • 超过时间发送结束信号

接收数据并处理函数

 
 
 
 
    
  
  
  
  
  1. func (w *Worker) Start() { 
    2.   
  
  
  
  2.  w.afterTimeStop() 
    3.   
  
  
  
  3.  for { 
    4.   
  
  
  
  4.   select { 
    5.   
  
  
  
  5.   case data, ok := <-w.stream: 
    6.   
  
  
  
  6.    if !ok { 
    7.   
  
  
  
  7.     return 
    8.   
  
  
  
  8.    } 
    9.   
  
  
  
  9.    fmt.Println(data) 
    10.   
  
  
  
  10.   case <-w.done: 
    11.   
  
  
  
  11.    close(w.done) 
    12.   
  
  
  
  12.    return 
    13.   
  
  
  
  13.   } 
    14.   
  
  
  
  14.  } 
    15.   
  
  
  
  15. } 
    16.
  • 收到结束信号关闭函数
  • 这样的方法就可以让程序在等待 1 秒后继续执行，而不会因为 ch 读取等待而导致程序停滞。
 
 
 
 
    
  
  
  
  
  1. func main() { 
    2.   
  
  
  
  2.  stream := make(chan int) 
    3.   
  
  
  
  3.  defer close(stream) 
    4.   
  
  
  
  4.  
    5.   
  
  
  
  5.  w := NewWorker(stream, 3) 
    6.   
  
  
  
  6.  w.Start() 
    7.   
  
  
  
  7. } 
    8.

实际 3 秒到程序运行结束。

这种方式巧妙地实现了超时处理机制，这种方法不仅简单，在实际项目开发中也是非常实用的。

小结

本节介绍了select的用法以及包含的陷阱，我们学会了

  • case是并发的
  • case只针对通道传输阻塞做特殊处理，如果有计算将会先进行计算
  • 扫描是从左到右从上到下的，按这个顺序先求值，如果是函数会先执行函数。如果函数运行时间长，时间会累计
  • 在case全部阻塞时，会执行default中的内容
  • 可使用结束信号，让select退出
  • 延时发送结束信号可以实现超时自动退出的功能

问题：为什么w.stream没有程序向他发送数据，却没有死锁呢?

本节源码位置 https://github.com/golang-minibear2333/golang/blob/master/4.concurrent/4.5-select”

本文转载自微信公众号「机智的程序员小熊」，可以通过以下二维码关注。转载本文请联系机智的程序员小熊公众号。

分享标题：讲透Go中的并发接收控制结构Select
网页网址：http://www.mswzjz.com/qtweb/news40/193090.html

网站建设、网络推广公司-创新互联，是专注品牌与效果的网站制作，网络营销seo公司；服务项目有等

广告

声明：本网站发布的内容（图片、视频和文字）以用户投稿、用户转载内容为主，如果涉及侵权请尽快告知，我们将会在第一时间删除。文章观点不代表本网站立场，如需处理请联系客服。电话：028-86922220；邮箱：631063699@qq.com。内容未经允许不得转载，或转载时需注明来源： 创新互联

猜你还喜欢下面的内容

微信公众号知识

各行业网站

成都网页设计　　 　香港空间　　 　宣威网站建设　　 　成发金汇　　 　雅安服务器托管　　 　渠县网站建设　　 　网站建设公司　　 　济桢教育　　 　手机软件开发　　 　成都西云机房　　 　手机网站制作　　 　网站空间　　 　成都恒翔机电维修　　 　成都租用服务器　　 　成都做网站　　 　成都网站建设　　 　德阳东方电机技改　　 　成都木托盘　　 　抖音短视频拍摄　　 　光华机房服务器托管