第golang并發(fā)工具M(jìn)apReduce降低服務(wù)響應(yīng)時(shí)間目錄前言并發(fā)處理工具M(jìn)apReduceMapReduce的用法演示MapReduce使用注意事項(xiàng)實(shí)現(xiàn)原理分析:文末

前言

在微服務(wù)中開(kāi)發(fā)中，api網(wǎng)關(guān)扮演對(duì)外提供restfulapi的角色，而api的數(shù)據(jù)往往會(huì)依賴其他服務(wù)，復(fù)雜的api更是會(huì)依賴多個(gè)甚至數(shù)十個(gè)服務(wù)。雖然單個(gè)被依賴服務(wù)的耗時(shí)一般都比較低，但如果多個(gè)服務(wù)串行依賴的話那么整個(gè)api的耗時(shí)將會(huì)大大增加。

那么通過(guò)什么手段來(lái)優(yōu)化呢？我們首先想到的是通過(guò)并發(fā)來(lái)的方式來(lái)處理依賴，這樣就能降低整個(gè)依賴的耗時(shí)，Go基礎(chǔ)庫(kù)中為我們提供了WaitGroup工具用來(lái)進(jìn)行并發(fā)控制，但實(shí)際業(yè)務(wù)場(chǎng)景中多個(gè)依賴如果有一個(gè)出錯(cuò)我們期望能立即返回而不是等所有依賴都執(zhí)行完再返回結(jié)果，而且WaitGroup中對(duì)變量的賦值往往需要加鎖，每個(gè)依賴函數(shù)都需要添加Add和Done對(duì)于新手來(lái)說(shuō)比較容易出錯(cuò)

基于以上的背景，go-zero框架中為我們提供了并發(fā)處理工具M(jìn)apReduce，該工具開(kāi)箱即用，不需要做什么初始化，我們通過(guò)下圖看下使用MapReduce和沒(méi)使用的耗時(shí)對(duì)比:

相同的依賴，串行處理的話需要200ms，使用MapReduce后的耗時(shí)等于所有依賴中最大的耗時(shí)為100ms，可見(jiàn)MapReduce可以大大降低服務(wù)耗時(shí)，而且隨著依賴的增加效果就會(huì)越明顯，減少處理耗時(shí)的同時(shí)并不會(huì)增加服務(wù)器壓力

并發(fā)處理工具M(jìn)apReduce

MapReduce是Google提出的一個(gè)軟件架構(gòu)，用于大規(guī)模數(shù)據(jù)集的并行運(yùn)算，go-zero中的MapReduce工具正是借鑒了這種架構(gòu)思想

go-zero框架中的MapReduce工具主要用來(lái)對(duì)批量數(shù)據(jù)進(jìn)行并發(fā)的處理，以此來(lái)提升服務(wù)的性能

MapReduce的用法演示

MapReduce主要有三個(gè)參數(shù)，第一個(gè)參數(shù)為generate用以生產(chǎn)數(shù)據(jù)，第二個(gè)參數(shù)為mapper用以對(duì)數(shù)據(jù)進(jìn)行處理，第三個(gè)參數(shù)為reducer用以對(duì)mapper后的數(shù)據(jù)做聚合返回，還可以通過(guò)opts選項(xiàng)設(shè)置并發(fā)處理的線程數(shù)量

場(chǎng)景一:

某些功能的結(jié)果往往需要依賴多個(gè)服務(wù)，比如商品詳情的結(jié)果往往會(huì)依賴用戶服務(wù)、庫(kù)存服務(wù)、訂單服務(wù)等等，一般被依賴的服務(wù)都是以rpc的形式對(duì)外提供，為了降低依賴的耗時(shí)我們往往需要對(duì)依賴做并行處理

funcproductDetail(uid,pidint64)(*ProductDetail,error){

varpdProductDetail

err:=mr.Finish(func()(errerror){

pd.User,err=userRpc.User(uid)

return

},func()(errerror){

pd.Store,err=storeRpc.Store(pid)

return

},func()(errerror){

pd.Order,err=orderRpc.Order(pid)

return

iferr!=nil{

log.Printf("productdetailerror:%v",err)

returnnil,err

returnpd,nil

}

該示例中返回商品詳情依賴了多個(gè)服務(wù)獲取數(shù)據(jù)，因此做并發(fā)的依賴處理，對(duì)接口的性能有很大的提升

場(chǎng)景二:

很多時(shí)候我們需要對(duì)一批數(shù)據(jù)進(jìn)行處理，比如對(duì)一批用戶id，效驗(yàn)每個(gè)用戶的合法性并且效驗(yàn)過(guò)程中有一個(gè)出錯(cuò)就認(rèn)為效驗(yàn)失敗，返回的結(jié)果為效驗(yàn)合法的用戶id

funccheckLegal(uids[]int64)([]int64,error){

r,err:=mr.MapReduce(func(sourcechan-interface{}){

for_,uid:=rangeuids{

source-uid

},func(iteminterface{},writermr.Writer,cancelfunc(error)){

uid:=item.(int64)

ok,err:=check(uid)

iferr!=nil{

cancel(err)

ifok{

writer.Write(uid)

},func(pipe-chaninterface{},writermr.Writer,cancelfunc(error)){

varuids[]int64

forp:=rangepipe{

uids=append(uids,p.(int64))

writer.Write(uids)

iferr!=nil{

log.Printf("checkerror:%v",err)

returnnil,err

returnr.([]int64),nil

funccheck(uidint64)(bool,error){

//dosomethingcheckuserlegal

returntrue,nil

}

該示例中，如果check過(guò)程出現(xiàn)錯(cuò)誤則通過(guò)cancel方法結(jié)束效驗(yàn)過(guò)程，并返回error整個(gè)效驗(yàn)過(guò)程結(jié)束，如果某個(gè)uid效驗(yàn)結(jié)果為false則最終結(jié)果不返回該uid

MapReduce使用注意事項(xiàng)

mapper和reducer中都可以調(diào)用cancel，參數(shù)為error，調(diào)用后立即返回，返回結(jié)果為nil,error

mapper中如果不調(diào)用writer.Write則item最終不會(huì)被reducer聚合

reducer中如果不調(diào)用writer.Wirte則返回結(jié)果為nil,ErrReduceNoOutput

reducer為單線程，所有mapper出來(lái)的結(jié)果在這里串行聚合

實(shí)現(xiàn)原理分析:

MapReduce中首先通過(guò)buildSource方法通過(guò)執(zhí)行g(shù)enerate(參數(shù)為無(wú)緩沖channel)產(chǎn)生數(shù)據(jù)，并返回?zé)o緩沖的channel，mapper會(huì)從該channel中讀取數(shù)據(jù)

funcbuildSource(generateGenerateFunc)chaninterface{}{

source:=make(chaninterface{})

gofunc(){

deferclose(source)

generate(source)

returnsource

}

在MapReduceWithSource方法中定義了cancel方法，mapper和reducer中都可以調(diào)用該方法，調(diào)用后主線程收到close信號(hào)會(huì)立馬返回

cancel:=once(func(errerror){

iferr!=nil{

retErr.Set(err)

}else{

//默認(rèn)的error

retErr.Set(ErrCancelWithNil)

drain(source)

//調(diào)用close(ouput)主線程收到Done信號(hào)，立馬返回

finish()

})

在mapperDispatcher方法中調(diào)用了executeMappers，executeMappers消費(fèi)buildSource產(chǎn)生的數(shù)據(jù)，每一個(gè)item都會(huì)起一個(gè)goroutine單獨(dú)處理，默認(rèn)最大并發(fā)數(shù)為16，可以通過(guò)WithWorkers進(jìn)行設(shè)置

varwgsync.WaitGroup

deferfunc(){

wg.Wait()//保證所有的item都處理完成

close(collector)

pool:=make(chanlang.PlaceholderType,workers)

writer:=newGuardedWriter(collector,done)//將mapper處理完的數(shù)據(jù)寫入collector

for{

select{

case-done://當(dāng)調(diào)用了cancel會(huì)觸發(fā)立即返回

return

casepool-lang.Placeholder://控制最大并發(fā)數(shù)

item,ok:=-input

if!ok{

-pool

return

wg.Add(1)

gofunc(){

deferfunc(){

wg.Done()

-pool

mapper(item,writer)//對(duì)item進(jìn)行處理，處理完調(diào)用writer.Write把結(jié)果寫入collector對(duì)應(yīng)的channel中

}

reducer單goroutine對(duì)數(shù)mapper寫入collector的數(shù)據(jù)進(jìn)行處理，如果reducer中沒(méi)有手動(dòng)調(diào)用writer.Write則最終會(huì)執(zhí)行finish方法對(duì)output進(jìn)行close避免死鎖

gofunc(){

deferfunc(){

ifr:=recover();r!=nil{

cancel(fmt.Errorf("%v",r))

}else{

finish()

reducer(collector,writer,cancel)

}()

在該工具包中還提供了許多針對(duì)不同業(yè)務(wù)場(chǎng)景的方法，實(shí)現(xiàn)原理與MapReduce大同小異，感興趣的同學(xué)可以查看源碼學(xué)習(xí)

MapReduceVoid功能和MapReduce類似但沒(méi)有結(jié)果返回只返回error

Finish處理固定數(shù)量的依賴，返回error，有一個(gè)error立即返回

FinishVoid和Finish方法功能類似，沒(méi)有返回