Golang channel 使用总结

olzhy · 2019-07-02 08:17:47 · 910 次点击 · 预计阅读时间 29 分钟 · 大约8小时之前 开始浏览    
这是一个创建于 2019-07-02 08:17:47 的文章,其中的信息可能已经有所发展或是发生改变。

以常规方式编写并发程序,需要对共享变量作正确的访问控制,处理起来很困难。而golang提出一种不同的方式,即共享变量通过channel传递,共享变量从不被各个独立运行的线程(goroutine)同时享有,在任一时刻,共享变量仅可被一个goroutine访问。所以,不会产生数据竞争。并发编程,golang鼓励以此种方式进行思考,精简为一句口号——“勿通过共享内存来进行通信,而应通过通信来进行内存共享”。
1 Unbuffered channels与Buffered channels
Unbuffered channels的接收者阻塞直至收到消息,发送者阻塞直至接收者接收到消息,该机制可用于两个goroutine的状态同步。Buffered channels在缓冲区未满时,发送者仅在值拷贝到缓冲区之前是阻塞的,而在缓冲区已满时,发送者会阻塞,直至接收者取走了消息,缓冲区有了空余。
1.1 Unbuffered channels
如下代码使用Unbuffered channel作同步控制。给定一个整型数组,在主routine启动另一个goroutine将该数组排序,当其完成时,给done channel发送完成消息,主routine会一直等待直至排序完成,打印结果。

  1. package main  
  2.   
  3. import (  
  4.     "fmt"  
  5.     "sort"  
  6.     "time"  
  7. )  
  8.   
  9. func main() {  
  10.     done := make(chan bool)  
  11.     nums := []int{21354}  
  12.     go func() {  
  13.         time.Sleep(time.Second)  
  14.         sort.Ints(nums)  
  15.         done <- true  
  16.     }()  
  17.     <-done  
  18.     fmt.Println(nums)  
  19. }  

1.2 Buffered channels
如下代码中,messages chan的缓冲区大小为2,因其为Buffered channel,所以消息发送与接收无须分开到两个并发的goroutine中。
  1. package main  
  2.   
  3. import (  
  4.     "fmt"  
  5. )  
  6.   
  7. func main() {  
  8.     messages := make(chan string, 2)  
  9.     messages <- "hello"  
  10.     messages <- "world"  
  11.     fmt.Println(<-messages, <-messages)  
  12. }  

2 配套使用
2.1 指明channel direction
函数封装时,对仅作消息接收或仅作消息发送的chan标识direction可以借用编译器检查增强类型使用安全。如下代码中,ping函数中pings chan仅用来接收消息,所以参数列表中将其标识为接收者。pong函数中,pings chan仅用来发送消息,pongs chan仅用来接收消息,所以参数列表中二者分别标识为发送者与接收者。
  1. package main  
  2.   
  3. import "fmt"  
  4.   
  5. func ping(pings chan<- string, msg string) {  
  6.     pings <- msg  
  7. }  
  8.   
  9. func pong(pings <-chan string, pongs chan<- string) {  
  10.     pongs <- <-pings  
  11. }  
  12.   
  13. func main() {  
  14.     pings, pongs := make(chan string, 1), make(chan string, 1)  
  15.     ping(pings, "ping")  
  16.     pong(pings, pongs)  
  17.     fmt.Println(<-pongs)  
  18. }  

2.2 select
使用select可以用来等待多个channel的消息,如下代码,创建两个chan,启动两个goroutine耗费不等时间计算结果,主routine监听消息,使用两次select,第一次接收到了ch2的消息,第二次接收到了ch1的消息,用时2.000521146s。
  1. package main  
  2.   
  3. import (  
  4.     "fmt"  
  5.     "time"  
  6. )  
  7.   
  8. func main() {  
  9.     c1, c2 := make(chan int1), make(chan int1)  
  10.     go func() {  
  11.         time.Sleep(2  time.Second)  
  12.         c1 <- 1  
  13.     }()  
  14.     go func() {  
  15.         time.Sleep(time.Second)  
  16.         c2 <- 2  
  17.     }()  
  18.     for i := 0; i < 2; i++ {  
  19.         select {  
  20.         case msg1 := <-c1:  
  21.             fmt.Println("received msg from c1", msg1)  
  22.         case msg2 := <-c2:  
  23.             fmt.Println("received msg from c2", msg2)  
  24.         }  
  25.     }  
  26. }  

2.3 select with default
select with default可以用来处理非阻塞式消息发送、接收及多路选择。如下代码中,第一个select为非阻塞式消息接收,若收到消息,则落入<-messages case,否则落入default。第二个select为非阻塞式消息发送,与非阻塞式消息接收类似,因messages chan为Unbuffered channel且无异步消息接收者,因此落入default case。第三个select为多路非阻塞式消息接收。
  1. package main  
  2.   
  3. import "fmt"  
  4.   
  5. func main() {  
  6.     messages := make(chan string)  
  7.     signal := make(chan bool)  
  8.   
  9.     // receive with default  
  10.     select {  
  11.     case <-messages:  
  12.         fmt.Println("message received")  
  13.     default:  
  14.         fmt.Println("no message received")  
  15.     }  
  16.   
  17.     // send with default  
  18.     select {  
  19.     case messages <- "message":  
  20.         fmt.Println("message sent successfully")  
  21.     default:  
  22.         fmt.Println("message sent failed")  
  23.     }  
  24.   
  25.     // muti-way select  
  26.     select {  
  27.     case <-messages:  
  28.         fmt.Println("message received")  
  29.     case <-signal:  
  30.         fmt.Println("signal received")  
  31.     default:  
  32.         fmt.Println("no message or signal received")  
  33.     }  
  34. }  

2.4 close
当无需再给channel发送消息时,可将其close。如下代码中,创建一个Buffered channel,首先启动一个异步goroutine循环消费消息,然后主routine完成消息发送后关闭chan,消费goroutine检测到chan关闭后,退出循环。
  1. package main  
  2.   
  3. import "fmt"  
  4.   
  5. func main() {  
  6.     messages := make(chan int10)  
  7.     done := make(chan bool)  
  8.   
  9.     // consumer  
  10.     go func() {  
  11.         for {  
  12.             msg, more := <-messages  
  13.             if !more {  
  14.                 fmt.Println("no more message")  
  15.                 done <- true  
  16.                 break  
  17.             }  
  18.             fmt.Println("message received", msg)  
  19.         }  
  20.     }()  
  21.   
  22.     // producer  
  23.     for i := 0; i < 5; i++ {  
  24.         messages <- i  
  25.     }  
  26.     close(messages)  
  27.     <-done  
  28. }  

2.5 for range
for range语法不仅可对基础数据结构(slice、map等)作迭代,还可对channel作消息接收迭代。如下代码中,给messages chan发送两条消息后将其关闭,然后迭代messages chan打印消息。
  1. package main  
  2.   
  3. import "fmt"  
  4.   
  5. func main() {  
  6.     messages := make(chan string, 2)  
  7.     messages <- "hello"  
  8.     messages <- "world"  
  9.     close(messages)  
  10.   
  11.     for msg := range messages {  
  12.         fmt.Println(msg)  
  13.     }  
  14. }  

3 应用场景
3.1 超时控制
资源访问、网络请求等场景作超时控制是非常必要的,可以使用channel结合select来实现。如下代码,对常规sum函数增加超时限制,sumWithTimeout函数中,select的v := <-rlt在等待计算结果,若在时限范围内计算完成,则正常返回计算结果,若超过时限则落入<-time.After(timeout) case,抛出timeout error。
  1. package main  
  2.   
  3. import (  
  4.     "errors"  
  5.     "fmt"  
  6.     "time"  
  7. )  
  8.   
  9. func sum(nums []intint {  
  10.     rlt := 0  
  11.     for _, num := range nums {  
  12.         rlt += num  
  13.     }  
  14.     return rlt  
  15. }  
  16.   
  17. func sumWithTimeout(nums []int, timeout time.Duration) (int, error) {  
  18.     rlt := make(chan int)  
  19.     go func() {  
  20.         time.Sleep(2  time.Second)  
  21.         rlt <- sum(nums)  
  22.     }()  
  23.     select {  
  24.     case v := <-rlt:  
  25.         return v, nil  
  26.     case <-time.After(timeout):  
  27.         return 0, errors.New("timeout")  
  28.     }  
  29. }  
  30.   
  31. func main() {  
  32.     nums := []int{12345}  
  33.     timeout := 3 * time.Second // time.Second  
  34.     rlt, err := sumWithTimeout(nums, timeout)  
  35.     if nil != err {  
  36.         fmt.Println("error", err)  
  37.         return  
  38.     }  
  39.     fmt.Println(rlt)  
  40. }  


本文代码托管地址:https://github.com/olzhy/go-excercises/tree/master/channels

原文地址:https://leileiluoluo.com/posts/golang-channels.html


有疑问加站长微信联系(非本文作者))

入群交流(和以上内容无关):加入Go大咖交流群,或添加微信:liuxiaoyan-s 备注:入群;或加QQ群:692541889

910 次点击  
加入收藏 微博
暂无回复
添加一条新回复 (您需要 登录 后才能回复 没有账号 ?)
  • 请尽量让自己的回复能够对别人有帮助
  • 支持 Markdown 格式, **粗体**、~~删除线~~、`单行代码`
  • 支持 @ 本站用户;支持表情(输入 : 提示),见 Emoji cheat sheet
  • 图片支持拖拽、截图粘贴等方式上传