|
一 Go并发模型 传统的编程语言C++ Java Python等，他们的并发逻辑多事基于操作系统的线程。并发执行单元（线程）之间的通信利用的就是操作系统提供的线程或进程间通信的原语。如：共享内存、信号、管道、消息队列、套接字等。在这些通信原语中，使用最广泛的就是共享内存。 如果你使用过这种共享内存的并发模型，其实是难用的和容易发生错误的，特别是在大型或复杂的业务场景中。 Go语言从程序设计当初，就将解决上面传统并发模型问题作为目标，并在新并发模型设计中借鉴注明的CSP(Communicationing Sequential Processes-通信顺序进程)并发模型。 CSP模型目的在于简化并发程序的编写，让并发程序的编写顺序与编写顺序程序一样简单。 生产者 —》输出数据 — 输入/输出原语 —》输出数据 为了实现CSP模型，GO语言引入了Channel.Goroutine可以读写channel中的数据，通过channel将goroutine组合连接在一起。 Go语言中CSP虽然是主流并发模型，但是还是支持共享内存并发模型。主要是在sync包中的互斥锁、读写锁、条件变量、原子操作等。那么我们该如何选择呢？ 第一种：创建模式 通常会使用下面的方式： type Worker struct { } func Do(f func()) chan Worker { w:= make(chan Worker) go func() { f() w<-Worker{} }() return w } func main() { c:=Do(func() { fmt.Print('到下班时间了...') }) <-c } Do函数内部创建了一个gorutine并且返回了一个channel类型的变量。Do函数创建的新goroutine与调用的Do函数的goroutine之间通过一个channel联系了起来，2个goroutine可以通过channel进行通讯。Do函数的实现因为channel在Go语言中是一等公民，channel可以像变量一样初始化、传递和赋值。上面的例子Do返回了一个变量，这个变量就是通道，实现了主goroutine和子goroutine的通信。 第二种：退出模式 a) 分离模式 分离模式使用最广泛的是goroutine退出模式。所谓分离模式就是创建它的goroutine不需要关心它的退出，这类goroutine启动后与其创建者彻底分离，其生命周期与其执行的主函数相关，函数返回即goroutine退出。 场景1：一次性任务 // $GOROOT/src/net/dial.go func (d *Dialer) DialContext(ctx context.Context, network, address string) (Conn, error) { ... ... if oldCancel := d.Cancel; oldCancel != nil { subCtx, cancel := context.WithCancel(ctx) defer cancel() go func() { select { case <-oldCancel: cancel() case <-subCtx.Done(): } }() ctx = subCtx } ... ... } 在DialContext方法中创建了一个goroutine，用来监听量个channel是否有数据，一旦有数据，处理后即退出。 场景2 常驻后台执行一些特定任务，比如常用for{…}或for{select{…}}形式，还可以用定时器或事件驱动执行。下面是Go给每个P内置的GC goroutine就是这种场景的。 // $GOROOT/src/runtime/mgc.go func gcBgMarkStartWorkers() { // Background marking is performed by per-P G's. Ensure that // each P has a background GC G. for _, p := range allp { if p.gcBgMarkWorker == 0 { go gcBgMarkWorker(p) // 这里每个P创建一个goroutine，以运行gcBgMarkWorker notetsleepg(&work.bgMarkReady, -1) noteclear(&work.bgMarkReady) } } } func gcBgMarkWorker(_p_ *p) { gp := getg() ... ... for { // 处理GC ... ... } } b) join模式 在线程模型中，父线程可以通过pthread join来等待子线程结束并获取子线程的结束状态。在Go中，我们有时候也有这种需求：goroutine的创建者需要等待新goroutine的结果。 type Worker struct { } func Do(f func()) chan Worker { w:= make(chan Worker) go func() { f() w<-Worker{} }() return w } func main() { c:=Do(func() { fmt.Print('到下班时间了...') }) <-c } 我们还是看刚刚上面的这个例子，Do函数使用典型的goroutine的创建模式创建了一个groutine，main的goroutine作为创建通过Do函数返回的channel与新goroutine建立关系，这个channel得用途就是在goroutine之间建立退出时间的“信号”通信机制。main goroutine在创建完新goroutine后就在该channel上阻塞等待了，直到新的goroutine退出前向该channel发送了一个”信号”。 运行代码，结果如下： 到下班时间了... Process finished with exit code 0 获取goroutine的退出状态 如果新goroutine的创建者不仅仅要等待goroutine的退出，还要知道结束状态，我们可以通过自定义类型的channel来实现这样的需求。 func add(a,b int) int{ return a+b } func Do(f func(a,b int) int,a,b int) chan int{ c:=make(chan int) go func() { r:=f(a,b) c<-r }() return c } func main() { c:=Do(add,1,5) fmt.Println(<-c) } 运行结果是 6 等待多个goroutine退出 func add(a,b int) int{ return a+b } func Do(f func(a,b int) int,a,b,n int) chan int{ c:=make(chan int) var wg sync.WaitGroup for i:=0;i<n;i++{ wg.Add(1) go func() { r:=f(a,b) fmt.Println(r) wg.Done() }() } go func() { wg.Wait() c<-100 }() go func() { }() return c } func main() { c:=Do(add,1,5,5) fmt.Println(<-c) } 运行结果 6 6 6 6 6 100 c) notify-wait模式 前面的场景中，goroutine的创建者都是在被动地等待新goroutine的退出。有些场景，goroutine的创建者需要主动通知那些新goroutine退出。 通知并等待一个goroutine的退出 func add(a, b int) int { return a + b } func Do(f func(a, b int) int, a, b int) chan int { quit := make(chan int) go func() { var job chan string for { select { case x := <-job: f(a, b) fmt.Println(x) case y := <-quit: quit <- y } } }() return quit } func main() { c := Do(add, 1, 5) fmt.Println('开始干活') time.Sleep(1 * time.Second) c <- 0 timer := time.NewTimer(time.Second * 10) defer timer.Stop() select { case status := <-c: fmt.Println(status) case <-timer.C: fmt.Println('等待...') } } 执行代码结果如下 开始干活 0 通知并等待多个goroutine退出 下面是通知并等待多个goroutine退出的场景。Go语言的channel有一个特性，那就是当使用close函数关闭channel时，所有阻塞到该channel上的goroutine都会得到通知。 func worker(x int) { time.Sleep(time.Second * time.Duration(x)) } func Do(f func(a int), n int) chan int { quit := make(chan int) 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) for { j,ok:=<-job if !ok{ fmt.Println(name,'done') return } worker(j) } }(i) } go func() { <-quit close(job) wg.Wait() quit<-200 }() return quit } func main() { quit:=Do(worker,5) fmt.Println('func Work...') quit<-1 timer := time.NewTimer(time.Second * 10) defer timer.Stop() select { case status := <-quit: fmt.Println(status) case <-timer.C: fmt.Println('等待...') } } 运行结果 func Work... worker-1 done worker-2 done worker-3 done worker-4 done worker-0 done 200 |
|
|
