技術標籤：佇列csvbbselasticsearchhttp

前言

上週六馬上就下班了，正興高采烈的想著下班吃什麼呢！突然QA找到我，說我們的DB與es無法同步資料了，真是令人頭皮發禿，好不容易休一天，啊啊啊，難受呀，沒辦法，還是趕緊找bug吧。下面我就把我這次的bug原因分享給大家，避免踩坑～。

bug原因之bulk隱藏錯誤資訊

第一時間，我去看了一下錯誤日誌，竟然沒有錯誤日誌，很是神奇，既然這樣，那我們就DEBUG一下吧，DEBUG之前我先貼一段程式碼：

func(es*UserES)batchAdd(ctxcontext.Context,user[]*model.UserEs)error{
req:=es.client.Bulk().Index(es.index)
for_,u:=rangeuser{
u.UpdateTime=uint64(time.Now().UnixNano())/uint64(time.Millisecond)
u.CreateTime=uint64(time.Now().UnixNano())/uint64(time.Millisecond)
doc:=elastic.NewBulkIndexRequest().Id(strconv.FormatUint(u.ID,10)).Doc(u)
req.Add(doc)
}
ifreq.NumberOfActions()<0{
returnnil
}
if_,err:=req.Do(ctx);err!=nil{
returnerr
}
returnnil
}
 

就是上面這段程式碼，使用es的bulk批量操作，經過DEBUG仍然沒有發現任何問題，臥槽！！！沒有頭緒了，那就看一看es原始碼吧，裡面是不是有什麼隱藏的點沒有注意到。還真被我找到了，我們先看一下req.Do(ctx)的實現：

//DosendsthebatchedrequeststoElasticsearch.Notethat,whensuccessful,
//youcanreusetheBulkServiceforthenextbatchasthelistofbulk
//requestsisclearedonsuccess.
func(s*BulkService)Do(ctxcontext.Context)(*BulkResponse,error){
/**
......省略部分程式碼
**/
//Getresponse
res,err:=s.client.PerformRequest(ctx,PerformRequestOptions{
Method:"POST",
Path:path,
Params:params,
Body:body,
ContentType:"application/x-ndjson",
Retrier:s.retrier,
Headers:s.headers,
})
iferr!=nil{
returnnil,err
}

//Returnresults
ret:=new(BulkResponse)
iferr:=s.client.decoder.Decode(res.Body,ret);err!=nil{
returnnil,err
}

//Resetsotherequestcanbereused
s.Reset()

returnret,nil
}
 

我只把重要部分程式碼貼出來，看這一段就好了，我來解釋一下：

• 首先構建Http請求

• 傳送Http請求並分析，並解析response

• 重置request可以重複使用

這裡的重點就是ret := new(BulkResponse)，new了一個BulkResponse結構，他的結構如下：

typeBulkResponsestruct{
Tookint`json:"took,omitempty"`
Errorsbool`json:"errors,omitempty"`
Items[]map[string]*BulkResponseItem`json:"items,omitempty"`
}
//BulkResponseItemistheresultofasinglebulkrequest.
typeBulkResponseItemstruct{
Indexstring`json:"_index,omitempty"`
Typestring`json:"_type,omitempty"`
Idstring`json:"_id,omitempty"`
Versionint64`json:"_version,omitempty"`
Resultstring`json:"result,omitempty"`
Shards*ShardsInfo`json:"_shards,omitempty"`
SeqNoint64`json:"_seq_no,omitempty"`
PrimaryTermint64`json:"_primary_term,omitempty"`
Statusint`json:"status,omitempty"`
ForcedRefreshbool`json:"forced_refresh,omitempty"`
Error*ErrorDetails`json:"error,omitempty"`
GetResult*GetResult`json:"get,omitempty"`
}
 

先來解釋一個每個欄位的意思：

• took：總共耗費了多長時間，單位是毫秒

• Errors：如果其中任何子請求失敗，該 errors 標誌被設定為 true ，並且在相應的請求報告出錯誤明細（看下面的Items解釋）

• Items：這個裡就是儲存每一個子請求的response，這裡的Error儲存的是詳細的錯誤資訊

現在我想大家應該知道為什麼我們的程式碼沒有報err資訊了，bulk的每個請求都是獨立的執行，因此某個子請求的失敗不會對其他子請求的成功與否造成影響，所以其中某一條出現錯誤我們需要從BulkResponse解出來。現在我們把程式碼改正確：

func(es*UserES)batchAdd(ctxcontext.Context,user[]*model.UserEs)error{
req:=es.client.Bulk().Index(es.index)
for_,u:=rangeuser{
u.UpdateTime=uint64(time.Now().UnixNano())/uint64(time.Millisecond)
u.CreateTime=uint64(time.Now().UnixNano())/uint64(time.Millisecond)
doc:=elastic.NewBulkIndexRequest().Id(strconv.FormatUint(u.ID,10)).Doc(u)
req.Add(doc)
}
ifreq.NumberOfActions()<0{
returnnil
}
res,err:=req.Do(ctx)
iferr!=nil{
returnerr
}
//任何子請求失敗，該`errors`標誌被設定為`true`，並且在相應的請求報告出錯誤明細
//所以如果沒有出錯，說明全部成功了，直接返回即可
if!res.Errors{
returnnil
}
for_,it:=rangeres.Failed(){
ifit.Error==nil{
continue
}
return&elastic.Error{
Status:it.Status,
Details:it.Error,
}
}
returnnil
}

這裡再解釋一下res.Failed方法，這裡會把items中bulk response帶錯誤的返回，所以在這裡面找錯誤資訊就可以了。

至此，這個bug原因終於被我找到了，接下來可以看下一個bug了，我們先簡單總結一下：

bulk API 允許在單個步驟中進行多次 create 、 index 、 update 或 delete 請求，每個子請求都是獨立執行，因此某個子請求的失敗不會對其他子請求的成功與否造成影響。bulk的response結構中Erros欄位，如果其中任何子請求失敗，該 errors 標誌被設定為 true ，並且在相應的請求報告出錯誤明細，items欄位是一個數組，，這個陣列的內容是以請求的順序列出來的每個請求的結果。所以在使用bulk時一定要從response中判斷是否有err。

bug原因之數值範圍越界

這裡完全是自己使用不當造成，但還是想說一說es的對映數字類型範圍的問題：

數字型別有如下分類:

這裡都是有符號型別的，無符號在es7.10.1版本才開始支援，有興趣的同學戳這裡。

這裡把這些數字型別及範圍列出來就是方便說我的bug原因，這裡直接解釋一下：

我在DB設定欄位的型別是tinyint unsigned，tinyint是一個位元組儲存，無符號的話範圍是0-255，而我在es中對映型別選擇的是byte，範圍是-128~127，當DB中數值超過這個範圍是，在進行同步時就會出現這個問題，這裡需要大家注意一下數值範圍的問題，不要像我一樣，因為這個還排查了好久的bug，有些空間沒必要省，反正也佔不了多少空間。

總結

這篇文章就是簡單總結一下我在工作中遇到的問題，發表出來就是給大家提個醒，有人踩過的坑，就不要在踩了，浪費時間！！！！

好啦，這篇文章就到這裡啦，素質三連（分享、點贊、在看）都是筆者持續創作更多優質內容的動力！

結尾給大家發一個小福利吧，最近我在看[微服務架構設計模式]這一本書，講的很好，自己也收集了一本PDF，有需要的小夥可以到自行下載。獲取方式：關注公眾號：[Golang夢工廠]，後臺回覆：[微服務]，即可獲取。

我翻譯了一份GIN中文文件，會定期進行維護，有需要的小夥伴後臺回覆[gin]即可下載。

翻譯了一份Machinery中文文件，會定期進行維護，有需要的小夥伴們後臺回覆[machinery]即可獲取。

我是asong，一名普普通通的程式猿，讓gi我一起慢慢變強吧。歡迎各位的關注，我們下期見~~~

推薦往期文章：

• machinery-go非同步任務佇列

• Leaf—Segment分散式ID生成系統（Golang實現版本）

• 十張動圖帶你搞懂排序演算法(附go實現程式碼)

• Go語言相關書籍推薦（從入門到放棄）

• go引數傳遞型別

• 手把手教姐姐寫訊息佇列

• 常見面試題之快取雪崩、快取穿透、快取擊穿

• 詳解Context包，看這一篇就夠了！！！

• go-ElasticSearch入門看這一篇就夠了(一)

• 面試官：go中for-range使用過嗎？這幾個問題你能解釋一下原因嗎