go并发编程

①、协程Goroutine

:::success Goroutine：每个协程的堆栈，go语言默认一开始分配2k但是很智能，可以自适应增长；Go语言中的goroutine就是这样一种机制，goroutine的概念类似于线程，但 goroutine是由Go的运行时（runtime）调度和管理的。Go程序会智能地将 goroutine 中的任务合理地分配给每个CPU。

语法：**<font style="color:#DF2A3F;">go</font>**** func**

子协程异常退出：子协程的异常退出会将异常传播到主协程，直接会导致主协程也跟着挂掉，然后整个程序就崩溃了；使用panic和recover解决。

:::

:::color2 主协程（**main**）退出了，其他该协程的子协程任务全部结束执行；

父协程（**非main**）退出了：只要main没结束，其他该协程的子协程任务继续执行；

:::

  1. package main
  3. import (
  4.     "fmt"
  5.     "time"
  6. )
  8. func main() {
  9.     // 合起来写
  10.     go func() {
  11.         i := 0
  12.         for {
  13.             i++
  14.             fmt.Printf("new goroutine: i = %d\n", i)
  15.             time.Sleep(time.Second)
  16.         }
  17.     }()
  18.     i := 0
  19.     for {
  20.         i++
  21.         fmt.Printf("main goroutine: i = %d\n", i)
  22.         time.Sleep(time.Second)
  23.         if i == 2 {
  24.             break
  25.         }
  26.     }
  27. }
  29. /*    输出打印
  30. main goroutine: i = 1
  31. new goroutine: i = 1
  32. new goroutine: i = 2
  33. main goroutine: i = 2
  34. */
  1. package main
  3. import (
  4.     "fmt"
  5.     "time"
  6. )
  8. func main() {
  9.     // 合起来写
  10.     go func() {
  11.         i := 0
  12.         go func() {
  13.             i := 0
  14.             for {
  15.                 i++
  16.                 fmt.Printf("zi zi  goroutine: i = %d\n", i)
  17.                 time.Sleep(time.Second)
  18.             }
  20.         }()
  21.         for {
  22.             i++
  23.             fmt.Printf("new goroutine: i = %d\n", i)
  24.             time.Sleep(time.Second)
  25.             if i == 2 {
  26.                 break
  27.             }
  28.         }
  30.     }()
  31.     i := 0
  32.     for {
  33.         i++
  34.         fmt.Printf("main goroutine: i = %d\n", i)
  35.         time.Sleep(time.Second)
  36.         if i == 10 {
  37.             break
  38.         }
  39.     }
  40. }
  45. /*  输出
  46. main goroutine: i = 1
  47. new goroutine: i = 1
  48. zi zi  goroutine: i = 1
  49. new goroutine: i = 2
  50. zi zi  goroutine: i = 2
  51. main goroutine: i = 2  
  52. zi zi  goroutine: i = 3
  53. main goroutine: i = 3
  54. zi zi  goroutine: i = 4
  55. main goroutine: i = 4
  56. zi zi  goroutine: i = 5
  57. main goroutine: i = 5
  58. zi zi  goroutine: i = 6
  59. main goroutine: i = 6
  60. zi zi  goroutine: i = 7
  61. main goroutine: i = 7
  62. zi zi  goroutine: i = 8
  63. main goroutine: i = 8
  64. zi zi  goroutine: i = 9
  65. main goroutine: i = 9
  66. zi zi  goroutine: i = 10
  67. main goroutine: i = 10
  68. zi zi  goroutine: i = 11
  69. */

:::color2 goroutine与OS线程对比：
可增长的栈：OS线程（操作系统线程）一般都有固定的栈内存（通常为2MB）,一个goroutine的栈在其生命周期开始时只有很小的栈（典型情况下2KB），goroutine的栈不是固定的，他可以按需增大和缩小，goroutine的栈大小限制可以达到1GB，虽然极少会用到这个大。所以在Go语言中一次创建十万左右的goroutine也是可以的。
goroutine调度：GPM是Go语言运行时（runtime）层面的实现，是go语言自己实现的一套调度系统（调度是在用户态下完成的）。不涉及内核态与用户态之间的频繁切换，包括内存的分配与释放，都是在用户态维护着一块大的内存池， 不直接调用系统的malloc函数（除非内存池需要改变），成本比调度OS线程低很多。 另一方面充分利用了多核的硬件资源，近似的把若干goroutine均分在物理线程上， 再加上本身goroutine的超轻量，以上种种保证了go调度方面的性能。

:::

②、通道channel

:::warning 协程的同步问题解决方案：1、channel 2、 锁

go语言中，控制协程间的并发主要是用channel；如果说goroutine是Go语言程序的并发体的话，那么channel是它们之间的通信机制（锁用的很少，协程间channel是并发安全的**）**。

channel：

  1. 作为协程的输出，通道是一个容器，它可以容纳数据。
  2. 作为协程的输入，通道是一个生产者，它可以向协程提供数据。
  3. 通道作为容器是有限定大小的，满了就写不进去，空了就读不出来。
  4. 通道有它自己的类型，它可以限定进入通道的数据的类型。

go并发编程（2） - 图1

:::

:::color2 channel是一种类型，一种引用类型，声明通道类型：var 变量 chan 元素类型如：

var ch1 chan int // 声明一个传递整型的双向通道 ch1 --> <nil>

var ch2 chan bool // 声明一个传递布尔型的双向通道 ch2 --> <nil>

var ch3 chan []int // 声明一个传递int切片的双向通道 ch3 --> <nil>

声明的通道后需要**使用make函数初始化之后才能使用**：创建channel的格式：

make(chan 元素类型, [缓冲大小])如：

ch4 := make(chan int)

ch5 := make(chan bool)

ch6 := make(chan []int)

「缓冲型通道」：上述通道就是缓冲型通道；

「非缓冲型通道」：省略[缓冲大小]为非缓冲型通道，只能写入一个，等对方读取了才能再写，对方没读，你还写你就会阻塞。

比较：两个相同类型的channel可以使用==运算符比较。如果两个channel引用的是相通的对象，那么比较的结果为真。一个channel也可以和nil进行比较

:::

:::color2 channel操作：

发送（send）：ch := make(chan int) //创建一个通道；ch <- 10 // 把10发送到ch中

接收(receive）：x := <- ch // 从ch中接收值并赋值给变量x、<-ch // 从ch中接收值，**<font style="color:#DF2A3F;">忽略结果</font>**、data, ok := <-ch // 从ch中接收值，并通过ok判断是否通道关闭

关闭（close）：**<font style="color:#DF2A3F;">内置函数</font>**close(ch)：关于关闭通道需要注意的事情是，只有在通知接收方goroutine所有的数据都发送完毕的时候才需要关闭通道。通道是可以被垃圾回收机制回收的，它和关闭文件是不一样的，在结束操作之后关闭文件是必须要做的，但关闭通道不是必须的。

关闭后的通道有以下特点：

  1. 对一个关闭的通道再发送值就会导致panic。
  2. 对一个关闭的通道进行接收会一直获取值直到通道为空。
  3. 对一个关闭的并且没有值的通道执行接收操作会得到对应类型的零值。
  4. 关闭一个已经关闭的通道会导致panic。

:::

:::success 无缓冲的通道：无缓冲的通道又称为阻塞的通道，无缓冲通道也被称为同步通道；

go并发编程（2） - 图2

无缓冲的通道主线程不能阻塞，deadlock只检查主线程，主线程发送时，必须有人接收值的时候才能发送值，主线程接收时，必须要有人发送值才能接收。

使用无缓冲通道进行通信将导致发送和接收的goroutine同步化。因此，无缓冲通道也被称为同步通道。 ::: go // ① 主发送，无接收 func main() { ch := make(chan int) ch <- 10 fmt.Println("发送成功") } /* 会报deadlock错，因为我们使用ch := make(chan int)创建的是无缓冲的通道，无缓冲的通道只有在有人接收值的时候才能发送值 fatal error: all goroutines are asleep - deadlock! goroutine 1 [chan send]: main.main() .../src/github.com/pprof/studygo/day06/channel02/main.go:8 +0x54 */ // ② 协程发送，无人接收 func recv(c chan int) { fmt.Println("发送成功") c <- 10 time.Sleep(time.Second*2) } func main() { ch := make(chan int) go recv(ch) // 启用goroutine从通道接收值 // // ret := <-ch time.Sleep(time.Second) //fmt.Println("接收成功", ret) } // 编译通过 // ③ 主线程接收，无人发送 func recv(c chan int) { // c <- 10 fmt.Println("发送成功") time.Sleep(time.Second*2) } func main() { ch := make(chan int) go recv(ch) // 启用goroutine从通道接收值 // ret := <-ch time.Sleep(time.Second) fmt.Println("接收成功", ret) } /* 发送成功 fatal error: all goroutines are asleep - deadlock! goroutine 1 [chan receive]: main.main() E:/桌面整理/00零声学院/0voice_go/2.资料/01-GO语言基础精讲和test方法/1-src/1-src/src/3/printa-z.go:17 +0x85 exit status 2 */ // ④ 协程接收，无人发送 func recv(c chan int) { ret := <-c time.Sleep(time.Second) fmt.Println("接收成功", ret) } func main() { ch := make(chan int) go recv(ch) // 启用goroutine从通道接收值 time.Sleep(time.Second*2) } /* 无输出结果 */ :::success 有缓冲区的通道：使用make函数初始化通道的时候为其指定通道的容量，如make(chan int, 1)； go并发编程（2） - 图3 只要通道的容量大于零，那么该通道就是有缓冲的通道，通道的容量表示通道中能存放元素的数量； 我们可以使用内置的len函数获取通道内元素的数量，使用cap函数获取通道的容量。

主线程不能永远阻塞**，若逻辑出现问题就会报deadlock的错误；**

:::

  1. func main() {
  2.     ch := make(chan int, 1) // 创建一个容量为1的有缓冲区通道
  3.     ch <- 10
  4.     fmt.Println("发送成功")
  5. }
  6. // 不会报错
  8. func main() {
  9.     ch := make(chan int, 1) // 创建一个容量为1的有缓冲区通道
  10.     ch <- 10
  11.     ch <- 10
  12.     fmt.Println("发送成功")
  13. }
  14. // 又会报死锁错，主要思想就是主线程不能阻塞

:::success 正确的通道读方式：当通过通道发送有限的数据时，我们可以通过close函数关闭通道来告知从该通道接收值的goroutine停止等待。当通道被关闭时，往该通道发送值会引发panic，从该通道里接收的值一直都是类型零值。如果通道里的元素是整型的，读操作是不能通过返回值来确定通道是否关闭的。

有两种方式在接收值的时候判断通道是否被关闭，我们通常使用的是for range的方式。

:::

  1. package main
  3. import "fmt"
  5. func main() {
  6.     c := make(chan int)
  7.     go func() {
  8.         for i := 0; i < 5; i++ {
  9.             c <- i
  10.         }
  11.         close(c)
  12.     }()
  13.     for {
  14.         if data, ok := <-c; ok {
  15.             fmt.Println(data)
  16.         } else {
  17.             break
  18.         }
  19.     }
  20.     fmt.Println("main结束")
  21. }
  22. /*
  23. PS E:\2.资料\01-GO语言基础精讲和test方法\1-src\1-src\src> go run .\3\printa-z.go
  24. 0
  25. 1
  26. 2
  27. 3
  28. 4
  29. main结束
  30. PS E:\2.资料\01-GO语言基础精讲和test方法\1-src\1-src\src> 
  31. */
  1. package main
  3. import "fmt"
  5. // channel 练习
  6. func main() {
  7.     ch1 := make(chan int)
  8.     ch2 := make(chan int)
  9.     // 开启goroutine将0~100的数发送到ch1中
  10.     go func() {
  11.         for i := 0; i < 100; i++ {
  12.             ch1 <- i
  13.         }
  14.         close(ch1)
  15.     }()
  16.     // 开启goroutine从ch1中接收值，并将该值的平方发送到ch2中
  17.     go func() {
  18.         for {
  19.             i, ok := <-ch1 // 通道关闭后再取值ok=false
  20.             if !ok {
  21.                 break
  22.             }
  23.             ch2 <- i * i
  24.         }
  25.         close(ch2)
  26.     }()
  27.     // 在主goroutine中从ch2中接收值打印
  28.     for i := range ch2 { // 通道关闭后会退出for range循环
  29.         fmt.Println(i)
  30.     }
  31. }
  32. /*
  33. PS E:\01-GO语言基础精讲和test方法\1-src\1-src\src> go run .\3\printa-z.go
  34. 0
  35. 1
  36. 4
  37. 9
  38. 16
  39. 25
  40. 36
  41. 49
  42. 64
  43. 81
  44. 100
  45. 121
  46. 144
  47. 169
  48. 196
  49. 225
  50. 256
  51. 289
  52. 324
  53. 361
  54. 400
  55. 441
  56. 484
  57. 529
  58. 576
  59. 625
  60. 676
  61. 729
  62. 784
  63. 841
  64. 900
  65. 961
  66. 1024
  67. 1089
  68. 1156
  69. 1225
  70. 1296
  71. 1369
  72. 1444
  73. 1521
  74. 1600
  75. 1681
  76. 1764
  77. 1849
  78. 1936
  79. 2025
  80. 2116
  81. 2209
  82. 2304
  83. 2401
  84. 2500
  85. 2601
  86. 2704
  87. 2809
  88. 2916
  89. 3025
  90. 3136
  91. 3249
  92. 3364
  93. 3481
  94. 3600
  95. 3721
  96. 3844
  97. 3969
  98. 4096
  99. 4225
  100. 4356
  101. 4489
  102. 4624
  103. 4761
  104. 4900
  105. 5041
  106. 5184
  107. 5329
  108. 5476
  109. 5625
  110. 5776
  111. 5929
  112. 6084
  113. 6241
  114. 6400
  115. 6561
  116. 6724
  117. 6889
  118. 7056
  119. 7225
  120. 7396
  121. 7569
  122. 7744
  123. 7921
  124. 8100
  125. 8281
  126. 8464
  127. 8649
  128. 8836
  129. 9025
  130. 9216
  131. 9409
  132. 9604
  133. 9801
  134. PS E:\01-GO语言基础精讲和test方法\1-src\1-src\src> 
  135. */

:::success 通道写安全：

单写多读确保通道写安全的最好方式是由负责写通道的协程自己来关闭通道，读通道的协程不要去关闭通道。

多写确保通道写安全方式是使用到内置 sync 包提供的 WaitGroup 对象，它使用计数来等待指定事件完成。

:::

:::success 单向通道：有的时候我们会将通道作为参数在多个任务函数间传递，很多时候我们在不同的任务函数中使用通道都会对其进行限制，比如限制通道在函数中只能发送或只能接收。

**chan<- int**是一个只能发送的通道，可以发送但是不能接收；

**<-chan int**是一个只能接收的通道，可以接收但是不能发送。

函数传参及任何赋值操作中将双向通道转换为单向通道是可以的，但反过来是不可以的。

:::

  1. func counter(out chan<- int) {
  2.     for i := 0; i < 100; i++ {
  3.         out <- i
  4.     }
  5.     close(out)
  6. }
  8. func squarer(out chan<- int, in <-chan int) {
  9.     for i := range in {
  10.         out <- i * i
  11.     }
  12.     close(out)
  13. }
  14. func printer(in <-chan int) {
  15.     for i := range in {
  16.         fmt.Println(i)
  17.     }
  18. }
  20. func main() {
  21.     ch1 := make(chan int)
  22.     ch2 := make(chan int)
  23.     go counter(ch1)
  24.     go squarer(ch2, ch1)
  25.     printer(ch2)
  26. }
  1. // 2-6 WaitGroup 在写端关闭channel对单写的程序有效，但是多写的时候呢？
  2. package main
  4. import (
  5.     "fmt"
  6.     "sync"
  7.     "time"
  8. )
  10. func send(ch chan int, wg *sync.WaitGroup) {
  11.     defer wg.Done() // 计数值减一
  12.     i := 0
  13.     for i < 4 {
  14.         i++
  15.         ch <- i
  16.     }
  17. }
  19. func recv(ch chan int) {
  20.     for v := range ch {
  21.         fmt.Println(v)
  22.     }
  23. }
  25. // 只要一个值能做界定符 比如nil， 比如0xfffe
  26. func main() {
  27.     var ch = make(chan int, 4)
  28.     var wg = new(sync.WaitGroup)
  29.     wg.Add(2)       // 增加计数值
  30.     go send(ch, wg) // 写
  31.     go send(ch, wg) // 写
  32.     go recv(ch)
  33.     // Wait() 阻塞等待所有的写通道协程结束
  34.     // 待计数值变成零，Wait() 才会返回
  35.     wg.Wait()
  36.     // 关闭通道
  37.     close(ch)
  38.     time.Sleep(time.Second)
  39. }
  40. /*
  41. PS E:\3-src\3-src\src>     go run .\2\2-6-channel-waitgroup.go
  42. 1
  43. 2
  44. 3
  45. 4
  46. 1
  47. 2
  48. 3
  49. 4
  50. PS E
  51. */

:::success go并发编程（2） - 图4

:::

③、WaitGroup

:::success **<font style="color:rgb(51, 51, 51);">sync.WaitGroup</font>**来实现并发任务的同步；引包<font style="color:rgb(51, 51, 51);">import "sync"</font>;

sync.WaitGroup内部维护着一个计数器，计数器的值可以增加和减少。例如当我们启动了N 个并发任务时，就将计数器值增加N。每个任务完成时通过调用Done()方法将计数器减1。通过调用Wait()来等待并发任务执行完，当计数器值为0时，表示所有并发任务已经完成。

:::

方法名 功能
(wg WaitGroup) *Add(delta int) 计数器+delta
(wg WaitGroup) *Done() 计数器-1
(wg WaitGroup) *Wait() 阻塞直到计数器变为0 
  1. var wg sync.WaitGroup
  3. func hello() {
  4.     defer wg.Done()
  5.     fmt.Println("Hello Goroutine!")
  6. }
  7. func main() {
  8.     wg.Add(1)
  9.     go hello() // 启动另外一个goroutine去执行hello函数
  10.     fmt.Println("main goroutine done!")
  11.     wg.Wait()
  12. }

:::success 需要注意sync.WaitGroup是一个结构体，传递参数的时候要传递指针。

:::

④、多路复用select

在某些场景下我们需要同时从多个通道接收数据。通道在接收数据时，如果没有数据可以接收将会发生阻塞。Go内置了**<font style="color:rgb(51, 51, 51);">select</font>**关键字，可以同时响应多个通道的操作。

<font style="color:rgb(51, 51, 51);">select</font>的使用类似于<font style="color:rgb(51, 51, 51);">switch</font>语句，它有一系列<font style="color:rgb(51, 51, 51);">case</font>分支和一个<font style="color:rgb(51, 51, 51);">默认</font>的分支。每个case会对应一个通道的通信（接收或发送）过程。<font style="color:rgb(51, 51, 51);">select</font>会一直等待，直到某个<font style="color:rgb(51, 51, 51);">case</font>的通信操作完成时，就会执行case分支对应的语句。具体格式如下：

  1. select {
  2.     case <-chan1:
  3.     // 如果chan1成功读到数据，则进行该case处理语句
  4.     case chan2 <- 1:
  5.     // 如果成功向chan2写入数据，则进行该case处理语句
  6.     default:
  7.     // 如果上面都没有成功，则进入default处理流程
  8. }

:::success select功能：

  • select可以同时监听一个或多个channel，直到其中一个channel ready；
  • 如果多个channel同时ready，则随机选择一个执行
  • 可以用于判断管道是否存满

:::

  1. package main
  3. import (
  4.    "fmt"
  5.    "time"
  6. )
  8. func test1(ch chan string) {
  9.    time.Sleep(time.Second * 5)
  10.    ch <- "test1"
  11. }
  12. func test2(ch chan string) {
  13.    time.Sleep(time.Second * 2)
  14.    ch <- "test2"
  15. }
  17. func main() {
  18.    // 2个管道
  19.    output1 := make(chan string)
  20.    output2 := make(chan string)
  21.    // 跑2个子协程，写数据
  22.    go test1(output1)
  23.    go test2(output2)
  24.    // 用select监控
  25.    select {
  26.    case s1 := <-output1:
  27.       fmt.Println("s1=", s1)
  28.    case s2 := <-output2:
  29.       fmt.Println("s2=", s2)
  30.    }
  31. }
  33. /*  s2= test2   */
  1. package main
  3. import (
  4.    "fmt"
  5. )
  7. func main() {
  8.    // 创建2个管道
  9.    int_chan := make(chan int, 1)
  10.    string_chan := make(chan string, 1)
  11.    go func() {
  12.       //time.Sleep(2 * time.Second)
  13.       int_chan <- 1
  14.    }()
  15.    go func() {
  16.       string_chan <- "hello"
  17.    }()
  18.    select {
  19.    case value := <-int_chan:
  20.       fmt.Println("int:", value)
  21.    case value := <-string_chan:
  22.       fmt.Println("string:", value)
  23.    }
  24.    fmt.Println("main结束")
  25. }
  26. /*
  27. string: hello
  28. main结束
  29. */
  1. package main
  3. import (
  4.    "fmt"
  5.    "time"
  6. )
  8. // 判断管道有没有存满
  9. func main() {
  10.    // 创建管道
  11.    output1 := make(chan string, 10)
  12.    // 子协程写数据
  13.    go write(output1)
  14.    // 取数据
  15.    for s := range output1 {
  16.       fmt.Println("res:", s)
  17.       time.Sleep(time.Second)
  18.    }
  19. }
  21. func write(ch chan string) {
  22.    for {
  23.       select {
  24.       // 写数据
  25.       case ch <- "hello":
  26.          fmt.Println("write hello")
  27.       default:
  28.          fmt.Println("channel full")
  29.       }
  30.       time.Sleep(time.Millisecond * 500)
  31.    }
  32. }
  33. /*
  34. res: hello
  35. write hello
  36. write hello
  37. write hello
  38. res: hello
  39. write hello
  40. res: hello
  41. write hello
  42. write hello
  43. res: hello
  44. write hello
  45. write hello
  46. res: hello
  47. write hello
  48. write hello
  49. res: hello
  50. write hello
  51. write hello
  52. res: hello
  53. write hello
  54. write hello
  55. res: hello
  56. write hello
  57. write hello
  58. res: hello
  59. write hello
  60. write hello
  61. res: hello
  62. write hello
  63. write hello
  64. res: hello
  65. write hello
  66. channel full
  67. res: hello
  68. write hello
  69. channel full
  70. res: hello
  71. write hello
  72. channel full
  73. res: hello
  74. write hello
  75. channel full
  76. res: hello
  77. write hello
  78. channel full
  79. res: hello
  80. write hello
  81. channel full
  82. res: hello
  83. write hello
  84. channel full
  85. res: hello
  86. write hello
  87. channel full
  88. res: hello
  89. write hello
  90. channel full
  91. res: hello
  92. write hello
  93. channel full
  94. res: hello
  95. write hello
  96. channel full
  97. res: hello
  98. write hello
  99. channel full
  100. res: hello
  101. write hello
  102. channel full
  103. res: hello
  104. write hello
  105. channel full
  106. res: hello
  107. write hello
  108. channel full
  109. res: hello
  110. write hello
  111. channel full
  112. res: hello
  113. write hello
  114. channel full
  115. res: hello
  116. write hello
  117. channel full
  118. res: hello
  119. write hello
  120. channel full
  121. res: hello
  122. write hello
  123. channel full
  124. res: hello
  125. write hello
  126. channel full
  127. res: hello
  128. write hello
  129. channel full
  130. res: hello
  131. write hello
  132. channel full
  133. res: hello
  134. write hello
  135. channel full
  136. res: hello
  137. write hello
  138. channel full
  139. res: hello
  140. write hello
  141. channel full
  142. res: hello
  143. write hello
  144. channel full
  145. res: hello
  146. write hello
  147. channel full
  148. res: hello
  149. write hello
  150. channel full
  151. res: hello
  152. write hello
  153. channel full
  154. res: hello
  155. */

:::success 非阻塞读写：当通道空时，读操作不会阻塞，当通道满时，写操作也不会阻塞。非阻塞读写需要依靠 select 语句的 default 分支。当 select 语句所有通道都不可读写时，如果定义了 default 分支，那就会执行 default 分支逻辑，这样就起到了不阻塞的效果。

:::

  1. // 非阻塞读写
  2. package main
  5. import (
  6.     "fmt"
  7.     "time"
  8. )
  11. func send(ch1 chan int, ch2 chan int) {
  12.     i := 0
  13.     for {
  14.         i++
  15.         select {
  16.         case ch1 <- i:
  17.             fmt.Printf("send ch1 %d\n", i)
  18.         case ch2 <- i:
  19.             fmt.Printf("send ch2 %d\n", i)
  20.         default:
  21.             fmt.Printf("ch block\n")
  22.             time.Sleep(2 * time.Second) // 这里只是为了演示
  23.         }
  24.     }
  25. }
  28. func recv(ch chan int, gap time.Duration, name string) {
  29.     for v := range ch {
  30.         fmt.Printf("receive %s %d\n", name, v)
  31.         time.Sleep(gap)
  32.     }
  33. }
  36. func main() {
  37.     // 无缓冲通道
  38.     var ch1 = make(chan int)
  39.     var ch2 = make(chan int)
  40.     // 两个消费者的休眠时间不一样，名称不一样
  41.     go recv(ch1, time.Second, "ch1")
  42.     go recv(ch2, 2*time.Second, "ch2")
  43.     send(ch1, ch2)
  44. }
  45. /*    截取开始的一部分
  46. ch block
  47. send ch2 2
  48. send ch1 3
  49. ch block
  50. receive ch1 3
  51. receive ch2 2
  52. send ch1 5
  53. send ch2 6
  54. ch block
  55. receive ch2 6
  56. receive ch1 5
  57. send ch1 8
  58. send ch2 9
  59. ch block
  60. receive ch2 9
  61. receive ch1 8
  62. send ch1 11
  63. send ch2 12
  64. ch block
  65. receive ch2 12
  66. receive ch1 11
  67. send ch2 14
  68. send ch1 15
  69. ch block
  70. */

⑤、生产者消费者模型

  1. // 生产者、消费者模型
  2. package main
  5. import (
  6.     "fmt"
  7.     "os"
  8.     "os/signal"
  9.     "syscall"
  10.     "time"
  11. )
  14. // 生产者
  15. func Producer(factor int, out chan<- int) {
  16.     for i := 0; ; i++ {
  17.         out <- i * factor
  18.         time.Sleep(3 * time.Second)
  19.     }
  20. }
  23. // 消费者
  24. func Consumer(in <-chan int) {
  25.     for v := range in {
  26.         fmt.Println(v)
  27.     }
  28. }
  31. func main() {
  32.     ch := make(chan int, 64)
  33.     go Producer(3, ch) // 生成3的倍数序列
  34.     go Producer(5, ch) // 生成5的倍数序列
  37.     go Consumer(ch)
  40.     //Ctrl +C 退出
  41.     sig := make(chan os.Signal, 1)
  42.     signal.Notify(sig, syscall.SIGINT, syscall.SIGTERM)
  43.     fmt.Printf("wait Ctrl +C\n")
  44.     fmt.Printf("quit (%v)\n", <-sig)
  45. }

⑥、并发安全和锁

:::success 数据竞态：可能会存在多个goroutine同时操作一个资源（临界区），这种情况会发生竞态问题；

竞态检查工具**：是基于运行时代码检查，而不是通过代码静态分析来完成的。这意味着那些没有机会运行到的代码逻辑中如果存在安全隐患，它是检查不出来的。需要加上-race 执行**。go run <font style="color:#DF2A3F;">-race</font> 代码

:::

:::success 互斥锁（**sync.Mutex**）：互斥锁是一种常用的控制共享资源访问的方法，它能够保证同时只有一个goroutine可以访问共享资源。Go语言中使用sync包的Mutex类型来实现互斥锁。

:::

  1. // 不加锁版本    输出6379
  2. var x int64
  3. var wg sync.WaitGroup
  4. var lock sync.Mutex
  6. func add() {
  7.     for i := 0; i < 5000; i++ {
  8.         lock.Lock() // 加锁
  9.         x = x + 1
  10.         lock.Unlock() // 解锁
  11.     }
  12.     wg.Done()
  13. }
  14. func main() {
  15.     wg.Add(2)
  16.     go add()
  17.     go add()
  18.     wg.Wait()
  19.     fmt.Println(x)
  20. }
  22. // 加锁版本  输出10000
  23. var x int64
  24. var wg sync.WaitGroup
  25. var lock sync.Mutex
  27. func add() {
  28.     for i := 0; i < 5000; i++ {
  29.         lock.Lock() // 加锁
  30.         x = x + 1
  31.         lock.Unlock() // 解锁
  32.     }
  33.     wg.Done()
  34. }
  35. func main() {
  36.     wg.Add(2)
  37.     go add()
  38.     go add()
  39.     wg.Wait()
  40.     fmt.Println(x)
  41. }

:::success 读写互斥锁：互斥锁是完全互斥的，但是有很多实际的场景下是读多写少的，当我们并发的去读取一个资源不涉及资源修改的时候是没有必要加锁的，这种场景下使用读写锁是更好的一种选择。读写锁在Go语言中使用sync包中的<font style="color:rgb(51, 51, 51);">RWMutex</font>类型。

读写锁分为两种：读锁和写锁。当一个goroutine获取读锁之后，其他的goroutine如果是获取读锁会继续获得锁，如果是获取写锁就会等待；当一个goroutine获取写锁之后，其他的goroutine无论是获取读锁还是写锁都会等待。

读写锁优势：读写锁非常适合读多写少的场景，如果读和写的操作差别不大，读写锁的优势就发挥不出来。

:::

  1. package main
  4. import (
  5.    "fmt"
  6.    "time"
  7.    "sync"
  8. )
  10. var (
  11.     x      int64
  12.     wg     sync.WaitGroup
  13.     lock   sync.Mutex
  14.     rwlock sync.RWMutex
  15. )
  18. func write() {
  19.     lock.Lock()   // 加互斥锁
  20.     // rwlock.Lock() // 加写锁
  21.     x = x + 1
  22.     time.Sleep(10 * time.Millisecond) // 假设写操作耗时10毫秒
  23.     // rwlock.Unlock()                   // 解写锁
  24.     lock.Unlock()                     // 解互斥锁
  25.     wg.Done()
  26. }
  29. func read() {
  30.     lock.Lock()                  // 加互斥锁
  31.     // rwlock.RLock()               // 加读锁
  32.     time.Sleep(time.Millisecond) // 假设读操作耗时1毫秒
  33.     // rwlock.RUnlock()             // 解读锁
  34.     lock.Unlock()                // 解互斥锁
  35.     wg.Done()
  36. }
  39. func main() {
  40.     start := time.Now()
  41.     for i := 0; i < 10; i++ {
  42.         wg.Add(1)
  43.         go write()
  44.     }
  47.     for i := 0; i < 1000; i++ {
  48.         wg.Add(1)
  49.         go read()
  50.     }
  53.     wg.Wait()
  54.     end := time.Now()
  55.     fmt.Println(end.Sub(start))
  56. }
  57. // 读写锁 146.8492ms
  58. // 互斥锁 15.0116383s

:::success 避免锁复制：sync.Mutex是一个结构体对象，这个对象在使用的过程中要避免被复制 —— 浅拷贝。复制会导致锁被「分裂」了，也就起不到保护的作用。所以在平时的使用中要尽量使用它的指针类型。

锁复制存在于结构体变量的赋值、函数参数传递、方法参数传递中，都需要注意。

:::

  1. package main
  3. import (
  4.     "fmt"
  5.     "sync"
  6. )
  8. type SafeDict struct {
  9.     data  map[string]int
  10.     mutex *sync.Mutex
  11. }
  13. func NewSafeDict(data map[string]int) *SafeDict {
  14.     return &SafeDict{
  15.         data:  data,
  16.         mutex: &sync.Mutex{},
  17.     }
  18. }
  20. // defer 语句总是要推迟到函数尾部运行，所以如果函数逻辑运行时间比较长，
  21. // 这会导致锁持有的时间较长，这时使用 defer 语句来释放锁未必是一个好注意。
  22. func (d *SafeDict) Len() int {
  23.     d.mutex.Lock()
  24.     defer d.mutex.Unlock()
  25.     return len(d.data)
  26. }
  28. // func (d *SafeDict) Test() int {
  29. //     d.mutex.Lock()
  30. //     length := len(d.data)
  31. //     d.mutex.Unlock() // 手动解锁 减少粒度    // 这种情况就不要用 defer d.mutex.Unlock()
  32. //     fmt.Println("length: ", length)
  33. //     // 这里还有耗时处理 耗时1000ms
  34. // }
  36. func (d *SafeDict) Put(key string, value int) (int, bool) {
  37.     d.mutex.Lock()
  38.     defer d.mutex.Unlock()
  39.     old_value, ok := d.data[key]
  40.     d.data[key] = value
  41.     return old_value, ok
  42. }
  44. func (d *SafeDict) Get(key string) (int, bool) {
  45.     d.mutex.Lock()
  46.     defer d.mutex.Unlock()
  47.     old_value, ok := d.data[key]
  48.     return old_value, ok
  49. }
  51. func (d *SafeDict) Delete(key string) (int, bool) {
  52.     d.mutex.Lock()
  53.     defer d.mutex.Unlock()
  54.     old_value, ok := d.data[key]
  55.     if ok {
  56.         delete(d.data, key)
  57.     }
  58.     return old_value, ok
  59. }
  61. func write(d *SafeDict) {
  62.     d.Put("banana", 5)
  63. }
  65. func read(d *SafeDict) {
  66.     fmt.Println(d.Get("banana"))
  67. }
  69. // go run -race 3-2-lock.go
  70. func main() {
  71.     d := NewSafeDict(map[string]int{
  72.         "apple": 2,
  73.         "pear":  3,
  74.     })
  75.     go read(d)
  76.     write(d)
  77. }

:::info 匿名锁字段：在结构体章节，我们知道外部结构体可以自动继承匿名内部结构体的所有方法。如果将上面的SafeDict 结构体进行改造，将锁字段匿名，就可以稍微简化一下代码。

:::

  1. package main
  3. import (
  4.     "fmt"
  5.     "sync"
  6. )
  8. type SafeDict struct {
  9.     data map[string]int
  10.     *sync.Mutex
  11. }
  13. func NewSafeDict(data map[string]int) *SafeDict {
  14.     return &SafeDict{
  15.         data,
  16.         &sync.Mutex{}, // 一样是要初始化的
  17.     }
  18. }
  20. func (d *SafeDict) Len() int {
  21.     d.Lock()
  22.     defer d.Unlock()
  23.     return len(d.data)
  24. }
  26. func (d *SafeDict) Put(key string, value int) (int, bool) {
  27.     d.Lock()
  28.     defer d.Unlock()
  29.     old_value, ok := d.data[key]
  30.     d.data[key] = value
  31.     return old_value, ok
  32. }
  34. func (d *SafeDict) Get(key string) (int, bool) {
  35.     d.Lock()
  36.     defer d.Unlock()
  37.     old_value, ok := d.data[key]
  38.     return old_value, ok
  39. }
  41. func (d *SafeDict) Delete(key string) (int, bool) {
  42.     d.Lock()
  43.     defer d.Unlock()
  44.     old_value, ok := d.data[key]
  45.     if ok {
  46.         delete(d.data, key)
  47.     }
  48.     return old_value, ok
  49. }
  51. func write(d *SafeDict) {
  52.     d.Put("banana", 5)
  53. }
  55. func read(d *SafeDict) {
  56.     fmt.Println(d.Get("banana"))
  57. }
  59. func main() {
  60.     d := NewSafeDict(map[string]int{
  61.         "apple": 2,
  62.         "pear":  3,
  63.     })
  64.     go read(d)
  65.     write(d)
  66. }

⑦、发布订阅练习

⑧、Context上下文

:::success Context：每一个处理都应该有个超时限制，需要在调用中传递这个超时，Context是协程安全的。代码中可以将单个Context传递给任意数量的goroutine，并在取消该Context时可以将信号传递给所有的goroutine。

对服务器传入的请求应该创建上下文，而对服务器的传出调用应该接受上下文。它们之间的函数调用链必须传递上下文，或者可以使用WithCancel、WithDeadline、WithTimeout或WithValue创建的派生上下文。当一个上下文被取消时，它派生的所有上下文也被取消。

:::

:::success Context接口：

◼ **Deadline**方法是获取设置的截止时间的意思，第一个返回式是截止时间，到了这个时间点，Context会自动发起取消请求；第二个返回值ok==false时表示没有设置截止时间，如果需要取消的话，需要调用取消函数进行取消；

◼ **<font style="color:#DF2A3F;">Done</font>**方法返回一个只读的chan，类型为struct{}，我们在goroutine中，如果该方法返回的chan可以读取，则意味着parent context已经发起了取消请求，我们通过Done方法收到这个信号后，就应该做清理操作，然后退出goroutine，释放资源；

◼ **Err**方法返回取消的错误原因，因为什么Context被取消；

  • 如果当前Context被取消就会返回<font style="color:rgb(51, 51, 51);">Canceled</font>错误；
  • 如果当前Context超时就会返回<font style="color:rgb(51, 51, 51);">DeadlineExceeded</font>错误；

◼ **Value**方法获取该Context上绑定的值，是一个键值对，所以要通过一个Key才可以获取对应的值，这个值一般是线程安全的。

:::

  1. type Context interface {
  2.     Deadline() (deadline time.Time, ok bool)
  3.     Done() <-chan struct{}
  4.     Err() error
  5.     Value(key interface{}) interface{}
  6. }

:::success Background()和TODO()：

Go内置两个函数：**<font style="color:#DF2A3F;">Background()和TODO()</font>**，这两个函数分别返回一个实现了Context接口的background和todo。我们代码中最开始都是以这两个内置的上下文对象作为最顶层的partent context，衍生出更多的子上下文对象。 Background()主要用于main函数、初始化以及测试代码中，作为Context这个树结构的最顶层的Context，也就是根Context。 TODO()，它目前还不知道具体的使用场景，如果我们不知道该使用什么Context的时候，可以使用这个。 background和todo本质上都是emptyCtx结构体类型，是一个不可取消，没有设置截止时间，没有携带任何值的Context。

:::

:::success With系列函数：四个With函数，接收的都有一个partent参数，就是父Context，我们要基于这个父Context创建出子Context的意思；

◼ **WithCancel**函数，传递一个父Context作为参数，返回子Context，以及一个取消函数用来取消Context

◼ **WithDeadline**函数，和WithCancel差不多，它会多传递一个截止时间参数，意味着到了这个时间点，会自动取消Context，当然我们也可以不等到这个时候，可以提前通过取消函数进行取消

◼ **WithTimeout**和WithDeadline基本上一样，这个表示是超时自动取消，是多少时间后自动取消Context的意思，只是传参数不一样。

◼ **WithValue**函数和取消Context无关，它是为了生成一个绑定了一个键值对数据的Context，这个绑定的数据可以通过Context.Value方法访问到，WithValue返回父节点的副本，其中与key关联的值为val。仅对API和进程间传递请求域的数据使用上下文值，而不是使用它来传递可选参数给函数。

:::

  1. func WithCancel(parent Context) (ctx Context, cancel CancelFunc)
  2. func WithDeadline(parent Context, deadline time.Time) (Context, CancelFunc)
  3. func WithTimeout(parent Context, timeout time.Duration) (Context, CancelFunc)
  4. func WithValue(parent Context, key interface{}, val interface{}) Context
  1. func gen(ctx context.Context) <-chan int {
  2.         dst := make(chan int)
  3.         n := 1
  4.         go func() {
  5.             for {
  6.                 select {
  7.                 case <-ctx.Done():
  8.                     return // return结束该goroutine，防止泄露
  9.                 case dst <- n:
  10.                     n++
  11.                 }
  12.             }
  13.         }()
  14.         return dst
  15.     }
  16. func main() {
  17.     ctx, cancel := context.WithCancel(context.Background())
  18.     defer cancel() // 当我们取完需要的整数后调用cancel
  20.     for n := range gen(ctx) {
  21.         fmt.Println(n)
  22.         if n == 5 {
  23.             break
  24.         }
  25.     }
  26. }
  27. /*上面的示例代码中，gen函数在单独的goroutine中生成整数并将它们发送到返回的通道。
  28. gen的调用者在使用生成的整数之后需要取消上下文，以免gen启动的内部goroutine发生泄漏*/
  29. /*  输出 
  30. 1
  31. 2
  32. 3
  33. 4
  34. 5
  35. */
  1. package main
  3. import (
  4.     "fmt"
  6.     "time"
  7.     "context"
  8. )
  11. func main() {
  12.     d := time.Now().Add(50 * time.Millisecond)
  13.     ctx, cancel := context.WithDeadline(context.Background(), d)
  15.     // 尽管ctx会过期，但在任何情况下调用它的cancel函数都是很好的实践。
  16.     // 如果不这样做，可能会使上下文及其父类存活的时间超过必要的时间。
  17.     defer cancel()
  19.     select {
  20.     case <-time.After(1 * time.Second):
  21.         fmt.Println("overslept")
  22.     case <-ctx.Done():
  23.         fmt.Println(ctx.Err())
  24.     }
  25. } 
  27. //  输出 context deadline exceeded
  1. package main
  3. import (
  4.     "context"
  5.     "fmt"
  6.     "sync"
  8.     "time"
  9. )
  11. // context.WithTimeout
  13. var wg sync.WaitGroup
  15. func worker(ctx context.Context) {
  16. LOOP:
  17.     for {
  18.         fmt.Println("db connecting ...")
  19.         time.Sleep(time.Millisecond * 10) // 假设正常连接数据库耗时10毫秒
  20.         select {
  21.         case <-ctx.Done(): // 50毫秒后自动调用
  22.             break LOOP
  23.         default:
  24.         }
  25.     }
  26.     fmt.Println("worker done!")
  27.     wg.Done()
  28. }
  30. func main() {
  31.     // 设置一个50毫秒的超时
  32.     ctx, cancel := context.WithTimeout(context.Background(), time.Millisecond*50)
  33.     wg.Add(1)
  34.     go worker(ctx)
  35.     time.Sleep(time.Second * 5)
  36.     cancel() // 通知子goroutine结束
  37.     wg.Wait()
  38.     fmt.Println("over")
  39. }
  40. /*
  41. db connecting ...
  42. db connecting ...
  43. db connecting ...
  44. worker done!
  45. over
  46. */
  1. package main
  3. import (
  4.     "context"
  5.     "fmt"
  6.     "sync"
  8.     "time"
  9. )
  11. // context.WithValue
  13. type TraceCode string
  15. var wg sync.WaitGroup
  17. func worker(ctx context.Context) {
  18.     key := TraceCode("TRACE_CODE")
  19.     traceCode, ok := ctx.Value(key).(string) // 在子goroutine中获取trace code
  20.     if !ok {
  21.         fmt.Println("invalid trace code")
  22.     }
  23. LOOP:
  24.     for {
  25.         fmt.Printf("worker, trace code:%s\n", traceCode)
  26.         time.Sleep(time.Millisecond * 10) // 假设正常连接数据库耗时10毫秒
  27.         select {
  28.         case <-ctx.Done(): // 50毫秒后自动调用
  29.             break LOOP
  30.         default:
  31.         }
  32.     }
  33.     fmt.Println("worker done!")
  34.     wg.Done()
  35. }
  37. func main() {
  38.     // 设置一个50毫秒的超时
  39.     ctx, cancel := context.WithTimeout(context.Background(), time.Millisecond*50)
  40.     // 在系统的入口中设置trace code传递给后续启动的goroutine实现日志数据聚合
  41.     ctx = context.WithValue(ctx, TraceCode("TRACE_CODE"), "12512312234")
  42.     wg.Add(1)
  43.     go worker(ctx)
  44.     time.Sleep(time.Second * 5)
  45.     cancel() // 通知子goroutine结束
  46.     wg.Wait()
  47.     fmt.Println("over")
  48. }
  50. /*
  51. worker, trace code:12512312234
  52. worker, trace code:12512312234
  53. worker, trace code:12512312234
  54. worker, trace code:12512312234
  55. worker done!
  56. over
  57. */

:::success Context使用原则：

◼ 不要把Context放在结构体中，要以参数的方式进行传递；

◼ 以 Context 作为参数的函数方法，应该把 Context 作为第一个参数；

◼ 给一个函数方法传递Context的时候，不要传递nil，如果不知道传递什么，就使用context.TODO；

◼ Context 的 Value 相关方法应该传递请求域的必要数据，不应该用于传递可选参数；

◼ Context 是线程安全的，可以放心的在多个 Goroutine 中传递。

:::

:::success contexts派生上下文：Context包提供了从现有Context值派生新Context值的函数。这些值形成一个树:当一个Context被取消时，从它派生的所有Context也被取消。

:::

⑨、调用服务端API时如何在客户端实现超时控制？

  1. // context_timeout/server/main.go
  2. package main
  4. import (
  5.     "fmt"
  6.     "math/rand"
  7.     "net/http"
  9.     "time"
  10. )
  12. // server端，随机出现慢响应
  14. func indexHandler(w http.ResponseWriter, r *http.Request) {
  15.     number := rand.Intn(2)
  16.     if number == 0 {
  17.         time.Sleep(time.Second * 10) // 耗时10秒的慢响应
  18.         fmt.Fprintf(w, "slow response")
  19.         return
  20.     }
  21.     fmt.Fprint(w, "quick response")
  22. }
  24. func main() {
  25.     http.HandleFunc("/", indexHandler)
  26.     err := http.ListenAndServe(":8000", nil)
  27.     if err != nil {
  28.         panic(err)
  29.     }
  30. }
  1. // context_timeout/client/main.go
  2. package main
  4. import (
  5.     "context"
  6.     "fmt"
  7.     "io/ioutil"
  8.     "net/http"
  9.     "sync"
  10.     "time"
  11. )
  13. // 客户端
  15. type respData struct {
  16.     resp *http.Response
  17.     err  error
  18. }
  20. func doCall(ctx context.Context) {
  21.     transport := http.Transport{
  22.        // 请求频繁可定义全局的client对象并启用长链接
  23.        // 请求不频繁使用短链接
  24.        DisableKeepAlives: true,     }
  25.     client := http.Client{
  26.         Transport: &transport,
  27.     }
  29.     respChan := make(chan *respData, 1)
  30.     req, err := http.NewRequest("GET", "http://127.0.0.1:8000/", nil)
  31.     if err != nil {
  32.         fmt.Printf("new requestg failed, err:%v\n", err)
  33.         return
  34.     }
  35.     req = req.WithContext(ctx) // 使用带超时的ctx创建一个新的client request
  36.     var wg sync.WaitGroup
  37.     wg.Add(1)
  38.     defer wg.Wait()
  39.     go func() {
  40.         resp, err := client.Do(req)
  41.         fmt.Printf("client.do resp:%v, err:%v\n", resp, err)
  42.         rd := &respData{
  43.             resp: resp,
  44.             err:  err,
  45.         }
  46.         respChan <- rd
  47.         wg.Done()
  48.     }()
  50.     select {
  51.     case <-ctx.Done():
  52.         //transport.CancelRequest(req)
  53.         fmt.Println("call api timeout")
  54.     case result := <-respChan:
  55.         fmt.Println("call server api success")
  56.         if result.err != nil {
  57.             fmt.Printf("call server api failed, err:%v\n", result.err)
  58.             return
  59.         }
  60.         defer result.resp.Body.Close()
  61.         data, _ := ioutil.ReadAll(result.resp.Body)
  62.         fmt.Printf("resp:%v\n", string(data))
  63.     }
  64. }
  66. func main() {
  67.     // 定义一个100毫秒的超时
  68.     ctx, cancel := context.WithTimeout(context.Background(), time.Millisecond*100)
  69.     defer cancel() // 调用cancel释放子goroutine资源
  70.     doCall(ctx)
  71. }
  72. /* 请求失败打印
  73. call api timeout
  74. client.do resp:<nil>, err:Get "http://127.0.0.1:8000/": context deadline exceeded   
  75. */
  76. /* 请求成功打印
  77. client.do resp:&{200 OK 200 HTTP/1.1 1 1 map[Content-Length:[14] Content-Type:[text/plain; charset=utf-8] Date:[Sat, 08 Apr 2023 13:55:48 GMT]] 0xc00019e040 14 [] true 
  78. false map[] 0xc00012e100 <nil>}, err:<nil>
  79. call server api success
  80. resp:quick response
  81. */