简析go语言并发模式
go语言实现了CSP并发模型：输入/输出为基本编程原语，数据处理逻辑（CSP模型中的P）仅需调用输入原语获取数据，顺序处理数据并将结果通过输出原语输出。Go引入了goroutine（P）之间的通信原语channel，goroutine可以从channel获取输入数据，再将处理后的数据结果通过channel输出。通过channel和goroutine组合连接在一起，使并发程序设计更为清晰，无须再为传统共享内存并发模型中的问题花费心智。引入select用于应对多路输入/输出，可以让goroutine同时协调处理多个channel操作。
Go语言也支持传统的基于共享内存的并发模型，并提供了基本的低级同步原语：主要是sync包中的互斥锁、条件变量、读写锁、原子操作等。但Go始终推荐以CSP模型风格构建并发程序，CSP模型风格go并发程序的并发模式大致分以下几种：
1.创建模式：使用go关键字+函数/方法创建一个goroutine，并返回一个channel类型变量的函数。这是Go中最常见的goroutine创建模式。
type T struct{}
func spawn(f func()) chan T {
c := make(chan T)
go func() {
f()
c <- T{}
}()
return c
}
func main() {
c := spawn(func())
<-c
}
spawn函数创建新的goroutine与调用spawn函数的goroutine之间通过一个channel建立联系，即两个goroutine通过channel进行通信。channel可以像变量一样被初始化、传递和赋值。
2.退出模式：goroutine的使用代价很低，官方推荐使用goroutine。在大多数情况下，我们无需考虑对goroutine的退出进行控制，goroutine的执行函数返回，就意味着goroutine的退出。但一些常驻后台服务程序可能会对goroutine有着优雅退出的要求。goroutine的退出模式大致分以下几种：
1）分离模式：分离模式是使用最广泛的goroutine退出模式。对于分离的goroutine，创建它的goroutine不需要关心它的退出，这类goroutine在创建后即与其创建者彻底分离，其生命周期与其执行的主函数相关，函数返回即goroutine退出。这类goroutine有如下两个常见用途。
a.一次性任务：新创建的goroutine用来执行一个简单任务，执行后退出。
import (
"fmt"
"time"
)
func worker(args ...interface{}) {
if len(args) == 0 {
return
}
interval, ok := args[0].(int)
if !ok {
return
}
time.Sleep(time.Second * time.Duration(interval))
}
func spawn(f func(args ...interface{}), args ...interface{}) {
go func() {
f(args...)
}()
}
func main() {
spawn(worker, 5)
}
b.常驻任务：常驻后台执行一些特定任务，如监视（monitor）、观察(watch)等。其实现通常采用for {...}或者for {select{...}}代码段形式，并多以定时器（timer）或事件（event）驱动执行。
import (
"fmt"
"time"
)
func worker(args ...interface{}) {
if len(args) == 0 {
fmt.Println("invalid args")
return
}
interval, ok := args[0].(int)
if !ok {
println("invalid interval args")
return
}
time.Sleep(time.Second * time.Duration(interval))
}
func spawn(f func(args ...interface{}), args ...interface{}) {
for {
go func() {
f(args...)
}()
}
}
func main() {
spawn(worker, 5)
}
2）join模式：goroutine的创建者等待新创建的goroutine结束。
a.等待一个goroutine退出
func worker(args ...interface{}) {
if len(args) == 0 {
return
}
interval, _ := args[0].(int)
time.Sleep(time.Second * time.Duration(interval))
}
func spawn(f func(args ...interface{}), args ...interface{}) chan struct{} {
c := make(chan struct{})
go func() {
f(args...)
c <- struct{}{}
time.Sleep(time.Second * 3) // 此处是为了说明向channel发送数据后，main goroutine是立刻感知还是等spawn函数返回后才感知
fmt.Println("worker go routine end")
}()
return c
}
func main() { // main goroutine
c := spawn(worker, 5) // 调用spawn，spawn函数创建了一个新的goroutine
fmt.Println("main goroutine start")
<-c
fmt.Println("main goroutine end")
time.Sleep(time.Second * 5) // 等待新的goroutine执行完成
}
$ go run wait-one-goroutine-demo.go
main goroutine start
main goroutine end
worker go routine end // 从结果可以看出，向channel发送信号是在新的goroutine退出前，新的goroutine发送信号后main goroutine立刻感知到接触阻塞执行了main函数体剩余的代码，而不是等待新的goroutine执行完毕main goroutine才感知到。
spawn函数使用典型的goroutine创建模式创建了一个goroutine，main goroutine作为创建者通过spawn函数返回的channel与新goroutine建立联系，这个channel的用途就是在两个goroutine之间建立退出事件的”信号“通信机制。main goroutine在创建完新的goroutine后便在该channel上阻塞等待，直到新的goroutine退出前向该channel发送一个信号。
b.获取goroutine的退出状态：新的goroutine创建者不仅要等待goroutine的退出，还要精准获取其结束状态。
import (
"errors"
"fmt"
"time"
)
var OK = errors.New("ok")
func worker(args ...interface{}) error {
if len(args) == 0 {
return errors.New("invalid args")
}
interval, ok := args[0].(int)
if !ok {
return errors.New("invalid interval args")
}
time.Sleep(time.Second * time.Duration(interval))
return OK
}
func spawn(f func(args ...interface{}) error, args ...interface{}) chan error {
c := make(chan error)
go func() {
c <- f(args...)
}()
return c
}
func main() {
done := spawn(worker, 5)
fmt.Println("worker1 start")
err := <-done
fmt.Println("worker1 done:", err)
done = spawn(worker, "a")
fmt.Println("worker2 start")
err = <-done
fmt.Println("worker2 done:", err)
}
$ go run wait-and-get-exit-code-one-goroutine-demo.go
worker1 start
worker1 done: ok
worker2 start
worker2 done: invalid interval args
上例中将channel承载的类型由struct{}改为error，这样channel承载的信息不仅仅是一个信号，还携带了有价值的信息：新goroutine的结束状态。
c.等待多个goroutine退出：在某些场景中，goroutine创建者可能需要等待多个goroutine退出。可以通过Go语言提供的sync.WaitGroup实现等待多个goroutine退出
import (
"fmt"
"sync"
"time"
)
func worker(args ...interface{}) {
if len(args) == 0 {
return
}
interval, ok := args[0].(int)
if !ok {
return
}
time.Sleep(time.Second * time.Duration(interval))
}
func spawnGroup(n int, f func(args ...interface{}), args ...interface{}) chan struct{} {
c := make(chan struct{})
var wg sync.WaitGroup
for i := 0; i < n; i++ {
wg.Add(1)
go func(i int) {
name := fmt.Sprintf("worker-%d", i)
f(args...)
println(name, "done")
wg.Done()
}(i)
}
go func() {
wg.Wait()
c <- struct{}{}
}()
return c
}
func main() {
fmt.Println("spawn a group of workers")
done := spawnGroup(5, worker, 3)
<-done
fmt.Println("group workers done")
}
$ go run wait-multiple-goroutine-exit-demo.go
spawn a group of workers
worker-1 done
worker-3 done
worker-4 done
worker-0 done
worker-2 done
group workers done
上例中通过sync.WaitGroup，spawnGorup每创建一个goroutine都会调用wg.Add(1)，新创建的goroutine会在退出前调用wg.Done。在spwanGroup中还创建了一个用于监视的goroutine，该goroutine调用wg.Wait方法等待所有goroutine退出。在所有新创建的goroutine退出后，wg.Wait方法返回，并向c这个channel中写入一个信号，这是main goroutine会从阻塞在channel上的状态中恢复，继续往下执行。
d.超时等待机制：有时不需要无限等待所有新创建的goroutine退出，而是仅等待一段合理的时间。如果在这段时间内goroutine没有退出，则创建者会继续往下执行或者主动退出。
func main() {
timer := time.NewTimer(time.Second * 5)
done := spawnGroup(5, worker, 10)
defer timer.Stop()
fmt.Println("spawn a group of workers")
select {
case <-timer.C:
println("wait group workers exit timeout")
case <-done:
println("group workers done")
}
}
$ go run wait-multiple-goroutine-exit-with-timeout-demo.go
spawn a group of workers
wait group workers exit timeout
上例中通过创建一个定时器（time.Timer）设置超时等待时间，通过select监听timer.C和done两个channel，哪个先返回就执行哪个case分支。
3）notify-and-wait模式：前面的例子中都是goroutine创建者被动等待新goroutine退出，但是更多实现goroutine创建者需要主动通知新goroutine退出，尤其是当main作为goroutine的时候。main goroutine的退出意味着Go程序的终止，而粗暴地直接让main goroutine退出的方式可能导致业务数据损坏、不完整或丢失。我们可以通过notify-and-wait这一模式来满足这一要求。
a.通知并等待一个goroutine退出
import (
"time"
)
func worker(m int) {
time.Sleep(time.Second * time.Duration(m))
}
func spawn(f func(int)) chan string {
quit := make(chan string)
go func() {
job := make(chan int)
select {
case n := <-job:
f(n)
case status := <-quit:
quit <- status
}
}()
return quit
}
func main() {
quit := spawn(worker)
println("work goroutine start")
time.Sleep(time.Second * 5)
println("notify the work to exit")
quit <- "exit"
timer := time.NewTimer(time.Second * 10)
defer timer.Stop()
select {
case status := <-quit:
println("worker done:", status)
case <-timer.C:
println("wait worker exit timeout")
}
}
$ go run notify-and-wait-one-goroutine-exit-demo.go
work goroutine start
notify the work to exit
worker done: exit
上例中使用创建模式创建goroutine的spawn函数返回的channel成了一个双向数据通道：既承载创建者发送给新goroutine的退出信号，也承载新goroutine返回给创建者的退出状态。
b.通知并等待多个goroutine退出：当使用close函数关闭channel时，所有阻塞在该channel上的goroutine都会得到通知，这也是Go语言channel的一个特性。我们可以利用这一特性实现通知并等待多个goroutine退出。
import (
"fmt"
"sync"
"time"
)
func worker(m int) {
fmt.Printf("In worker-%d", m)
time.Sleep(time.Second * (time.Duration(m)))
}
func spawnGroup(n int, f func(int)) chan struct{} {
quit := make(chan struct{})
job := make(chan int)
var wg sync.WaitGroup
for i := 0; i < n; i++ {
wg.Add(1)
go func(i int) {
defer wg.Done()
name := fmt.Sprintf("worker-%d", i)
j, ok := <-job // j, ok会阻塞在这里，直到job这个channel被close函数关闭
if !ok {
println(name, "done")
return
}
f(j)
}(i)
}
go func() {
<-quit
close(job)
wg.Wait()
quit <- struct{}{}
}()
return quit
}
func main() {
quit := spawnGroup(5, worker)
println("spawn a group of workers")
time.Sleep(time.Second * 5)
println("notify the group worker to exit...")
quit <- struct{}{}
time.Sleep(time.Second * 5)
timer := time.NewTimer(time.Second * 10)
defer timer.Stop()
select {
case <-quit:
println("group workers done")
case <-timer.C:
println("wait group workers exit timeout")
}
}
$ go run notify-and-wait-group-goroutines-exit-demo.go
spawn a group of workers
notify the group worker to exit...
worker-4 done
worker-0 done
worker-2 done
worker-3 done
worker-1 done
group workers done
上例利用了worker goroutine接收任务的channel来广播退出通知，而实现这一广播就是close(job)。此时各个worker goroutine监听job channel，当创建者关闭job channel时，通过comma ok惯用法获取ok值为fasle，worker goroutine执行退出模式。
Go语言学习笔记、语法知识、技术要点和个人理解及实战
查看3道真题和解析
