作者：dcguo，騰訊 CSIG 電子簽開放平臺中心
分享 Golang 并發(fā)基礎庫，擴展以及三方庫的一些常見問題、使用介紹和技巧，以及對一些并發(fā)庫的選擇和優(yōu)化探討。
提倡的原則
不要通過共享內存進行通信;相反，通過通信來共享內存。
Goroutine
goroutine 并發(fā)模型
調度器主要結構
主要調度器結構是 M，P，G
P 的數(shù)量由環(huán)境變量中的 GOMAXPROCS 決定，通常來說和核心數(shù)對應。
用戶空間線程和內核空間線程映射關系有如下三種:
關系如圖，灰色的 G 則是暫時還未運行的，處于就緒態(tài)，等待被調度，這個隊列被 P 維護
注: 簡單調度圖如上，有關于 P 再多個 M 中切換，公共 goroutine 隊列，M 從線程緩存中創(chuàng)建等步驟沒有體現(xiàn)，復雜過程可以參考文章簡單了解 goroutine 如何實現(xiàn)。
go list.Sort()func Announce(message string, delay time.Duration) {go func() {time.Sleep(delay)fmt.println(message)}()}channel 特性
創(chuàng)建
// 創(chuàng)建 channela := make(chan int)b := make(chan int, 10)// 單向 channelc := make(chan<- int)d := make(<-chan int)
tip:
v, ok := <-a // 檢查是否成功關閉(ok = false：已關閉)ci := make(chan int)cj := make(chan int, 0)cs := make(chan *os.File, 100)c := make(chan int)go func() {list.Sort()c <- 1}()doSomethingForValue<- cfunc Server(queue chan *Request) {for req := range queue {sem <- 1go func() {process(req)<- sem}()}}func Server(queue chan *Requet) {for req := range queue {sem <- 1go func(req *Request) {process(req)<- sem}(req)}}func Serve(queue chan *Request) {for req := range queue {req := reqsem <- 1go func() {process(req)<-sem}()}}c := make(chan bool)go func() { // close 的 channel 會讀到一個零值 close(c)}()<-c開源項目【是一個支持集群的 im 及實時推送服務】里面的基準測試的案例
func main() { ret := make(chan string, 3) for i := 0; i < cap(ret); i++ { go call(ret) } fmt.Println(<-ret)}func call(ret chan<- string) { // do something // ... ret <- 'result'}注: 協(xié)同多個 goroutines 方案很多，這里只展示 channel 的一種。
limits := make(chan struct{}, 2)for i := 0; i < 10; i++ { go func() { // 緩沖區(qū)滿了就會阻塞在這 limits <- struct{}{} do() <-limits }()}for { select { case a := <- testChanA: // todo a case b, ok := testChanB: // todo b, 通過 ok 判斷 tesChanB 的關閉情況 default: // 默認分支 }}func worker(testChan chan bool) { for { select { // todo some // case ... case <- testChan: testChan <- true return } }}func main() { testChan := make(chan bool) go worker(testChan) testChan <- true <- testChan}testChan := make(chan bool)close(testChan)zeroValue := <- testChanfmt.Println(zeroValue) // falsetestChan <- true // panic: send on closed channel注: 如果是 buffered channel, 即使被 close, 也可以讀到之前存入的值，讀取完畢后開始讀零值，寫入則會觸發(fā) panic
略
for rangec := make(chan int, 20)go func() { for i := 0; i < 10; i++ { c <- i } close(c)}()// 當 c 被關閉后，取完里面的元素就會跳出循環(huán)for x := range c { fmt.Println(x)}func newUniqueIdService() <-chan string { id := make(chan string) go func() { var counter int64 = 0 for { id <- fmt.Sprintf('%x', counter) counter += 1 } }() return id}func newUniqueIdServerMain() { id := newUniqueIdService() for i := 0; i < 10; i++ { fmt.Println(<- id) }}略
func main() {done := do()select {case <-done:// logiccase <-time.After(3 * time.Second):// timeout}}demo
開源 im/goim 項目中的應用
2.心跳
done := make(chan bool)defer func() {close(done)}()ticker := time.NewTicker(10 * time.Second)go func() {for {select {case <-done:ticker.Stop()returncase <-ticker.C:message.Touch()}}}()}func main() { c := make(chan struct{}) for i := 0; i < 5; i++ { go do(c) } close(c)}func do(c <-chan struct{}) { // 會阻塞直到收到 close <-c fmt.Println('hello')}func channel() { count := 10 // 最大并發(fā) sum := 100 // 總數(shù) c := make(chan struct{}, count) sc := make(chan struct{}, sum) defer close(c) defer close(sc) for i:=0; i<sum; i++ { c <- struct{} go func(j int) { fmt.Println(j) <- c // 執(zhí)行完畢，釋放資源 sc <- struct {}{} // 記錄到執(zhí)行總數(shù) } } for i:=sum; i>0; i++ { <- sc }}這塊東西為什么放到 channel 之后，因為這里包含了一些低級庫，實際業(yè)務代碼中除了 context 之外用到都較少(比如一些鎖 mutex，或者一些原子庫 atomic)，實際并發(fā)編程代碼中可以用 channel 就用 channel，這也是 go 一直比較推崇得做法 Share memory by communicating; don’t communicate by sharing memory
鎖，使用簡單，保護臨界區(qū)數(shù)據(jù)
使用的時候注意鎖粒度，每次加鎖后都要記得解鎖
Mutex demo
package mainimport ( 'fmt' 'sync' 'time')func main() { var mutex sync.Mutex wait := sync.WaitGroup{} now := time.Now() for i := 1; i <= 3; i++ { wait.Add(1) go func(i int) { mutex.Lock() time.Sleep(time.Second) mutex.Unlock() defer wait.Done() }(i) } wait.Wait() duration := time.Since(now) fmt.Print(duration)}結果: 可以看到整個執(zhí)行持續(xù)了 3 s 多，內部多個協(xié)程已經(jīng)被 “鎖” 住了。
RWMutex demo
注意: 這東西可以并發(fā)讀，不可以并發(fā)讀寫/并發(fā)寫寫，不過現(xiàn)在即便場景是讀多寫少也很少用到這，一般集群環(huán)境都得分布式鎖了。
package mainimport ( 'fmt' 'sync' 'time')var m *sync.RWMutexfunc init() { m = new(sync.RWMutex)}func main() { go read() go read() go write() time.Sleep(time.Second * 3)}func read() { m.RLock() fmt.Println('startR') time.Sleep(time.Second) fmt.Println('endR') m.RUnlock()}func write() { m.Lock() fmt.Println('startW') time.Sleep(time.Second) fmt.Println('endW') m.Unlock()}輸出:
可以對簡單類型進行原子操作
int32
int64
uint32
uint64
uintptr
unsafe.Pointer
可以進行得原子操作如下
增/減
比較并且交換
假定被操作的值未曾被改變, 并一旦確定這個假設的真實性就立即進行值替換
載入
為了原子的讀取某個值(防止寫操作未完成就發(fā)生了一個讀操作)
存儲
原子的值存儲函數(shù)
交換
原子交換
demo:增
package mainimport ( 'fmt' 'sync' 'sync/atomic')func main() { var sum uint64 var wg sync.WaitGroup for i := 0; i < 100; i++ { wg.Add(1) go func() { for c := 0; c < 100; c++ { atomic.AddUint64(&sum, 1) } defer wg.Done() }() } wg.Wait() fmt.Println(sum)}結果:
waitGroup 是一個 waitGroup 對象可以等待一組 goroutinue 結束，但是他對錯誤傳遞，goroutinue 出錯時不再等待其他 goroutinue(減少資源浪費) 都不能很好的解決，那么 errGroup 可以解決這部分問題
注意
demo: errGroup
package mainimport ( 'golang.org/x/sync/errgroup' 'log' 'net/http')func main() { var g errgroup.Group var urls = []string{ 'https://github.com/', 'errUrl', } for _, url := range urls { url := url g.Go(func() error { resp, err := http.Get(url) if err == nil { _ = resp.Body.Close() } return err }) } err := g.Wait() if err != nil { log.Fatal('getErr', err) return }}結果:
保證了傳入的函數(shù)只會執(zhí)行一次，這常用在單例模式，配置文件加載，初始化這些場景下。
demo:
times := 10 var ( o sync.Once wg sync.WaitGroup ) wg.Add(times) for i := 0; i < times; i++ { go func(i int) { defer wg.Done() o.Do(func() { fmt.Println(i) }) }(i) } wg.Wait()結果:
go 開發(fā)已經(jīng)對他了解了太多
可以再多個 goroutinue 設置截止日期，同步信號，傳遞相關請求值
對他的說明文章太多了，詳細可以跳轉看這篇 一文理解 golang context
這邊列一個遇到得問題：
我們可以再每個 CPU 上進行循環(huán)無關的迭代計算，我們僅需要創(chuàng)建完所有的 goroutine 后，從 channel 中讀取結束信號進行計數(shù)即可。
這部分如需自己開發(fā)，內容其實可以分為兩部分能力去做
并發(fā)編程增強方案
工作流解決方案
需要去解決一些基礎問題
并發(fā)編程:
啟動 goroutine 時，增加防止程序 panic 能力
去封裝一些更簡單的錯誤處理方案，比如支持多個錯誤返回
限定任務的 goroutine 數(shù)量
工作流:
在每個工作流執(zhí)行到下一步前先去判斷上一步的結果
工作流內嵌入一些攔截器
一般系統(tǒng)重要的查詢增加了緩存后，如果遇到緩存擊穿，那么可以通過任務計劃，加索等方式去解決這個問題，singleflight 這個庫也可以很不錯的應對這種問題。
它可以獲取第一次請求得結果去返回給相同得請求 核心方法 Do 執(zhí)行和返回給定函數(shù)的值，確保某一個時間只有一個方法被執(zhí)行。
如果一個重復的請求進入，則重復的請求會等待前一個執(zhí)行完畢并獲取相同的數(shù)據(jù)，返回值 shared 標識返回值 v 是否是傳遞給重復的調用請求。一句話形容他的功能，它可以用來歸并請求，但是最好加上超時重試等機制，防止第一個 執(zhí)行 得請求出現(xiàn)超時等異常情況導致同時間大量請求不可用。
場景: 數(shù)據(jù)變化量小(key 變化不頻繁，重復率高)，但是請求量大的場景
demo
package mainimport ( 'golang.org/x/sync/singleflight' 'log' 'math/rand' 'sync' 'time')var ( g singleflight.Group)const ( funcKey = 'key' times = 5 randomNum = 100)func init() { rand.Seed(time.Now().UnixNano())}func main() { var wg sync.WaitGroup wg.Add(times) for i := 0; i < times; i++ { go func() { defer wg.Done() num, err := run(funcKey) if err != nil { log.Fatal(err) return } log.Println(num) }() } wg.Wait()}func run(key string) (num int, err error) { v, err, isShare := g.Do(key, func() (interface{}, error) { time.Sleep(time.Second * 5) num = rand.Intn(randomNum) //[0,100) return num, nil }) if err != nil { log.Fatal(err) return 0, err } data := v.(int) log.Println(isShare) return data, nil}連續(xù)執(zhí)行 3 次，返回結果如下，全部取了共享得結果：
但是注釋掉 time.Sleep(time.Second * 5) 再嘗試一次看看。
這次全部取得真實值
實踐: 伙伴部門高峰期可以減少 20% 的 Redis 調用, 大大減少了 Redis 的負載
注: 下面用到的方案因為開發(fā)時間較早，并不一定是以上多種方案中最優(yōu)的，選擇有很多種，使用那種方案只有有所考慮可以自圓其說即可。
建議: 項目中逐漸形成統(tǒng)一解決方案，從混亂到統(tǒng)一，逐漸小團隊內對此類邏輯形成統(tǒng)一的一個解決標準，而不是大家對需求之外的控制代碼寫出各式各樣的控制邏輯。
批量校驗接口限頻單賬戶最高 100qps/s，整個系統(tǒng)多個校驗場景公用一個賬戶限頻需要限制批量校驗最高為 50~80 qps/s(需要預留令牌供其他場景使用，否則頻繁調用批量接口時候其他場景均會失敗限頻)。
1.使用 go routine 來并發(fā)進行三要素校驗，因為 go routinue，所以每次開啟 50 ~ 80 go routine 同時進行單次三要素校驗；
2.每輪校驗耗時 1s，如果所有 routinue 校驗后與校驗開始時間間隔不滿一秒，則需要主動程序睡眠至 1s，然后開始下輪校驗；
3.因為只是校驗場景，如果某次校驗失敗，最容易的原因其實是校驗方異常，或者被其他校驗場景再當前 1s 內消耗過多令牌；那么整個批量接口返回 err，運營同學重新發(fā)起就好。
代碼需要進行的優(yōu)化點：
1.sleep 1s 這個操作可以從調用前開始計時，調用完成后不滿 1s 補充至 1s，而不是每次最長調用時間 elapsedTime + 1s；
2.通道中獲取的三要素校驗結果順序和入?yún)?shù)據(jù)數(shù)組順序不對應，這里通過兩種方案：
3.分組調用
getElementResponseConcurrent 方法時，傳入切片可以省略部分計算，直接使用切片表達式。
elementNum := len(elements)m := elementNum / 80n := elementNum % 80if m < 1 {if results, err := getElementResponseConcurrent(ctx, elements, conn, caller); err != nil {return nil, err} else {response.Results = resultsreturn response, nil}} else {results := make([]int64, 0)if n != 0 {m = m + 1}var result []int64for i := 1; i <= m; i++ {if i == m {result, err = getElementResponseConcurrent(ctx, elements[(i-1)*80:(i-1)*80+n], conn, caller)} else {result, err = getElementResponseConcurrent(ctx, elements[(i-1)*80:i*80], conn, caller)}if err != nil {return nil, err}results = append(results, result...)}response.Results = results}// getElementResponseConcurrentfunc getElementResponseConcurrent(ctx context.Context, elements []*api.ThreeElements, conn *grpc.ClientConn,caller *api.Caller) ([]int64, error) {results := make([]int64, 0)var chResult = make(chan int64)chanErr := make(chan error)defer close(chanErr)wg := sync.WaitGroup{}faceIdClient := api.NewFaceIdClient(conn)for _, element := range elements {wg.Add(1)go func(element *api.ThreeElements) {param := element.ParamverificationRequest := &api.CheckMobileVerificationRequest{Caller: caller,Param: param,}if verification, err := faceIdClient.CheckMobileVerification(ctx, verificationRequest); err != nil {chanErr <- errreturn} else {result := verification.ResultchanErr <- nil chResult <- result}defer wg.Done()}(element)}for i := 0; i < len(elements); i++ {if err := <-chanErr; err != nil {return nil, err}var result = <-chResultresults = append(results, result)}wg.Wait()time.Sleep(time.Second)return results, nil}場景: 產(chǎn)品上線一年，逐步開始做數(shù)據(jù)分析和統(tǒng)計需求提供給運營使用，接入 Tdw 之前是直接采用接口讀歷史表進行的數(shù)據(jù)分析，涉及全量用戶的分析給用戶記錄打標簽，數(shù)據(jù)效率較低，所以采用并發(fā)分組方法，考慮協(xié)程比較輕量，從開始上線時間節(jié)點截止當前時間分共 100 組，代碼較為簡單。
問題: 本次接口不是上線最終版，核心分析方法僅測試環(huán)境少量數(shù)據(jù)就會有 N 多條慢查詢，所以這塊還需要去對整體資源業(yè)務背景問題去考慮，防止線上數(shù)據(jù)量較大還有慢查詢出現(xiàn) cpu 打滿。
func (s ServiceOnceJob) CompensatingHistoricalLabel(ctx context.Context,request *api.CompensatingHistoricalLabelRequest) (response *api.CompensatingHistoricalLabelResponse, err error) {if request.Key != interfaceKey {return nil, transform.Simple('err')}ctx, cancelFunc := context.WithCancel(ctx)var (wg = new(sync.WaitGroup)userRegisterDb = new(datareportdb.DataReportUserRegisteredRecords)startNum = int64(0))wg.Add(1)countHistory, err := userRegisterDb.GetUserRegisteredCountForHistory(ctx, historyStartTime, historyEndTime)if err != nil {return nil, err}div := decimal.NewFromFloat(float64(countHistory)).Div(decimal.NewFromFloat(float64(theNumberOfConcurrent)))f, _ := div.Float64()num := int64(math.Ceil(f))for i := 0; i < theNumberOfConcurrent; i++ {go func(startNum int64) {defer wg.Done()for {select {case <- ctx.Done():returndefault:userDataArr, err := userRegisterDb.GetUserRegisteredDataHistory(ctx, startNum, num)if err != nil {cancelFunc()}for _, userData := range userDataArr {if err := analyseUserAction(userData); err != nil {cancelFunc()}}}}}(startNum)startNum = startNum + num}wg.Wait()return response, nil}實現(xiàn)思路和上面其實差不多，都是需要支持批量的特性，基本上現(xiàn)在業(yè)務中統(tǒng)一使用多協(xié)程處理。
基本上可以這樣理解這件事
一個 goroutine 啟動后沒有正常退出，而是直到整個服務結束才退出，這種情況下，goroutine 無法釋放，內存會飆高，嚴重可能會導致服務不可用
goroutine 的退出其實只有以下幾種方式可以做到
大多數(shù)引起 goroutine 泄露的原因基本上都是如下情況
杜絕:
排查:
案例:
package mainimport ( 'fmt' 'net/http' _ 'net/http/pprof' 'runtime' 'time')func toLeak() { c := make(chan int) go func() { <-c }()}func main() { go toLeak() go func() { _ = http.ListenAndServe('0.0.0.0:8080', nil) }() c := time.Tick(time.Second) for range c { fmt.Printf('goroutine [nums]: %d\n', runtime.NumGoroutine()) }}輸出:
pprof:
復雜情況也可以用其他的可視化工具:
使用方便，支持鏈式調用
父協(xié)程捕獲子協(xié)程 panic
有鎖的地方就去用 channel 優(yōu)化，這句話可能有點絕對，肯定不是所有場景都可以做到，但是大多數(shù)場景絕 X 是可以的，干掉鎖去使用 channel 優(yōu)化代碼進行解耦絕對是一個有趣的事情。
分享一個很不錯的優(yōu)化 demo:
場景:
分析:
問題:
解決
增加鎖機制，解決針對鏈接池的并發(fā)問題發(fā)送消息也需要去加鎖因為要防止出現(xiàn) panic: concurrent write to websocket connection
假設網(wǎng)絡延時，用戶新增時候還有消息再發(fā)送中，新加入的用戶就無法獲得鎖了，后面其他的相關操作都會被阻塞導致問題。
使用 channel 優(yōu)化:
2.使用通道
3.通道消息方法，代碼來自于開源項目 簡單聊天架構演變：
// 處理所有管道任務func (room *Room) ProcessTask() {log := zap.S()log.Info('啟動處理任務')for {select {case c := <-room.register:log.Info('當前有客戶端進行注冊')room.clientsPool[c] = truecase c := <-room.unregister:log.Info('當前有客戶端離開')if room.clientsPool[c] {close(c.send)delete(room.clientsPool, c)}case m := <-room.send:for c := range room.clientsPool {select {case c.send <- m:default:break}}}}}結果:
成功使用 channel 替換了鎖。
聯(lián)系客服
