1. 并发和并行的区别
并行:多核cpu在同一时间片内并行处理多个任务。
并发:如单核cpu在多个任务间进行时间片切换,并非同一时间片执行多个任务,只是上下文切换时间很短,看似多个任务并行。
多线程和多线程是并行的基本前提条件,单线程也可用协程做到并发。
在golang中是通过goroutine来实现并发的,goroutine并不能简单的归纳为协程,其运行时会创建多个线程来实现并发任务,且任务单元可被调度到其他线程并行执行。所以goroutine更像是多线程和协程的综合体,能最大限度提升执行效率,发挥多核处理能力。
goroutine
关键字go并非执行并发操作,而是创建一个并发任务单元。新建任务被放置在系统队列中,等待调度器安排合适的系统线程去获取执行权。
当前流程不会阻塞,不会等待该任务启动,且运行时也不保证并发任务的执行顺序。
每个任务单元除保存函数指针、调用参数外,还会分配执行所需的栈内存空间。相比系统默认MB级别的线程栈,goroutine自定义栈仅需2KB,所以才能创建成千上万的并发任务。自定义栈采取按需分配策略,在需要时仅需扩容,最大能到GB规模。
与defer一样,goroutine也会因延迟执行而立即计算并复制执行参数。
var c intfunc counter()int{ c++ return c }func main() { a:=100 go func(x,y int) { time.Sleep(time.Second) fmt.Println("go:",x,y) }(a,counter()) a+=100 fmt.Println("main:",a,counter()) time.Sleep(time.Second*3) }
输出:
main: 200 2 go: 100 1
wait
进程退出时不会等待并发任务结束,可用channel阻塞,然后发出退出信号。
func main() { exit:=make(chan interface{}) //创建通道。因为仅是通知,此处channel可为任何类型。 go func() { time.Sleep(time.Second) fmt.Println("goroutine done") close(exit) //关闭通道,发出信号。 }() fmt.Println("main...") <-exit //通道关闭则立即解除。 fmt.Println("main exit") }
输出:
main... goroutine donemain exit
除了关闭通道外,向通道内写入数据也可解除阻塞。channel的更多信息,后面再做详述。
如要等待多个任务结束,推荐使用sync.WaitGruop。通过设定计数器,让每个goroutine在退出前递减,直至归零时解除阻塞。
func main() { var wg sync.WaitGroup for i:=0;i<10;i++{ wg.Add(1) //累加计数 go func(id int) { defer wg.Done() //递减计数 time.Sleep(time.Second) fmt.Println("goroutine",id,"done") }(i) } fmt.Println("main...") wg.Wait() //阻塞,直到计数归零 fmt.Println("main exit") }
尽管WaitGroup.Add实现了原子操作,但建议在goroutine外累加计数器,以免Add尚未执行,Wait以及推出。
func main() { var wg sync.WaitGroup go func() { wg.Add(1) //可以运行试一下,不是每次都能设置上 defer wg.Done() //递减计数 fmt.Println("goroutine", "done") }() fmt.Println("main...") wg.Wait() //阻塞,直到计数归零 fmt.Println("main exit") }
可在多处用Wait阻塞,他们都能接收到通知。上栗就可在go func前加wg.Wait().
GOMAXPROCS
运行时可能会创建很多线程,但任何时候仅有限的几个线程参与并发任务执行。该数量默认与CPU核数相等,可用runtime.GOMAXPROCS函数(或环境变量)修改。
如参数小于1,GOMAXPROCS仅返回当前设置值,不做任何调整。
import ( "math" "fmt" "sync" "runtime")//测试目标函数func count(){ x:=0 for i:=0;i<math.MaxUint32;i++{ x+=i } fmt.Println(x) }//循环执行func test(n int){ for i:=0;i<n;i++{ count() } }//并发执行func test2(n int){ var wg sync.WaitGroup wg.Add(n) for i:=0;i<n;i++{ go func() { count() wg.Done() }() } wg.Wait() }func main() { n:=runtime.GOMAXPROCS(0) n1:=runtime.NumCPU() fmt.Println(n1) test(n) }
n:=runtime.GOMAXPROCS(0) n1:=runtime.NumCPU()
上述两个都可用来获取当前系统的cpu核数。
Local Storage
与线程不同,goroutine任务无法设置优先级,无法获取编号,没有局部存储(TLS),甚至连返回值都会被抛弃。但除优先级外,其他功能都很容易实现。
func main() { var wg sync.WaitGroup var gs [5]struct{ //用于实现类似TLS功能 id int //编号 result int //返回值 } for i:=0;i<len(gs);i++{ wg.Add(1) go func(id int) { //使用参数避免参数闭包延迟求值 defer wg.Done() gs[id].id = id gs[id].result=(id + 1) * 100 }(i) } wg.Wait() fmt.Printf("%+v\n",gs) }
输出:
[{id:0 result:100} {id:1 result:200} {id:2 result:300} {id:3 result:400} {id:4 result:500}]
**如使用map作为局部存储容器,建议做同步处理,因为运行时会对其做并发读写检查。
Gosched
暂停,释放线程去执行其他任务。当前任务被放回队列,等待下次调度时恢复执行。
func main() { runtime.GOMAXPROCS(1) exit:=make(chan struct{}) go func() { //任务a defer close(exit) go func() { //任务b,放在此处是为了确保a先执行。 fmt.Println("b") }() for i:=0;i<4;i++{ fmt.Println("a:", i) if i==1{ runtime.Gosched() } } }() <-exit}
输出:
a: 0 a: 1 b a: 2 a: 3
该函数很少被使用,因为运行时会主动向长时间运行(10ms)的任务发出抢占调度。
Goexit
Goexit立即终止当前任务,运行时确保所有已注册延迟调用被执行。该函数不会影响其他并发任务,不会引发panic,自然也就无法捕获。
func main() { exit:=make(chan struct{}) go func() { defer close(exit) defer println("a") func(){ defer func() { println("b",recover() ==nil) //执行recover返回nil }() func(){ //在多层调用中执行Goexit println("c") runtime.Goexit() //立即终止整个调用堆栈 println("c done.") //不会执行 }() println("b done.") //不会被执行 }() println("a done.") //不会执行 }() <- exit println("main exit.") }
输出:
c b truea main exit.
如果在main.main里调用Goexit,它会等待其他任务结束,然后让进程直接崩溃。
无论身处哪一层,Goexit都能立即终止整个调用堆栈,这与return仅退出当前函数不同。 标准库函数os.Exit可终止进程,但不会执行延迟调用。
2. 通道
Go并未实现严格的并发安全。
允许全局变量、指针、引用类型这些非安全内存共享操作,就需要开发人员自行维护数据一致性和完整性。Go鼓励使用CSP通道,以通信代替内存共享,实现并发安全。
通过消息来避免竟态的模型除了CSP,还有Actor。但两者区别较大。
作为CSP核心,通道是显式的,要求操作双方必须知道数据类型和具体通道,并不关心另一端操作者身份和数量。可如果另一端未准备妥当,或消息未能及时处理时,会阻塞当前端。
相比起来,Actor是透明的,它不在乎数据类型及通道,只要知道接收者信箱即可。默认就是异步方式,发送方对消息是否被接收和处理并不关心。
从底层实现上来说,通道只是一个队列。同步模式下,发送和接收双方配对,然后直接赋值数据给对方。如配对失败,则置入等待队列,直到另一方出现后才被唤醒。异步模式抢夺的则是数据缓冲槽。发送方要求有空槽可供写入,而接收方则要求有缓冲数据可读。需求不符时,同样加入等待队列,直到有另一方写入数据或腾出空槽后被唤醒。
除传递消息外,通道还被用作时间通知。
func main() { done:=make(chan struct{}) c:=make(chan string) go func() { s:=<-c fmt.Println(s) close(done) }() c<-"hello!" <-done //阻塞,直到有数据或通道关闭 }
同步模式必须有配对操作的goroutine出现,否则会一直阻塞。而异步模式在缓冲区未满或数据未读完前,不会阻塞。
func main() { c:=make(chan int,3) //创建带三个缓冲槽的异步通道 c<-1 //缓冲区未满,不会阻塞 c<-2 println(<-c) println(<-c) } 输出: 1 2
多数时候,异步通道有助于提升性能,减少排队阻塞。
缓冲区大小仅仅是内部属性,不属于类型组成部分。另外通道变量本身就是指针,可用相等操作符判断是否为同一对象或nil。
func main(){ var a,b chan int = make(chan int,3),make(chan int) var c chan bool fmt.Println(a==b) fmt.Println(c==nil) fmt.Printf("%p,%d\n",a,unsafe.Sizeof(a)) } 输出:falsetrue0xc04207a000,8
虽然可传递指针来避免数据复制,但须额外注意数据并发安全。
内置函数cap和len返回缓冲区大小和当前已缓冲数量;而对于同步通道则都返回0,据此可判断通道是同步还是异步。
func main(){ a,b:=make(chan int),make(chan int,3) b<-1 b<-2 println("a",len(a),cap(a)) println("b",len(b),cap(b)) }
输出:
a 0 0 b 2 3
收发
除使用简单的发送和接收操作符外,还可用ok-idom或range模式处理数据。
func main() { done :=make(chan struct{}) c:=make(chan int) go func() { defer close(done) for{ x,ok:=<-c if !ok{ //据此判断通道是否关闭 return } fmt.Println(x) } }() c<-1 c<-2 c<-3 close(c) <-done}
输出:1,2,3
对于循环接收数据,range模式更简洁一些。
[...] go func() { defer close(done) for x:=range c{ //循环获取消息,直到通道被关闭。 println(x) } }() [...]
及时用close函数关闭通道引发结束通知,否则可能会导致死锁。
通知可以是群体性的。也未必就是通知结束,可以是任何需要表达的事件。
一次性事件用close效率更好,没有多余开销。连续或多样性事件,可传递不同数据标志实现。还可使用sync.Cond实现单播或广播事件。
对于closed或nil通道,发送或接收操作都有相应规则:
向已关闭通道发送数据,引发panic。
从已关闭通道接收数据,返回已缓冲数据或零值。
无论收发,nil通道都会阻塞。
func main(){ c:=make(chan int,3) c<-10 c<-20 close(c) for i:=0;i<cap(c)+1;i++{ x,ok:=<-c println(i,":",ok,x) } }
输出:
0 : true 10 1 : true 20 2 : false 0 3 : false 0
重复关闭,或关闭nil通道都会引发panic错误。
单向
通道默认都是双向的,并不区分发送和接收端。但某些时候,我们可限制收发操作的方向来获得更严谨的操作逻辑。
尽管可用make创建单向通道,但那没有任何意义。通常使用类型转换来获取单向通道,并分别赋予操作双方。
func main() { var wg sync.WaitGroup wg.Add(2) c:=make(chan int) var send chan<- int =c var recv <-chan int =c go func() { defer wg.Done() for x:=range recv{ println(x) } }() go func() { defer wg.Done() defer close(c) for i:=0;i<3;i++{ send<-i } }() wg.Wait() }
不能在单向通道上做逆向操作。close也不能用于接收端。也无法将单向通道重新转换回去。
选择
如要同时处理多个通道,可选用Select语句。它会随机选择一个可用通道做收发操作。
func main() { var wg sync.WaitGroup wg.Add(2) a, b := make(chan int), make(chan int) go func() { //接收端 defer wg.Done() for { var ( name string x int ok bool ) select { case x, ok = <-a: //随机选择可用channel接收数据 name = "a" case x, ok = <-b: name = "b" } if !ok { //如果任一通道关闭则终止接收。 return } println(name, x) //输出接收的数据信息 } }() go func() { //发送端 defer wg.Done() defer close(a) defer close(b) for i := 0; i < 10; i++ { select { case a <- i: case b <- i * 10: } } }() wg.Wait() }
输出:
b 0 a 1 a 2 b 30 a 4 b 50 a 6 a 7 a 8 b 90
如果要等全部的通道消息处理结束,可将已完成的通道设置为nil,这样她就会被阻塞,而不再被Select选中。
以下示例是两个独立的通道,逻辑是等两个通道都结束了收发才最终close,哪个先完成哪个阻塞住在那等待。
func main() { var wg sync.WaitGroup wg.Add(3) a, b := make(chan int), make(chan int) go func() { //接收端 defer wg.Done() for { select { case x, ok := <-a: if !ok { a = nil break } println("a",x) case x, ok := <-b: if !ok { b = nil break } println("b", x) } if a == nil && b == nil { return } } }() go func() { defer wg.Done() defer close(a) for i := 0; i < 10; i++ { a <- i } }() go func() { defer wg.Done() defer close(b) for i := 100; i < 105; i++ { b <- i } }() wg.Wait() }
输出:
a 0 a 1 b 100 a 2 a 3 a 4 b 101 b 102 b 103 a 5 b 104 a 6 a 7 a 8 a 9
即使是同一通道,也会随机选择case执行。
func main() { var wg sync.WaitGroup wg.Add(2) c := make(chan int) go func() { //接收端 defer wg.Done() for { var x int var ok bool select { case x, ok = <-c: println("a1",x) case x, ok = <-c: println("a2", x) } if !ok { return } } }() go func() { defer wg.Done() defer close(c) for i := 0; i < 10; i++ { select { case c<-i: case c<-i*10: } } }() wg.Wait() }
输出:
a1 0 a1 1 a2 2 a2 3 a1 4 a1 50 a1 60 a1 7 a2 80 a2 90 a2 0
当所有通道都不可用时,Select会执行default语句。如此可避开Select阻塞,但须注意处理外层循环,以免陷入空耗。
func main() { c:=make(chan int) done:=make(chan bool) go func() { defer close(done) for { select { case x,ok:=<-c: if !ok{ return } fmt.Println("data:",x) default: //避免Select阻塞 } fmt.Println(time.Now()) time.Sleep(time.Second) } }() time.Sleep(5*time.Second) c<-100 close(c) <-done }
输出:略
也可以用default处理一些默认逻辑。
func main() { done := make(chan struct{}) data := []chan int{ //数据缓冲区 make(chan int, 3), } go func() { //生产数据 defer close(done) for i := 0; i < 10; i++ { select { case data[len(data)-1] <- i: //生产数据 default: //数据通道已满则新建chan data = append(data, make(chan int, 3)) } } }() <-done for x := 0; x < len(data); x++ { c := data[x] close(c) //关闭通道后也能从中读取数据 for i := range (c) { fmt.Println(i) } } }
输出:
0 1 2 4 5 6 8 9
可以看到,channel缓存满了后的第一个数据会被丢弃,直接走default创建新的通道了。
模式
通常使用工厂方法将goroutine和通道绑定。
type receiver struct { wg sync.WaitGroup data chan int}func newReceiver() *receiver { r := &receiver{ data: make(chan int), } r.wg.Add(1) go func() { defer r.wg.Done() for x := range r.data { //接收消息,直到通道关闭 println("recv:", x) } }() return r } func main() { r := newReceiver() r.data <- 1 r.data <- 2 close(r.data) //关闭通道,发出结束通知 r.wg.Wait() //等待接收者处理结束}
输出:
recv: 1 recv: 2
鉴于通道本身就是一个并发安全的队列,可用作ID generator、Pool等用途。
type pool chan []bytefunc newPool(cap int)pool{ return make(chan []byte,cap) } func (p pool)get() []byte{ var v []byte select { case v=<-p: //返回 default: v=make([]byte,10) //返回失败,新建 } return v } func (p pool)put(b []byte){ select { case p<-b: //放回 default: //放回失败,新建 } }
用通道实现信号量(semaphore)。
func main() { runtime.GOMAXPROCS(4) var wg sync.WaitGroup sem:=make(chan struct{}, 2) //最多允许两个并发同时执行 for i:=0;i<5;i++{ wg.Add(1) go func(id int) { defer wg.Done() sem<- struct{}{} // acquire: 获取信号 defer func() {<-sem}() //release: 释放信号 time.Sleep(time.Second * 2) fmt.Println(id,time.Now()) }(i) } wg.Wait() }
标准库time提供了timeout和tick channel实现。
package mainimport ( "time" "fmt" "os")func main() { go func() { for{ select { case <-time.After(time.Second*5): fmt.Println("timeout...") os.Exit(0) } } }() go func() { tick:=time.Tick(time.Second) //for _=range tick{ // fmt.Println(time.Now(),"test") //} for { select { case <-tick: fmt.Println(time.Now()) } } }() <-(chan struct {})(nil) //直接用nil channel阻塞进程}
捕获INT、TERM信号,顺便实现一个简易的atexit函数。
atexit函数是一个特殊的函数,它是在正常程序退出时调用的函数,我们把他叫为登记函数(函数原型:int atexit (void (*)(void))):
⼀个进程可以登记若⼲个(具体⾃⼰验证⼀下)个函数,这些函数由exit⾃动调⽤,这些函数被称为终⽌处理函数, atexit函数可以登记这些函数。 exit调⽤终⽌处理函数的顺序和atexit登记的顺序相反(网上很多说造成顺序相反的原因是参数压栈造成的,参数的压栈是先进后出,和函数的栈帧相同),如果⼀个函数被多次登记,也会被多次调⽤。
python中有专门的atexit模块,简介如下:
从模块的名字也可以看出来,atexit模块主要的作用就是在程序即将结束之前执行的代码,atexit模块使用register函数用于注册程序退出时的回调函数,然后在回调函数中做一些资源清理的操作。
注意:
1,如果程序是非正常crash,或通过os._exit()退出,注册的回调函数将不会被调用。
2,也可以通过sys.exitfunc来注册回调,但通过它只能注册一个回调,而且还不支持参数。
3,建议使用atexit来注册回调函数。
import ( "sync" "os" "os/signal" "syscall" "fmt")//type atexits struct {// sync.WaitGroup// signal chan os.Signal// funcs []func()//}var exits=&struct { sync.RWMutex signals chan os.Signal funcs []func() }{}func atexit(f func()){ exits.Lock() defer exits.Unlock() exits.funcs=append(exits.funcs,f) }func waitExit(){ if exits.signals==nil{ exits.signals = make(chan os.Signal) signal.Notify(exits.signals,syscall.SIGINT,syscall.SIGTERM) fmt.Println("test") } exits.RLock() for _,f:=range exits.funcs{ defer f() //延迟调用函数采用FILO顺序执行。即便某些函数panic,延迟调用也能确保后续函数执行。 } fmt.Println("after range exits.funcs") exits.RUnlock() fmt.Println("after exits.Runlock") <-exits.signals } func main() { atexit(func() { println("exit1...") }) atexit(func() { println("exit2...") }) fmt.Println("befor exit") waitExit() }
性能
将发往通道的数据打包,减少传输次数,可有效提升性能。从实现上来说,通道队列依旧使用锁同步机制,单次获取更多数据(批处理),可改善因频繁加锁造成的性能问题。
const ( max = 500000 //数据统计上限 block = 500 //数据块大小 bufsize = 100 //缓冲区大小)func test() { //普通模式,每次传递一个整数 done := make(chan struct{}) c := make(chan int, bufsize) go func() { count := 0 for x := range c { count += x } close(done) }() for i := 0; i < max; i++ { c <- i } close(c) <-done }func testBlock() { //块模式:每次将500个数字打包成块传输 done := make(chan struct{}) c:=make(chan [block]int,bufsize) go func() { count:=0 for a:=range c{ for _,x:=range a{ //a 是[block]int数组 count +=x } } fmt.Println(count) close(done) }() for i:=0;i<max;i+=block{ var b [block]int //使用数组对数据打包 for n:=0; n<block;n++{ b[n] = i+n if i+n == max -1{ break } } c <- b } close(c) <-done }
BenchmarkTest
虽然单次消耗更多内存,但性能提升非常明显。如将数组改成切片会造成更多内存分配次数。
资源泄漏
通道可能会引发goroutine leak,确切地说,是指goroutine处于发送或接受阻塞状态,但一直未被唤醒。垃圾回收器并不手机此类资源,导致它们会在等待队列里长久休眠,形成资源泄漏。
3. 同步
通道并不是用来取代锁的,它们有各自不同的应用场景。通道倾向于解决逻辑层次的并发处理架构,而锁则用来保护局部范围内的数据安全。
标准库sync提供了互斥和读写锁,另有原子操作等。mutex、rwmutex的使用并不复杂,只有几个地方需要注意。
将Mutex作为匿名字段时,相关方法必须实现为pointer-receiver,否则会因复制导致锁机制失效。
type data struct { sync.Mutex } func (d data)test(s string){ d.Lock() defer d.Unlock() for i:=0;i<5;i++{ fmt.Println(s, i) time.Sleep(time.Second) } }func main() { var wg sync.WaitGroup var d data wg.Add(2) go func() { defer wg.Done() d.test("Read") }() go func() { defer wg.Done() d.test("write") }() wg.Wait() }
上述代码运行后会发现锁机制已失效,解决方案是将data 改为data.
也可用嵌入Mutex来避免复制问题,但那需要专门初始化。
应将Mutex锁粒度控制在最小范围内,及早释放。
Mutex不支持递归锁,即锁里面不允许有锁,否则即使在同一goroutine下也会导致死锁。
在设计并发安全类型时,千万注意此类问题。
import "sync"type cache struct { sync.Mutex data []int} func (c *cache)count()int{ c.Lock() n:=len(c.data) c.Unlock() return n } func (c *cache)get() int{ c.Lock() defer c.Unlock() var d int if n:=c.count();n>0{ //锁中套锁 d=c.data[0] c.data=c.data[1:] } return d } func main() { c:=cache{data:[]int{1,2,3,4}} c.get() }
输出:
fatal error: all goroutines are asleep - deadlock!
相关建议:
对性能要求较高时,应避免使用defer Unlock.
读写并发时,用RWMutex性能会更好一些。
对单个数据读写保护,可尝试用原子操作。
执行严格测试,尽可能打开数据竞争检查。
作者:勇Max
链接:https://www.jianshu.com/p/1132baa9475c
共同學(xué)習(xí),寫下你的評(píng)論
評(píng)論加載中...
作者其他優(yōu)質(zhì)文章