Channel

声明Channel

直接声明

  1. var c chan int
  2. fmt.Println(c) // nil
  • 声明类型为 chan int 的变量c, 通道的数据类型是int
  • chan int 的 zero-value 是 nil

通过make关键字声明

  1. c := make(chan int)
  3. fmt.Printf("Type of c is %T\n", c)  // chan int 
  4. fmt.Printf("Value of c is %v\n", c) // 0xc000070060
  • channel 值实际是指针 (指针类型的元素值总计有切片、channel和map)

向通道写读数据

写数据

  1. c <- data

读数据

从空channel中读数据会出现异常

  1. <- c
  2. 或
  3. val, ok := <- c   // 若为false，则说明channel关闭了或者channel为空

使用channel阻塞进程

  • 当向channel中写数据时，程序会阻塞直至该channel中的数据被消费
  • 当从channel中读数据时，若channel为空程序就会阻塞
  1. package main
  2. import "fmt"
  4. func greet(c chan string) {
  5.     fmt.Printf("Hello " + <- c + " !")
  6. }
  8. func main() {
  9.     fmt.Println("Main go start")
  10.     c := make(chan string)
  11.     go greet(c)
  12.     c <- "YY"
  13.     fmt.Println("Main go end")
  14. }

死锁

fatal error: all goroutines are asleep — deadlock!.

  • 向channel中写入数据，并没有其他协程读取该channel的数据
  • 从channel中读取数据，若channel中并没有数据就会一直死锁

For Loop

  1. package main
  2. import "fmt"
  4. func squares(c chan int) {
  5.     for i := 0; i<= 9; i++ {
  6.         c <- i * i
  7.     }
  8.     close(c)
  9. }
  11. func main() {
  12.     fmt.Println("Main start")
  13.     c := make(chan int)
  15.     go squares(c)
  16.     for val := range c {
  17.         fmt.Println(val)
  18.     }
  19.     fmt.Println("Main end")
  20. }

Channel的缓冲长度和容量

  1. c := make(chen Type, n)
  • len() 函数用来计算channel的缓冲长度
  • cap()函数用来计算channel的容量

  • for range

    用来读取channel中数据，即便channel已经关闭只要channel变量中存在数据，都可以读取数据

  1. 当向channel中写入的数据个数大于channel的容量时，程序就会阻塞
  2. 当程序阻塞的同时系统会调用其他的协程来消费channel
  3. 直至channel中的数据被消费完后，程序才会将执行权移交给刚刚阻塞的进程

单向Channel

只写Channel

  1. roc := make(<-chan int)

只读Channel

  1. soc := make(chan<- int)
  1. package main
  2.  import "fmt"
  4.  func greet(roc <-chan string) {
  5.     fmt.Printf("Hello %v\n", <- roc)
  6.  }
  8.  func main() {
  9.         fmt.Println("Main start ...")
  10.          c := make(chan string)
  11.         go greet(c)
  12.         c <- "YY"
  13.          fmt.Println("Main end ...")
  14.  }

匿名Goroutine

  1. package main
  2. import "fmt"
  4. func main() {
  5.     fmt.Println("Main start ...")
  6.     c := make(chan string)
  7.     go func(c chan string) {
  8.         fmt.Printf("The value is %v\n", <- c)
  9.     }(c)
  10.     c <- "YY"
  11.     fmt.Println("Main end ...")
  12. }

Channel作为Channel数据类型

  1. package main
  2. import "fmt"
  4. func greet(c chan string){
  5.     fmt.Printf("Hello %v !\n", <-c)
  6. }
  8. func greeter(cc chan chan string) {
  9.     c := make(chan string)
  10.     cc <- c
  11. }
  13. func main(){
  14.     fmt.Println("Main start ...")
  15.     cc := make(chan chan string)
  16.     go greeter(cc)
  18.     c := <-cc
  19.     go greet(c)
  21.     c <- "YY"
  22.     fmt.Println("Main end ...")
  23. }

Select 关键字

Select VS Switch

  • select 就像 switch 一样根据不同的条件执行不同的分支
  • select 不需要任何输入参数
  • 如果没有default case，select不会阻塞进程
  • 一旦有一个case分支符合条件，程序就会执行
  • 如果所有的分支都符合条件，程序会随机选择一个分支执行
  1. package main
  2. import (
  3.     "fmt"
  4.     "time"
  5. )
  6. var start time.Time 
  7. func init() {
  8.     start = time.Now()
  9. }
  11. func Service1(c chan string) {
  12.     c <- "Hello Service1"              
  13. }
  15. func Service2(c chan string) {
  16.     c <- "Hello Service2"              
  17. }
  19. func main() {
  20.     fmt.Println("Main start at ", time.Since(start))
  21.     chan1 := make(chan string)
  22.     chan2 := make(chan string)
  24.     go Service1(chan1)
  25.     go Service2(chan2)
  27.     select {
  28.     case res := <- chan1:
  29.         fmt.Println("Response from service1 ", res, time.Since(start))
  30.     case res := <- chan2:
  31.         fmt.Println("Response from service2 ", res, time.Since(start))
  32.     }
  34.     fmt.Println("Main end at ", time.Since(start))
  35. }

添加default timeout

  1. ...
  3. select {
  4.     case res := <- chan1:
  5.         fmt.Println()
  6.     case res := <- chan2:
  7.         fmt.Println()
  8.     case <- time.After(2 * time.Second)
  9.         fmt.Println(()
  10. }

空Select

空Select将会导致进程的阻塞

WaitGroup

sync.WaitGroup

当WaitGroup被创建后它的计数（Counter）默认值是0

  • Add 该方法需要一个int类型的参数，通常是1
  • Wait 当计数Counter不为0时，调用该方法时程序就会阻塞
  • Done 每当调用该方法计数Counter就会自动减1
  1. package main
  2. import (
  3.     "fmt"
  4.     "sync"
  5.     "time"
  6. )
  8. func Service(wg *sync.WaitGroup, instance int) {
  9.     time.Sleep(2 * time.Second)
  10.     fmt.Println("Service on instance ", instance)
  11.     wg.Done()
  12. }
  14. func main() {
  15.     fmt.Println("Main start ...")
  16.     var wg sync.WaitGroup
  18.     for i := 1; i <= 3; i ++{
  19.         wg.Add(1)
  20.         go Service(&wg, i)
  21.     }
  23.     wg.Wait()
  24.     fmt.Println("Main End ...")
  25. }

Worker pool

  1. package main
  2. import (
  3.     "fmt"
  4.     "time"
  5. )
  7. func sqrWorker(tasks <-chan int, results chan<- int, instance int) {
  8.     for num := range tasks {
  9.         time.Sleep(time.Millisecond)
  10.         fmt.Printf("[worker %v] Sending result by worker %v\n", instance, instance)
  11.         results <- num * num
  12.     }
  13. }
  15. func main() {
  16.     fmt.Println("[main] main() start")
  18.     tasks := make(chan int, 10)
  19.     results := make(chan int, 10)
  21.     for i := 0; i < 3; i++ {
  22.         go sqrWorker(tasks, results, i)
  23.     }
  25.     for i := 0; i < 5; i++ {
  26.         tasks <- i * 2
  27.     }
  29.     fmt.Printf("[main] wrote 5 tasks")
  30.     close(tasks)
  32.     for i := 0; i < 5; i ++ {
  33.         result := <-results
  34.         fmt.Println("[main] result ", i, ":", result)
  35.     }
  37.     fmt.Println("[main] main stop")
  38. }

使用WaitGroup实现上述功能

  • 避免一些不必要的上下文的切换
  • 但是所有的执行必须要等到wg的计数器减少至0
  1. package main
  2. import (
  3.     "fmt"
  4.     "time"
  5.     "sync"
  6. )
  8. func sqrWorker(wg *sync.WaitGroup, tasks <-chan int, results chan<- int, instance int) {
  9.     for num := range tasks {
  10.         time.Sleep(time.Millisecond)
  11.         fmt.Printf("[worker %v] Sending result by worker %v\n", instance, instance)
  12.         results <- num * num
  13.     }
  14.     wg.Done()
  15. }
  17. func main() {
  18.     fmt.Println("[main] main() start")
  20.     tasks := make(chan int, 10)
  21.     results := make(chan int, 10)
  22.     var wg sync.WaitGroup
  24.     for i := 0; i < 3; i++ {
  25.         wg.Add(1)
  26.         go sqrWorker(&wg, tasks, results, i)
  27.     }
  29.     for i := 0; i < 5; i++ {
  30.         tasks <- i * 2
  31.     }
  33.     fmt.Printf("[main] wrote 5 tasks")
  34.     close(tasks)
  36.     wg.Wait()
  37.     for i := 0; i < 5; i ++ {
  38.         result := <-results
  39.         fmt.Println("[main] result ", i, ":", result)
  40.     }
  42.     fmt.Println("[main] main stop")
  43. }

Mutex 竞态

如果避免不同的goroutines互相竞争资源
**

  1. package main
  2. import (
  3.     "fmt"
  4.     "sync"
  5. )
  7. var i int
  9. func Worker(wg *sync.WaitGroup, m *sync.Mutex) {
  10.     m.Lock()
  11.     i = i + 1
  12.     m.Unlock()
  13.     wg.Done()
  14. }
  16. func main() {
  17.     fmt.Println("[main] main start")
  18.     var wg sync.WaitGroup
  19.     var m sync.Mutex
  21.     for i := 1; i<= 1000; i ++ {
  22.         wg.Add(1)
  23.         go Worker(&wg, &m)
  24.     }
  25.     wg.Wait()
  26.     fmt.Printf("The i is %v", i)
  27.     fmt.Println("[main] main end")
  28. }