2020-11-1616:48大小Go并發編程實戰16:48大小在前面的課程里，我們學習了標準庫的并發原語、原子操作和Channel，掌握了這些，你就可以解決80%的并發編程問題了。但是，如果你要想進一步提升你的并發編程能力，就所以，在接下來的幾節課里，我會給你分享Go官方或者其他人提供的第三方庫，這節課我們先來學習信號量，信號量（Semaphore）是用來控制多個goroutine源的并發原語。信號量的概念是荷蘭計算機科學家Edsgerjstra在3不同的操作系統中。在系統中，會給每一個進程一個信號量，代表每個進程目前的狀態。未得到控制權的進程，會在特定的地方被迫停下來，等待可以繼續進行的信號到來。最簡單的信號量就是一個變量加一些并發控制的能力，這個變量是0到n之間的一個數值。當goroutine完成對此信號量的等待（wait）時，該計數值就減1，當goroutine完成對此信號量的釋放（release）時，該計數值就加1。當計數值為0的時候，goroutinewait0，goroutine10本《Go1書，“僧多粥少”。所以，圖書館管理員先會讓這1萬個同學進行登記，按照登記的順書了，圖書館管理員就會通知下一位同學來借閱這本書。這里的資源是《Go并發編程的獨家秘籍》這十本書，想讀此書的同學就是goroutine，圖書管理員就是信號量。怎么樣，現在是不是很好理解了？那么，接下來，我們來學習下信號量的P/V操作。P/VDijkstra在他的論文中為信號量定義了兩個操作P和V。P操作（descrease、wait、acquire）V（increase、signal、release）是增加信號11functionV(semaphoreS,integer[[S←S+functionP(semaphoreS,integerI):[ifS≥I:S←S?IS（n）n子。P操作相當于請求資源，如果資源可用，就立即返回；如果沒有資源或者不夠，那么，它可以不斷嘗試或者阻塞等待。V操作會釋放自己持有的資源，把資源返還給信號量。信號量的值除了初始化的操作以外，只能由P/V操作改變。P1，如果新值已經為負，那么調用者會被阻塞并加入到V1，P講到這里，我想再稍微說一個題外話，我們在第2場景下，會有些goroutine始終搶不到鎖。為了處理饑餓的問題，你可以在等待隊列中做其實，信號量可以分為計數信號量（countingsemaphre）和二進位信號量（binarysemaphore）。剛剛所說的圖書館借書的例子就是一個計數信號量，它的計數可以是任意一個整數。在特殊的情況下，如果計數值只能是0或者1，那么，這個信號量就是二進位信號量，提供了互斥的功能（要么是0，要么是），信號量來實現。我們一般用信號量保護一組資源，比如數據庫連接池、一組客戶端的連接、幾個打印機資源，等等。如果信號量蛻變成二進位信號量，那么，它的P/V就和互斥鎖的一樣了。有人會很細致地區分二進位信號量和互斥鎖。比如說，有人提出，在Windows系統中， Stack上也有相關的討論）。實際上，雖然在Windows系統中，它們的確有些區別，但是對Go語言來說，互斥鎖也可以由非持有的goroutine來釋放，所以，從行為上來說，它們并沒有嚴Go在運行時，GogoroutinetypetypeMutexstruct{stateint32semauint324信號量的P/Vfuncfuncruntime_Semacquire(sfuncruntime_SemacquireMutex(s*uint32,lifobool,skipframesint)funcruntime_Semrelease(s*uint32,handoffbool,skipframesint)遺憾的是，它是Go運行時內部使用的，并沒有封裝暴露成一個對外的信號量并發原語，原則上我們沒有辦法使用。不過沒關系，Go在它的擴展包中提供了信號量semaphore，不過這個信號量的類型名并不叫Semaphore，而是叫Weighted。之所以叫做Weighted，（初始化的資源數），其實我覺得叫SemaphoreAcquirePContext，Context超時或者cancel的機制。如果是正常獲取了資源，就返回nil；否則，就返回ctx.Err()，Release方法：相當于V操作，可以將nTryAcquirenn回true，要么一個也不獲取，返回false。WorkerPoolCPUWorker，4slice。每個Worker一次只能處理一個整數，處理完之后，才能處理下一個。11varmaxWorkers=//worker8:=9i:=rangetask//如果沒有worker可用，會阻塞在這里，直到某個workeriferr:=sema.Acquire(ctx,1);err!=nil}//啟動workergofunc(iint)defertime.Sleep(100*time.Millisecond)//task[i]=i+}iferr:=sema.Acquire(ctx,int64(maxWorkers));err!=nillog.Printf("獲取所有的worker失敗:%v", 30在這段代碼中，maingoroutine相當于一個dispacher，負責任務的分發。它先請求信號量，如果獲取成功，就會啟動一個goroutine去處理計算，然后，這個goroutine會釋放這個信號量（有意思的是，信號量的獲取是在maingoroutine，信號量的釋放是在workergoroutine中），如果獲取不成功，就等到有信號量可以使用的時候，再去獲取。需要提醒你的是，其實，在這個例子中，還有一個值得我們學習的知識點，就是最后的那一段處理（第5行）。rr。Go擴展庫中的信號量是使用互斥鎖+List實現的。互斥鎖實現其它字段的保護，而List實現了一個等待隊列，等待者的通知是通過Channel的通知機制實現的。我們來看一下信號量Weighted2//34//56//在信號量的幾個實現方法里，Acquire否可用，而且還要檢測Context的Done是否已關閉。我們來看下它的實現代碼。func(s*Weighted)Acquire(ctxcontext.Context,nint64)error//fastpath,如果有足夠的資源，都不考慮ctx.Done的狀態，將cur加上nifs.size-s.cur>=n&&s.waiters.Len()==0s.cur+=return 9ifn>s.size//依賴ctxreturn ////創建了一個readychan,ready:=make(chanw:=waiter{n:n,ready:elem:=//selectcase<-ctx.Done()://context的Doneerr:=selectcase<-ready://如果被喚醒了，忽略ctxerr=default:通知isFront:=s.waiters.Front()==//通知其它的waiters,ifisFront&&s.size>s.cur}}returncase<-ready://return}}fastpathacquireSlow法，以便其它Acquire被內聯。Releasen，并喚醒等待隊列中的調用者，看是否有funcfunc(s*Weighted)Release(nint64){s.cur-=ifs.cur<0{panic("semaphore:releasedmorethan 10notifyWaiters2for3next4if==nil5//Nomorewaiters6}78ifs.size-s.cur<w.ns.cur+=w.nnotifyWaiters100個資源的時候，如果第一個等待者需要101個資源，那么，隊列中的所有等待者都會繼續等待，即使有的等待1序panic。候，也可能會出現死鎖現象，比如哲學家就餐問題。GoRelease但是，如果你傳遞一個比請求到的數量大的錯誤的數值，程序就會panic。如果傳遞一個所以，使用信號量遵循的原則就是請求多少資源，就釋放多少資源源數。Channel來實現。ChannelGobuffer為n的Channel很容易實現信號量，比如下面的代碼，我們就是使用chan用Lock和Unlock方法實現請求資源和釋放資源，正好實現了Locker接口。1//Semaphore數據結構，并且還實現了Locker2typesemaphorestruct34chchan5}678funcNewSemaphore(capacityint){9ifcapacity<=0capacity=1//容量為1}return&semaphore{ch:make(chan}func(s*semaphore)Lock()s.ch<-}//func(s*semaphore)Unlock() 當然，你還可以自己擴展一些方法，比如在請求資源的時候使用Context參數（Acquire(ctx)）、實現TryLock量的場景，為什么官方擴展庫的信號量的實現不采用這種方法呢？其實，具體是什么原因，我也不知道，但是我必須要強調的是，官方的實現方式有這樣一個功能：l。除了 marusama/semaphore也實現了一個可以動態更改資源容量的號量，也是一個非常有特色的實現。如果你的資源數量并不是固定的，而是動態變化的，我建議你考慮一下這個信號量庫。這是一個很奇怪的現象：標準庫中實現基本并發原語（比如Mut）號量實現等待隊列和通知喚醒，但是，