Go的并发

Goroutine

Go实现了goroutine这一由Go运行时（runtime）负责调度的、轻量的用户级线程，为并发程序设计提供原生支持

优势

1. 资源占用小，每个goroutine的初始栈大小仅为2k
2. 由Go运行时而不是操作系统调度，goroutine上下文切换在用户层完成，开销更小
3. 在语言层面而不是通过标准库提供。goroutine由go关键字创建，一退出就会被回收或销毁，开发体验更佳
4. 语言内置channel作为goroutine间通信原语，为并发设计提供强大支撑

基本用法

创建

Go 语言通过go关键字+函数/方法的方式创建一个 goroutine。创建后，新 goroutine 将拥有独立的代码执行流，并与创建它的 goroutine 一起被 Go 运行时调度

退出

多数情况下，我们不需要考虑对 goroutine 的退出进行控制：goroutine 的执行函数的返回，就意味着 goroutine 退出

goroutine间的通信

CSP（Communicating Sequential Processes，通信顺序进程）并发模型

Tony Hoare 的 CSP 模型旨在简化并发程序的编写，让并发程序的编写与编写顺序程序一样简单

Tony Hoare 认为输入输出应该是基本的编程原语，数据处理逻辑（也就是 CSP 中的 P）只需调用输入原语获取数据，顺序地处理数据，并将结果数据通过输出原语输出就可以了

在 Tony Hoare 眼中，一个符合 CSP 模型的并发程序应该是一组通过输入输出原语连接起来的 P 的集合

Goroutine调度器

任务

将Goroutine按照一定算法放到不同的操作系统线程中去执行

Goroutine调度器模型与演化过程

G-M模型

每个Goroutine对应于运行中的一个抽象结构：G(Goroutine)；被视作“物理CPU”的操作系统线程，则被抽象为另外一个结构：M(machine)

调度器的工作就是将G调度到M上去运行

不足

限制了Go并发程序的伸缩性，尤其是那些有高吞吐或并行计算需求的服务程序

1. 单一全局互斥锁（Sched.Lock）和集中状态存储的存在，导致所有Goroutine相关操作，比如创建、重新调度等，都要上锁
2. Goroutine传递问题：M经常在M之间传递“可运行”的Goroutine，这导致调度延迟增大，也增加了额外的性能损耗
3. 每个M都做内存缓存，导致内存占用过高，数据局部性较差
4. 由于系统调用（syscall）而形成的频繁的工作线程阻塞和解除阻塞，导致额外的性能损耗

G-P-M调度模型

增加一个间接的中间层

P 是一个“逻辑 Proccessor”，每个 G（Goroutine）要想真正运行起来，首先需要被分配一个 P，也就是进入到 P 的本地运行队列（local runq）中。对于 G 来说，P 就是运行它的“CPU”，可以说：在 G 的眼里只有 P。但从 Go 调度器的视角来看，真正的“CPU”是 M，只有将 P 和 M 绑定，才能让 P 的 runq 中的 G 真正运行起来

问题

不支持抢占式调度

G

代表Goroutine，存储了Goroutine的执行栈信息、Goroutine状态以及Goroutine的任务函数等，而且G对象是可以重用的

P

代表逻辑processor，P的数量决定了系统内最大可并行的G的数量，P的最大作用还是其拥有的各种G对象队列、链表、一些缓存和状态

M

M代表着真正的执行计算资源。在绑定有效的P之后，进入一个调度循环，而调度循环的机制大致是从P的本地运行队列以及全局队列中获取G，切换到G的执行栈上并执行G的函数，调用goexit做清理工作并回到M，如此反复。M并不保留G状态，这是G可以跨M调度的基础

基于协作的“抢占式”调度

原理：Go 编译器在每个函数或方法的入口处加上了一段额外的代码 (runtime.morestack_noctxt)，让运行时有机会在这段代码中检查是否需要执行抢占调度

这种解决方案只能说局部解决了“饿死”问题，只在有函数调用的地方才能插入“抢占”代码（埋点），对于没有函数调用而是纯算法循环计算的 G，Go 调度器依然无法抢占

对非协作的抢占式调度

种抢占式调度是基于系统信号的，也就是通过向线程发送信号的方式来抢占正在运行的 Goroutine

channel

基本语法

创建

var ch chan int //声明一个元素为int类型的channel类型变量ch
//为channel类型变量赋初值的唯一方法就是使用make这个Go预定义的函数
ch1 := make(chan int) //声明一个无缓冲channel
ch2 := make(chan int,5) //声明一个带缓冲channel，缓冲区长度为5

//使用操作符<-，我们还可以声明只发送 channel 类型（send-only）和只接收 channel 类型（recv-only）
ch1 := make(chan<- int, 1) // 只发送channel类型
ch2 := make(<-chan int, 1) // 只接收channel类型
<-ch1 // invalid operation: <-ch1 (receive from send-only type chan<- int)
ch2 <- 13 // invalid operation: ch2 <- 13 (send to receive-only type <-chan int)
1
2
3
4
5
6
7
8
9
10

发送与接收

Go提供了<-操作符用于对channel类型变量进行发送和接收操作

ch1 <- 13    // 将整型字面值13发送到无缓冲channel类型变量ch1中
n := <- ch1  // 从无缓冲channel类型变量ch1中接收一个整型值存储到整型变量n中
ch2 <- 17    // 将整型字面值17发送到带缓冲channel类型变量ch2中
m := <- ch2  // 从带缓冲channel类型变量ch2中接收一个整型值存储到整型变量m中
1
2
3
4

channel 是用于 Goroutine 间通信的，所以绝大多数对 channel 的读写都被分别放在了不同的 Goroutine 中

无缓冲channel类型变量

发送和接收

对同一个无缓冲 channel，只有对它进行接收操作的 Goroutine 和对它进行发送操作的 Goroutine 都存在的情况下，通信才能得以进行，否则单方面的操作会让对应的 Goroutine 陷入挂起状态

对无缓冲 channel 类型的发送与接收操作，一定要放在两个不同的 Goroutine 中进行，否则会导致 deadlock

惯用法

用作信号传递

1对1或者1对n

用于替代锁机制

带缓冲channel类型变量

发送和接收

对带缓冲 channel 的发送操作在缓冲区未满、接收操作在缓冲区非空的情况下是异步的（发送或接收不需要阻塞等待）

对一个带缓冲 channel 来说，在缓冲区未满的情况下，对它进行发送操作的 Goroutine 并不会阻塞挂起；在缓冲区有数据的情况下，对它进行接收操作的 Goroutine 也不会阻塞挂起

但当缓冲区满了的情况下，对它进行发送操作的 Goroutine 就会阻塞挂起；当缓冲区为空的情况下，对它进行接收操作的 Goroutine 也会阻塞挂起

惯用法

用作消息队列

用作计数信号量（counting semaphore）

len(channel)的应用

针对 channel ch 的类型不同，len(ch) 有如下两种语义：

1. 当 ch 为无缓冲 channel 时，len(ch) 总是返回 0
2. 当 ch 为带缓冲 channel 时，len(ch) 返回当前 channel ch 中尚未被读取的元素个数

nil channel的妙用

nil channel 有一个特性，那就是对 nil channel 的读写都会发生阻塞

关闭

在发送完数据后，调用 Go 内置的 close 函数关闭了 channel。channel 关闭后，所有等待从这个 channel 接收数据的操作都将返回

close(chan) //关闭channel

n := <- ch      // 当ch被关闭后，n将被赋值为ch元素类型的零值
m, ok := <-ch   // 当ch被关闭后，m将被赋值为ch元素类型的零值, ok值为false
for v := range ch { // 当ch被关闭后，for range循环结束    
    ... ...
}
//通过“comma, ok”惯用法或 for range 语句，我们可以准确地判定 channel 是否被关闭。而单纯采用n := <-ch形式的语句，我们就无法判定从 ch 返回的元素类型零值，究竟是不是因为 channel 被关闭后才返回的
1
2
3
4
5
6
7
8

channel使用惯例：发送端负责关闭 channel

与select使用结合的惯用法

利用default分支避免阻塞

实现超时机制

//下面示例代码实现了一次具有 30s 超时的 select
func worker() {
  select {
  case <-c:
       // ... do some stuff
  case <-time.After(30 *time.Second):
      return
  }
}
//在应用带有超时机制的 select 时，我们要特别注意 timer 使用后的释放，尤其在大量创建 timer 的时候
1
2
3
4
5
6
7
8
9
10

实现心跳机制

//结合 time 包的 Ticker，我们可以实现带有心跳机制的 select。这种机制让我们可以在监听 channel 的同时，执行一些周期性的任务
func worker() {
  heartbeat := time.NewTicker(30 * time.Second)
  defer heartbeat.Stop()
  for {
    select {
    case <-c:
      // ... do some stuff
    case <- heartbeat.C:
      //... do heartbeat stuff
    }
  }
}
1
2
3
4
5
6
7
8
9
10
11
12
13

select

通过 select，我们可以同时在多个 channel 上进行发送 / 接收操作

select {
case x := <-ch1:     // 从channel ch1接收数据
  ... ...

case y, ok := <-ch2: // 从channel ch2接收数据，并根据ok值判断ch2是否已经关闭
  ... ...

case ch3 <- z:       // 将z值发送到channel ch3中:
  ... ...

default:             // 当上面case中的channel通信均无法实施时，执行该默认分支
}
1
2
3
4
5
6
7
8
9
10
11
12