1. Goroutine

1.1. Go实现并发的方式

1.1.1. 并发与并行

  • 并发：同一时间段内执行多个任务，不同的任务通过上下文切换实现调度

  • 并行：同一时刻执行多个任务，多个任务在不同的CPU上同时运行

    1.1.2. Goroutine

    在java/c++中我们要实现并发编程的时候，我们通常需要自己维护一个线程池，并且需要自己去包装一个又一个的任务，同时需要自己去调度线程执行任务并维护上下文切换。
    Go语言中的goroutine概念类似于线程，但 goroutine是由Go的运行时（runtime）调度和管理的。Go程序会智能地将 goroutine 中的任务合理地分配给每个CPU。Go语言之所以被称为现代化的编程语言，就是因为它在语言层面已经内置了调度和上下文切换的机制。
    在Go语言编程中你不需要去自己写进程、线程、协程，你的技能包里只有一个技能–goroutine，当你需要让某个任务并发执行的时候，你只需要把这个任务包装成一个函数，开启一个goroutine去执行这个函数就可以了。
    OS线程（操作系统线程）一般都有固定的栈内存（通常为2MB）,一个goroutine的栈在其生命周期开始时只有很小的栈（典型情况下2KB），goroutine的栈不是固定的，他可以按需增大和缩小，goroutine的栈大小限制可以达到1GB，虽然极少会用到这个大。所以在Go语言中一次创建十万左右的goroutine也是可以的。
    单从线程调度讲，Go语言相比起其他语言的优势在于OS线程是由OS内核来调度的，goroutine则是由Go运行时（runtime）自己的调度器调度的，这个调度器使用一个称为m:n调度的技术（复用/调度m个goroutine到n个OS线程）。 其一大特点是goroutine的调度是在用户态下完成的， 不涉及内核态与用户态之间的频繁切换，包括内存的分配与释放，都是在用户态维护着一块大的内存池， 不直接调用系统的malloc函数（除非内存池需要改变），成本比调度OS线程低很多。 另一方面充分利用了多核的硬件资源，近似的把若干goroutine均分在物理线程上， 再加上本身goroutine的超轻量，以上种种保证了go调度方面的性能。

    1.1.3. GPM调度系统

    GPM是Go语言运行时（runtime）层面的实现，是go语言自己实现的一套调度系统。区别于操作系统调度OS线程。

  • G很好理解，就是个goroutine的，里面除了存放本goroutine信息外 还有与所在P的绑定等信息。

  • P管理着一组goroutine队列，P里面会存储当前goroutine运行的上下文环境（函数指针，堆栈地址及地址边界），P会对自己管理的goroutine队列做一些调度（比如把占用CPU时间较长的goroutine暂停、运行后续的goroutine等等）当自己的队列消费完了就去全局队列里取，如果全局队列里也消费完了会去其他P的队列里抢任务。P的个数是通过 runtime.GOMAXPROCS 设定（最大256），Go1.5版本之后默认为物理线程数。 在并发量大的时候会增加一些P和M，但不会太多，切换太频繁的话得不偿失
  • M（machine）是Go运行时（runtime）对操作系统内核线程的虚拟， M与内核线程一般是一一映射的关系， 一个groutine最终是要放到M上执行的；P与M一般也是一一对应的。他们关系是： P管理着一组G挂载在M上运行。当一个G长久阻塞在一个M上时，runtime会新建一个M，阻塞G所在的P会把其他的G 挂载在新建的M上。当旧的G阻塞完成或者认为其已经死掉时 回收旧的M。

    1.1.4. 原子操作

    原子操作是指并发编程中最小且不可以并行的操作，当并发线程对一个共享资源的的操作时原子操作的话，那么同一时间就只能有一个线程能操作该资源，实现方式就是加锁。

    1.2. Goroutine的使用

    1.2.1. goroutine

    golang 中使用 goroutine 非常方便，只需要将需要放到代码封装为一个函数(命名函数或者匿名函数)，然后使用 go func 执行函数即可。 ```go package main

import ( “fmt” “runtime” “time” )

func f0() { , , line, _ := runtime.Caller(0) fmt.Printf(“time:%v; 当前行号:%v\n”, time.Now().Format(“15:04:05”), line+1) }

func main() { go func() { , , line, _ := runtime.Caller(0) fmt.Printf(“time:%v; 当前行号:%v\n”, time.Now().Format(“15:04:05”), line+1) }() // 执行匿名函数 go f0() // 执行命名函数

  1. fmt.Printf("time:%v; 当前行号:%v\n", time.Now().Format("15:04:05"), 22)
  2. time.Sleep(time.Second)
  3. fmt.Printf("time:%v; 当前行号:%v\n", time.Now().Format("15:04:05"), 24)

}

  1. ```
  2. E:\Projects\Go\src\learn\day10-goroutine\goroutine>go run mian.go
  3. time:11:31:06; 当前行号:22
  4. time:11:31:06; 当前行号:17
  5. time:11:31:06; 当前行号:11
  6. time:11:31:07; 当前行号:24

1.2.2. WatiGroup

上述案例中，如果不加入 line22 会出现mian函数在go线程之前退出，main退出后，整个程序就结束了。在实际的编程中，主进程需要等待子进程完成所有任务后才能退出， sync.WaitGroup 用于实现goroutine同步。

  1. 1.    type WaitGroup struct { }
  2.         WaitGroup用于等待一组线程的结束。父线程调用Add方法来设定应等待的线程的数量。
  3.     每个被等待的线程在结束时应调用Done方法。同时，主线程里可以调用Wait方法阻塞至所有线程结束。
  5. 2.    func (wg *WaitGroup) Add(delta int)
  6.         Add方法向内部计数加上delta，delta可以是负数；
  7.     如果内部计数器变为0，Wait方法阻塞等待的所有线程都会释放，如果计数器小于0，方法panic。
  8.     注意Add加上正数的调用应在Wait之前，否则Wait可能只会等待很少的线程。
  9.     一般来说本方法应在创建新的线程或者其他应等待的事件之前调用。
  11. 3.    func (wg *WaitGroup) Done()
  12.         Done方法减少WaitGroup计数器的值，应在线程的最后执行，一般配合 defer 使用
  14. 4.    func (wg *WaitGroup) Wait()
  15.         Wait方法阻塞直到WaitGroup计数器减为0
  1. package main
  3. import (
  4.     "fmt"
  5.     "runtime"
  6.     "sync"
  7.     "time"
  8. )
  10. var wg sync.WaitGroup
  12. func f0() {
  13.     defer wg.Done()
  14.     _, _, line, _ := runtime.Caller(0)
  15.     fmt.Printf("time:%v; 当前行号:%v\n", time.Now().Format("15:04:05"), line+1)
  16. }
  18. func main() {
  19.     wg.Add(2)
  21.     go func() {
  22.         defer wg.Done()
  23.         _, _, line, _ := runtime.Caller(0)
  24.         fmt.Printf("time:%v; 当前行号:%v\n", time.Now().Format("15:04:05"), line+1)
  25.     }()
  26.     go f0()
  28.     fmt.Printf("time:%v; 当前行号:%v\n", time.Now().Format("15:04:05"), 28)
  29.     wg.Wait()
  30. }
  1. time:14:08:11; 当前行号:28
  2. time:14:08:11; 当前行号:15
  3. time:14:08:11; 当前行号:24

1.2.3. GOMAXPROCS

Go语言中可以通过runtime.GOMAXPROCS()函数设置当前程序并发时占用的CPU逻辑核心数。Go1.5版本之前，默认使用的是单核心执行。Go1.5版本之后，默认使用全部的CPU逻辑核心数。
在部分场景中，比如日志采集插件或者监控插件等非业务进程中，需要降低对业务进程的性能影响，需要开发者手动调整GOMAXPROCS的值。

  1. package main
  3. import (
  4.     "fmt"
  5.     "runtime"
  6.     "sync"
  7. )
  9. var wg sync.WaitGroup
  11. func func01() {
  12.     defer wg.Done()
  13.     for i := 0; i < 10; i++ {
  14.         fmt.Printf("函数:%v; value:%v\n", "func01", i)
  15.     }
  16. }
  18. func func02() {
  19.     defer wg.Done()
  20.     for i := 0; i < 10; i++ {
  21.         fmt.Printf("函数:%v; value:%v\n", "func02", i)
  22.     }
  23. }
  25. func main() {
  26.     wg.Add(2)
  27.     runtime.GOMAXPROCS(2)
  28.     go func01()
  29.     go func02()
  30.     wg.Wait()
  31. }

2. channel

单纯地将函数并发执行是没有意义的。函数与函数间需要交换数据才能体现并发执行函数的意义。虽然可以使用共享内存进行数据交换，但是共享内存在不同的goroutine中容易发生竞态问题。为了保证数据交换的正确性，必须使用互斥量对内存进行加锁，这种做法势必造成性能问题。
Go语言的并发模型是CSP（Communicating Sequential Processes），提倡通过通信共享内存而不是通过共享内存而实现通信。Go 语言中的通道（channel）是一种特殊的类型。通道像一个传送带或者队列，总是遵循先入先出（First In First Out）的规则，保证收发数据的顺序。每一个通道都是一个具体类型的导管，也就是声明channel的时候需要为其指定元素类型。

2.1. Channel基本使用

07-并发编程 - 图1

2.1.1. channel定义

channel 是引用类型，声明后需要使用make初始化分配内存才能使用。

  1. func main()  {
  2.     var ch1 chan int
  3.     fmt.Printf("type:%T;value:%v\n", ch1, ch1)
  4.     ch1 = make(chan int, 10)
  5.     fmt.Printf("type:%T;value:%v\n", ch1, ch1)
  6. }
  1. [root@heyingsheng channel]# go run main.go
  2. type:chan int;value:<nil>
  3. type:chan int;value:0xc0000b6000
  1. type student struct {
  2.     id   string
  3.     name string
  4.     age  uint8
  5. }
  7. func main() {
  8.     stuChan := make(chan student, 10)
  9.     fmt.Printf("type:%T;value:%v;cap:%d\n", stuChan, stuChan, cap(stuChan))
  10. }
  1. [root@heyingsheng channel]# go run main.go
  2. type:chan main.student;value:0xc0000b0120;cap:10

2.1.2. channel 简单使用

  1. package main
  3. import "fmt"
  5. type student struct {
  6.     name string
  7.     age  uint8
  8. }
  10. func main() {
  11.     ch1 := make(chan student, 2)
  12.     stu1 := student{
  13.         name: "张三",
  14.         age:  18,
  15.     }
  16.     ch1 <- stu1 // 向 ch1 写入数据
  17.     ch1 <- student{
  18.         name: "李四",
  19.         age:  20,
  20.     }
  22.     <-ch1             // 取出ch1中的值，并丢弃
  23.     stu1 = <-ch1      // 取出ch1中的值，并赋值给 stu1
  24.     close(ch1)        // 关闭通道 ch1
  25.     fmt.Println(stu1) // {李四 20}
  26. }

2.1.3. 遍历channel

遍历channel中的值，有两种方式：

  • 使用 for 配合 if 判断，使用 v, ok := ch 取值时，如果channel关闭则ok为 false，否则为 true
  • 使用 range 遍历ch，当ch关闭时，则自动退出 range 遍历

注意：如果注释掉下面代码的 line 17，则会出现死锁 fatal error: all goroutines are asleep - deadlock! 

  1. package main
  3. import (
  4.     "fmt"
  5.     "sync"
  6.     "time"
  7. )
  9. var wg sync.WaitGroup
  11. func f0(ch1 chan int)  {
  12.     defer wg.Done()
  13.     for i:=0;i<100;i++ {
  14.         ch1 <- i
  15.         time.Sleep(time.Millisecond * 50)
  16.     }
  17.     close(ch1)
  18. }
  20. func f1(ch1 chan int)  {
  21.     defer wg.Done()
  22.     for {
  23.         if v, ok := <- ch1; ok {
  24.             fmt.Printf("time:%v;func:f1;value:%v\n", time.Now().Format("15:04:05"), v)
  25.             continue
  26.         }
  27.         break
  28.     }
  29. }
  31. func f2(ch1 chan int)  {
  32.     defer wg.Done()
  33.     for v:=range ch1 {
  34.         fmt.Printf("time:%v;func:f2;value:%v\n", time.Now().Format("15:04:05"), v)
  35.     }
  36. }
  38. func main()  {
  39.     wg.Add(3)
  40.     num := make(chan int, 15)
  41.     go f0(num)
  42.     go f1(num)
  43.     go f2(num)
  44.     wg.Wait()
  45. }

2.1.4. 阻塞通道

上述案例中，每个通道指定了容量 cap，如果在make初始化时，不指定容量，则会通道cap为0，该通道称为阻塞通道或者无缓冲通道，实际开发中使用的很少。

  1. // 以下代码编译会通过，但是运行会报错: fatal error: all goroutines are asleep - deadlock!
  2. func main() {
  3.     ch0 := make(chan int)
  4.     ch0 <- 100         
  5.     fmt.Println(<-ch0) 
  6. }
  1. // 正确的代码如下
  2. func main() {
  3.     ch0 := make(chan int)
  4.     go func() {
  5.         ch0 <- 100
  6.     }()
  7.     fmt.Println(<- ch0)
  8. }

2.1.5. 单向通道

通常的channel都是可读可写，但是在部分场景中，需要对通道进行限制，如设置只读通道，或者只写通道，通常是作为函数的参数传递。

  1. package main
  3. import (
  4.     "fmt"
  5.     "sync"
  6. )
  8. type result struct {
  9.     num int
  10.     res int
  11. }
  13. var wg0 sync.WaitGroup
  15. // 生产 0-100 的数字，存入只写通道 cha
  16. func createNum(cha chan <- int)  {
  17.     defer wg0.Done()
  18.     for i:=0; i<100; i++ {
  19.         cha <- i
  20.     }
  21.     close(cha)
  22. }
  24. // 取出只读通道 in 中的数字，取平方后存入只写通道 out
  25. func squarer(in <- chan int, out chan <- result)  {
  26.     defer wg0.Done()
  27.     for i:=range in {
  28.         out <- result{
  29.             num: i,
  30.             res: i*i,
  31.         }
  32.     }
  33.     close(out)
  34. }
  36. // 从只读通道 in 中读取计算结果
  37. func printRes(in <- chan result)  {
  38.     defer wg0.Done()
  39.     for i := range in {
  40.         fmt.Printf("num:%d --> res:%d\n", i.num, i.res)
  41.     }
  42. }
  44. func main()  {
  45.     ch0 := make(chan int,10)
  46.     ch1 := make(chan result, 10)
  47.     wg0.Add(3)
  48.     go createNum(ch0)
  49.     go squarer(ch0, ch1)
  50.     go printRes(ch1)
  51.     wg0.Wait()
  52. }

2.2. goroutine 池

用于指定某类goroutine的数量，并不是总的goroutine的数量，类似于其它编程语言中的线程池。其目的在于避免突如其来的的高并发导致goroutine暴涨或者泄露。控制思路如下：

  1. 1.    设置一个任务channel，所有的任务都发往整个队列
  2. 2.    设置一个 for 循环，该循环的次数就是woker的数量，也就是控制的 goroutine 数量
  3. 3.    这些 worker 都监听同一个任务队列，当有新的任务来了，会被其中一个 worker 抢到
  1. package main
  3. import (
  4.     "fmt"
  5.     "math/rand"
  6.     "sync"
  7. )
  9. var wg1 sync.WaitGroup
  11. type student struct {
  12.     name string
  13.     age  int
  14. }
  16. // 随机产生n位字符串
  17. func getName(n int) string {
  18.     ret := make([]byte, n)  // [a-z] --> 97-->122 --> 5
  19.     for i:=0; i < n; i ++ {
  20.         ret[i] = byte(rand.Intn(26) + 97)
  21.     }
  22.     return string(ret)
  23. }
  25. // 随机产生学生信息
  26. func producer(n int, out chan <- *student)  {
  27.     defer wg1.Done()
  28.     for i:=0; i<n; i++ {
  29.         out <- &student{
  30.             name: getName(5),
  31.             age : rand.Intn(16) + 10,  // [10,25]
  32.          }
  33.     }
  34.     close(out)
  35. }
  37. // 工作函数，即消费者
  38. func worker(n int, job chan *student) {
  39.     defer wg1.Done()
  40.     for stu := range job {
  41.         ret := fmt.Sprintf("name:%v;age:%d", stu.name, stu.age)
  42.         fmt.Printf("worker-%d --> %v\n", n, ret)
  43.     }
  44. }
  46. func main()  {
  47.     students := make(chan *student, 10)
  48.     for i:=0; i<5 ; i ++ {
  49.         wg1.Add(1)
  50.         go worker(i, students)
  51.     }
  52.     go producer(15, students)
  53.     wg1.Wait()
  54. }

2.3. Select

select的使用类似于switch语句，它有一系列case分支和一个默认的分支。每个case会对应一个通道的通信（接收或发送）过程。select会一直等待，直到某个case的通信操作完成时，就会执行case分支对应的语句。除了空 select{} ，其它一般放在for循环中处理通道数据。

  • 可处理一个或多个channel的发送/接收操作
  • 如果多个case同时满足，select会随机选择一个
  • 对于没有case的select{}会一直等待，可用于阻塞main函数 ```go package main

import “fmt”

func main() { ch := make(chan int, 1) for i:=0;i<10;i++ { select { case x := <-ch:
fmt.Println(x) case ch <- i:
} } }

  1. ```
  2. [root@heyingsheng channel]# go run select.go
  3. 0
  4. 2
  5. 4
  6. 6
  7. 8

3. 并发安全

3.1. 互斥锁

在实际并发编程中，会涉及到多个goroutine对一个公共变量进行修改，比如同时来了A和B两个任务需要对同一个实例的字段进行更新，此时就出现了资源争抢，需要对这个实例进行加锁，避免数据异常。
互斥锁是一种常用的控制共享资源访问的方法，它能够保证同时只有一个goroutine可以访问共享资源。Go语言中使用sync包的Mutex类型来实现互斥锁。使用互斥锁能够保证同一时间有且只有一个goroutine进入临界区，其他的goroutine则在等待锁；当互斥锁释放后，等待的goroutine才可以获取锁进入临界区，多个goroutine同时等待一个锁时，唤醒的策略是随机的。

  1. type student struct {
  2.     name   string
  3.     metric int
  4. }
  6. func main() {
  7.     stu01 := student{
  8.         name:   "张三",
  9.         metric: 0,
  10.     }
  12.     var wg01 sync.WaitGroup
  13.     wg01.Add(1000)
  14.     for i := 0; i < 1000; i++ {
  15.         go func(stu *student) {
  16.             defer wg01.Done()
  17.             stu.metric++
  18.         }(&stu01)
  19.     }
  20.     wg01.Wait()
  21.     fmt.Println(stu01) // metric 数字不固定，因为不同的 goroutine 直接有竞争
  22. }
  1. type student struct {
  2.     lock   sync.Mutex   // 添加一把互斥锁
  3.     name   string
  4.     metric int
  5. }
  7. func main() {
  8.     stu01 := student{
  9.         name:   "张三",
  10.         metric: 0,
  11.     }
  13.     var wg01 sync.WaitGroup
  14.     wg01.Add(1000)
  15.     for i := 0; i < 1000; i++ {
  16.         go func(stu *student) {
  17.             defer wg01.Done()
  18.             stu.lock.Lock()
  19.             stu.metric++
  20.             stu.lock.Unlock()
  21.         }(&stu01)
  22.     }
  23.     wg01.Wait()
  24.     fmt.Printf("name:%v;metric:%v\n", stu01.name, stu01.metric)
  25. }

3.2. 读写锁

互斥锁是完全互斥的，但是有很多实际的场景下是读多写少的，当我们并发的去读取一个资源不涉及资源修改的时候是没有必要加锁的，这种场景下使用读写锁是更好的一种选择。读写锁在Go语言中使用sync包中的RWMutex类型。
读写锁分为两种：读锁和写锁。当一个goroutine获取读锁之后，其他的goroutine如果是获取读锁会继续获得锁，如果是获取写锁就会等待；当一个goroutine获取写锁之后，其他的goroutine无论是获取读锁还是写锁都会等待。
在读多写少的情况，且数据并不是始终存在内存中的情况下，使用读写锁比互斥锁有着更高的性能

  1. var wg02 sync.WaitGroup
  3. type Person struct {
  4.     rwlock sync.RWMutex
  5.     lock   sync.Mutex
  6.     name   string
  7.     metric int
  8. }
  10. func (p *Person) read(lockType string) {
  11.     defer wg02.Done()
  12.     if lockType == "rwlock" {
  13.         p.rwlock.RLock()
  14.         time.Sleep(time.Millisecond)  // 模拟 1ms I/O 阻塞
  15.         _ = p.name
  16.         _ = p.metric
  17.         p.rwlock.RUnlock()
  18.         return
  19.     }
  20.     p.lock.Lock()
  21.     time.Sleep(time.Millisecond)  // 模拟 1ms I/O 阻塞
  22.     _ = p.name
  23.     _ = p.metric
  24.     p.lock.Unlock()
  25. }
  27. func (p *Person) write(lockType string) {
  28.     wg02.Done()
  29.     if lockType == "rwlock" {
  30.         p.rwlock.Lock()
  31.         time.Sleep(time.Millisecond*2)  // 模拟 1ms I/O 阻塞
  32.         p.metric++
  33.         p.rwlock.Unlock()
  34.         return
  35.     }
  36.     p.lock.Lock()
  37.     time.Sleep(time.Millisecond*2)  // 模拟 1ms I/O 阻塞
  38.     p.metric++
  39.     p.lock.Unlock()
  40. }
  42. func main() {
  43.     p0 := &Person{name: "张三"}
  44.     // 开始测试读写锁性能
  45.     start := time.Now()
  46.     wg02.Add(11000)
  47.     for i := 0; i < 1000; i++ {
  48.         go p0.write("rwlock")  // 写1000次
  49.     }
  50.     for i := 0; i < 10000; i++ {
  51.         go p0.read("rwlock")   // 读10000次
  52.     }
  53.     wg02.Wait()
  54.     fmt.Printf("读写锁耗时:%v\n", time.Now().Sub(start))
  55.     // 开始测试互斥锁性能
  56.     start = time.Now()
  57.     wg02.Add(11000)
  58.     for i := 0; i < 1000; i++ {
  59.         go p0.write("lock")  // 写1000次
  60.     }
  61.     for i := 0; i < 10000; i++ {
  62.         go p0.read("lock")   // 读10000次
  63.     }
  64.     wg02.Wait()
  65.     fmt.Printf("互斥锁耗时:%v\n", time.Now().Sub(start))
  66. }
  1. [root@heyingsheng locks]# go run rwlock.go
  2. 读写锁耗时:10.9732ms
  3. 互斥锁耗时:14.5726255s

3.3. sync.Once

在编程的很多场景下我们需要确保某些操作在高并发的场景下只执行一次，例如只加载一次配置文件、只关闭一次通道等，为了避免多次操作造成报错或者不必要的资源浪费，Go提供了 sync.Once 结构体，该结构体只有一个方法 func (o *Once) Do(f func()) {} ，该方法接收一个没有参数和返回值的函数。

3.3.1. 加载配置文件

  1. var icons map[string]image.Image
  3. func loadIcons() {
  4.     icons = map[string]image.Image{
  5.         "left":  loadIcon("left.png"),
  6.         "up":    loadIcon("up.png"),
  7.         "right": loadIcon("right.png"),
  8.         "down":  loadIcon("down.png"),
  9.     }
  10. }
  12. // Icon 被多个goroutine调用时不是并发安全的
  13. func Icon(name string) image.Image {
  14.     if icons == nil {
  15.         loadIcons()
  16.     }
  17.     return icons[name]
  18. }

多个goroutine并发调用Icon函数时不是并发安全的，现代的编译器和CPU可能会在保证每个goroutine都满足串行一致的基础上自由地重排访问内存的顺序。loadIcons函数可能会被重排为以下结果：

  1. func loadIcons() {
  2.     icons = make(map[string]image.Image)
  3.     icons["left"] = loadIcon("left.png")
  4.     icons["up"] = loadIcon("up.png")
  5.     icons["right"] = loadIcon("right.png")
  6.     icons["down"] = loadIcon("down.png")
  7. }

在这种情况下就会出现即使判断了icons不是nil也不意味着变量初始化完成了。考虑到这种情况，我们能想到的办法就是添加互斥锁，保证初始化icons的时候不会被其他的goroutine操作，但是这样做又会引发性能问题。使用sync.Once改造的示例代码如下：

  1. var icons map[string]image.Image
  2. var loadIconsOnce sync.Once
  3. func loadIcons() {
  4.     icons = map[string]image.Image{
  5.         "left":  loadIcon("left.png"),
  6.         "up":    loadIcon("up.png"),
  7.         "right": loadIcon("right.png"),
  8.         "down":  loadIcon("down.png"),
  9.     }
  10. }
  11. // Icon 是并发安全的
  12. func Icon(name string) image.Image {
  13.     loadIconsOnce.Do(loadIcons)
  14.     return icons[name]
  15. }

3.3.2. 并发安全的单例模式

下面是借助sync.Once实现的并发安全的单例模式：

  1. package singleton
  2. import (
  3.     "sync"
  4. )
  5. type singleton struct {}
  6. var instance *singleton
  7. var once sync.Once
  8. func GetInstance() *singleton {
  9.     once.Do(func() {
  10.         instance = &singleton{}
  11.     })
  12.     return instance
  13. }

sync.Once其实内部包含一个互斥锁和一个布尔值，互斥锁保证布尔值和数据的安全，而布尔值用来记录初始化是否完成。这样设计就能保证初始化操作的时候是并发安全的并且初始化操作也不会被执行多次。

3.4. sync.Map

在Go语言中，Map并不是并发安全的，如下的案例中，就可能出现报错: fatal error: concurrent map writes ，当然可以在 setMap() 和 getMap 上加互斥锁或者读写锁来实现数据安全，避免报错，但是Go本身提供了一个带锁的map和对对应的方法，可以开箱即用。

  1. var wg3 sync.WaitGroup
  3. func setMap(m map[int]string, key int, value string) {
  4.     defer wg3.Done()
  5.     wg3.Add(1)
  6.     m[key] = value
  7. }
  9. func main() {
  10.     m0 := make(map[int]string, 10)
  11.     for i := 0; i < 10000; i++ {
  12.         go setMap(m0, i, strconv.Itoa(i))
  13.     }
  14.     wg3.Wait()
  15. }
  1. # 手动添加互斥锁
  2. var wg3 sync.WaitGroup
  3. var lock sync.Mutex
  5. func setMap(m map[int]string, key int, value string) {
  6.     defer wg3.Done()
  7.     wg3.Add(1)
  8.     lock.Lock()
  9.     m[key] = value
  10.     lock.Unlock()
  11. }
  13. func main() {
  14.     m0 := make(map[int]string, 10)
  15.     for i := 0; i < 10000; i++ {
  16.         go setMap(m0, i, strconv.Itoa(i))
  17.     }
  18.     wg3.Wait()
  19. }
  1. var wg3 sync.WaitGroup
  3. func main() {
  4.     m0 := sync.Map{}
  5.     wg3.Add(10000)
  6.     for i := 0; i < 10000; i++ {
  7.         go func(i int) {
  8.             defer wg3.Done()
  9.             m0.Store(i, strconv.Itoa(i))
  10.         }(i)
  11.     }
  12.     wg3.Wait()
  13. }
  1. 1.    func (m *Map) Store(key, value interface{}) {}  
  2.         存数据
  4. 2.    func (m *Map) Load(key interface{}) (value interface{}, ok bool) {}
  5.         取数据
  7. 3.    func (m *Map) LoadOrStore(key, value interface{}) (actual interface{}, loaded bool) {}
  8.         存在则返回值，不存在存储并返回
  10. 4.    func (m *Map) Delete(key interface{}) {}
  11.         删除值

4. Context包

在Go语言中，常常用一个goroutine来处理一个请求，而这个请求往往需要继续开子goroutine来处理其中部分请求，达能请求超时需要关闭这个goroutine和它的子goroutine，这种需求可以使用全局变量或者管道来实现。

4.1. 通知子goroutine关闭

4.1.1. 全局变量方式

  1. var wg01 sync.WaitGroup
  2. var flag bool
  4. func func01() {
  5.     defer wg01.Done()
  6.     for ! flag {
  7.         fmt.Println(time.Now().Format("2006-01-02 15:04:05"))
  8.         time.Sleep(time.Second)
  9.     }
  10. }
  12. func main() {
  13.     wg01.Add(1)
  14.     go func01()
  15.     time.Sleep(time.Second * 5)
  16.     flag = true
  17.     wg01.Wait()
  18. }

4.1.2. 管道方式

  1. var wg02 sync.WaitGroup
  2. var exitChan = make(chan struct{}) // 空结构体或者bool都行
  4. func func02() {
  5.     defer wg02.Done()
  6.     for {
  7.         fmt.Println(time.Now().Format("2006-01-02 15:04:05"))
  8.         time.Sleep(time.Second)
  9.         select {
  10.         case <-exitChan:
  11.             fmt.Println("exit")
  12.             return
  13.         default:
  15.         }
  16.     }
  17. }
  19. func main() {
  20.     wg02.Add(1)
  21.     go func02()
  22.     time.Sleep(time.Second * 5)
  23.     exitChan <- struct{}{}
  24.     wg02.Wait()
  25. }

4.1.3. context方式

  1. var wg03 sync.WaitGroup
  3. func func03(ctx context.Context) {
  4.     defer wg03.Done()
  5.     for {
  6.         fmt.Println(time.Now().Format("2006-01-02 15:04:05"))
  7.         time.Sleep(time.Second)
  8.         select {
  9.         case <-ctx.Done():
  10.             return
  11.         default:
  13.         }
  14.     }
  15. }
  17. func main() {
  18.     ctx, cancel := context.WithCancel(context.Background())
  19.     wg03.Add(1)
  20.     go func03(ctx)
  21.     time.Sleep(time.Second * 5)
  22.     cancel()
  23.     wg03.Wait()
  24. }

4.1.4. 多级goroutine情况

如果存在多级的goroutine调用，把ctx继续传递下去就行，ctx内部实现了多级goroutine的安全退出，不会因为其中一个goroutine取了管道中的值，而导致其它goroutine无法取值。

  1. var wg04 sync.WaitGroup
  3. func childWorker(ctx context.Context)  {
  4.     for {
  5.         fmt.Println("child", time.Now().Format("2006-01-02 15:04:05"))
  6.         time.Sleep(time.Second)
  7.         select {
  8.         case <-ctx.Done():
  9.             return
  10.         default:
  11.         }
  12.     }
  13. }
  15. func func04(ctx context.Context) {
  16.     defer wg04.Done()
  17.     go childWorker(ctx)
  18.     for {
  19.         fmt.Println("parent", time.Now().Format("2006-01-02 15:04:05"))
  20.         time.Sleep(time.Second)
  21.         select {
  22.         case <-ctx.Done():
  23.             return
  24.         default:
  25.         }
  26.     }
  27. }
  29. func main() {
  30.     ctx, cancel := context.WithCancel(context.Background())
  31.     wg04.Add(1)
  32.     go func04(ctx)
  33.     time.Sleep(time.Second * 5)
  34.     cancel()
  35.     wg04.Wait()
  36. }

4.1.5. goroutine kill 问题

golang 并不能停止一个正在运行的 goroutine，上面提到的通过goroutine关闭，并不是最底层处理阻塞任务的那个gorountine，比如下面的这个案例：
Run() 方法可以提前结束退出，但是 exec() 方法需要等待 ansible执行完毕。无法在 Run()中强制结束 exec() 方法。为了避免exec()超时后，仍然修改执行结果，增加了 timeout 这个字段，当该字段为 true 时，主动放弃exec() 的执行结果。

  1. type Ansible struct {
  2.     ticker  <-chan time.Time // 定时器，用于超时处理
  3.     result  string           // ansible运行结果
  4.     timeout bool             // 超时状态
  5.     done    chan bool        // 完成任务的标志位
  6. }
  8. func NewAnsible(duration time.Duration) *Ansible {
  9.     return &Ansible{
  10.         ticker: time.After(duration),
  11.         done:   make(chan bool),
  12.     }
  13. }
  15. // 执行ansible 任务，并更新实例的状态，比如将 result 写入数据等
  16. func (a *Ansible) Run() {
  17.     go a.exec()
  18.     for {
  19.         select {
  20.         case <-a.ticker:
  21.             a.timeout = true
  22.             a.result = "task failed, error: timeout"
  23.             return
  24.         case <-a.done:
  25.             a.result = "task success, ..... "
  26.             return
  27.         }
  28.     }
  29. }
  31. // 正在执行ansible命令的函数，该函数是阻塞的，会等待 ansible 执行完毕
  32. func (a *Ansible) exec() {
  33.     time.Sleep(time.Second * 2) // 模拟ansible 执行，此处为了快速出结果，仅休眠2秒
  34.     if a.timeout {
  35.         return
  36.     }
  37.     a.done <- true
  38. }
  40. func main() {
  41.     task01 := NewAnsible(time.Second * 1)
  42.     task01.Run()
  43.     fmt.Printf("task01 resutl:%s\n", task01.result)
  45.     task02 := NewAnsible(time.Second * 3)
  46.     task02.Run()
  47.     fmt.Printf("task02 resutl:%s\n", task02.result)
  48. }
  1. [root@duduniao go_learn]# go run day23/groutine/test01.go
  2. task01 resutl:task failed, error: timeout
  3. task02 resutl:task success, .....

4.2. Context基础

Go1.7加入了一个新的标准库context，它定义了Context类型，专门用来简化 对于处理单个请求的多个 goroutine 之间与请求域的数据、取消信号、截止时间等相关操作，这些操作可能涉及多个 API 调用。

  1. 1.    type Context interface 
  2.         type Context interface {
  3.             Deadline() (deadline time.Time, ok bool)
  4.         Done() <-chan struct{}
  5.         Err() error
  6.         Value(key interface{}) interface{}
  7.     }
  8.     定义接口
  10. 2.    func Background() Context
  11.         返回一个不会被取消，没有值，也不会过期的Context类型，一般作为顶层节点的 Context，用于mian函数、init函数等。
  13. 3.    func TODO() Context
  14.         保留项，目前官方还没有定义具体的用途
  16. 4.    type CancelFunc func()
  17.         定义一个没有参数和返回值的函数类型: CancelFunc
  18.     调用后，释放 Context 上下文资源。
  20. 5.    func WithCancel(parent Context) (ctx Context, cancel CancelFunc)
  21.         返回带有新Done通道的父节点的副本，从 ctx.Done() 中获取空结构体，用于goroutine之间的通信
  23. 6.    func WithDeadline(parent Context, deadline time.Time) (Context, CancelFunc)
  24.         带过期时间的context
  26. 7.    func WithTimeout(parent Context, timeout time.Duration) (Context, CancelFunc)
  27.         带过期时间的context
  29. 8.    func WithValue(parent Context, key, val interface{}) Context
  30.         可以在不同的API和进程之间传递数据，一般用于追踪一个请求的链路，比如用 traceID 作为唯一值，在不同API之间转递
  31.     仅对API和进程间传递请求域的数据使用上下文值，而不是使用它来传递可选参数给函数。
  32.     所提供的键必须是可比较的，并且不应该是string类型或任何其他内置类型，以避免使用上下文在包之间发生冲突。
  33.     WithValue的用户应该为键定义自己的类型。

4.3. 案例

4.3.1. timeout

  1. func main()  {
  2.     ctx, cancel := context.WithTimeout(context.Background(), time.Second * 5)
  3.     defer cancel()
  5.     for {
  6.         fmt.Println(time.Now().Format("15:04:05"))
  7.         time.Sleep(time.Second)
  8.         select {
  9.         case <- ctx.Done():
  10.             return
  11.         default:
  12.         }
  13.     }
  14. }

4.3.2. deadline

  1. var wg01 sync.WaitGroup
  3. func func01(ctx context.Context)  {
  4.     defer wg01.Done()
  5.     for {
  6.         fmt.Println(time.Now().Format("15:04:05"))
  7.         time.Sleep(time.Second)
  8.         select {
  9.         case <- ctx.Done():
  10.             return
  11.         default:
  12.         }
  13.     }
  14. }
  16. func main()  {
  17.     ctx, cancel := context.WithDeadline(context.Background(), time.Now().Add(time.Second * 5))
  18.     defer cancel()
  20.     wg01.Add(2)
  21.     go func01(ctx)
  22.     go func01(ctx)
  23.     wg01.Wait()
  24. }

4.3.3. value

  1. type TraceCode string
  3. var wg sync.WaitGroup
  5. func worker(ctx context.Context) {
  6.     key := TraceCode("TRACE_CODE")
  7.     traceCode, ok := ctx.Value(key).(string) // 在子goroutine中获取trace code
  8.     if !ok {
  9.         fmt.Println("invalid trace code")
  10.     }
  11. LOOP:
  12.     for {
  13.         fmt.Printf("worker, trace code:%s\n", traceCode)
  14.         time.Sleep(time.Millisecond * 10) // 假设正常连接数据库耗时10毫秒
  15.         select {
  16.         case <-ctx.Done(): // 50毫秒后自动调用
  17.             break LOOP
  18.         default:
  19.         }
  20.     }
  21.     fmt.Println("worker done!")
  22.     wg.Done()
  23. }
  25. func main() {
  26.     // 设置一个50毫秒的超时
  27.     ctx, cancel := context.WithTimeout(context.Background(), time.Millisecond*50)
  28.     // 在系统的入口中设置trace code传递给后续启动的goroutine实现日志数据聚合
  29.     ctx = context.WithValue(ctx, TraceCode("TRACE_CODE"), "12512312234")
  30.     wg.Add(1)
  31.     go worker(ctx)
  32.     time.Sleep(time.Second * 5)
  33.     cancel() // 通知子goroutine结束
  34.     wg.Wait()
  35.     fmt.Println("over")
  36. }

5. 并发模式

5.1. 生产者和消费者模型

并发编程中最常见的例子就是生产者消费者模式，该模式主要通过平衡生产线程和消费线程的工作能力来提高程序的整体处理数据的速度。简单地说，就是生产者生产一些数据，然后放到成果队列中，同时消费者从成果队列中来取这些数据。这样就让生产消费变成了异步的两个过程。当成果队列中没有数据时，消费者就进入饥饿的等待中；而当成果队列中数据已满时，生产者则面临因产品挤压导致CPU被剥夺的下岗问题。

  1. package main
  3. import (
  4.     "fmt"
  5.     "time"
  6. )
  8. var (
  9.     ch = make(chan int, 20)
  10. )
  12. func producer(num int, queue chan<- int) {
  13.     for i := 0; ; i++ {
  14.         queue <- num * i
  15.         time.Sleep(100 * time.Millisecond)
  16.     }
  17. }
  19. func consumer(name string, queue <-chan int) {
  20.     for v := range queue {
  21.         fmt.Printf("consumer:%s;res:%d\n", name, v)
  22.     }
  23. }
  25. func main() {
  26.     go producer(5, ch)
  27.     go producer(7, ch)
  28.     go consumer("c1", ch)
  29.     go consumer("c2", ch)
  30.     select {}
  31. }

5.2. 发布订阅模型

发布订阅者模型(publish-and-subscribe)，一般是多个服务应用之间，通过中间件消息队列，如Kafka实现。消费者和生产者模型中，是通过队列进行异步的消息传递的。而发布订阅模型则引入了Topic概念，发布者将消息发布到一个主题，只有订阅了该主题的程序才能收到和消费这条消息。

5.3. 赢者为王

并发编程在大部分情况下，目的是为了避免阻塞，提高程序运行的性能。但是在部分场景中，如果一个问题由多个处理方法，但是不同情况下不同处理方式耗费的时间不同，此时可以让多个方式一起运行，仅取第一个完成的结果。

  1. func main() {
  2.     ch := make(chan string, 32)
  3.     go func() {
  4.         ch <- searchByBing("golang")
  5.     }()
  6.     go func() {
  7.         ch <- searchByGoogle("golang")
  8.     }()
  9.     go func() {
  10.         ch <- searchByBaidu("golang")
  11.     }()
  12.     fmt.Println(<-ch)
  13. }