
問題描述
底下步驟來產生資料
- 建立使用者,並且初始化每人 $1000 USD
- 接到新的交易請求
- 讀取使用者帳戶剩餘存款
- 將該帳號增加 $50 USD
func main() { session, _ := mgo.Dial("localhost:27017") globalDB = session.DB("queue") globalDB.C("bank").DropCollection() user := currency{Account: account, Amount: 1000.00, Code: "USD"} err := globalDB.C("bank").Insert(&user) if err != nil { panic("insert error") } log.Println("Listen server on 8000 port") http.HandleFunc("/", pay) http.ListenAndServe(":8000", nil) }上述是主程式,新增一個 Handle 為 pay,用來處理交易。
func pay(w http.ResponseWriter, r *http.Request) { entry := currency{} // step 1: get current amount err := globalDB.C("bank").Find(bson.M{"account": account}).One(&entry) if err != nil { panic(err) } wait := Random(1, 100) time.Sleep(time.Duration(wait) * time.Millisecond) //step 3: subtract current balance and update back to database entry.Amount = entry.Amount + 50.000 err = globalDB.C("bank").UpdateId(entry.ID, &entry) if err != nil { panic("update error") } fmt.Printf("%+v\n", entry) io.WriteString(w, "ok") }
解決方式
這邊提供幾個解決方式,第一種就是透過sync.Mutex
方式,直接將交易區段程式碼 lock 住,這樣可以避免同時寫入或讀出的問題。在 Handler 內直接新增底下程式碼就可以解決,詳細程式碼請參考 safe.go
mu.Lock() defer mu.Unlock()第二種方式可以用 Go 語言內的優勢: goroutine + channel,在這邊我們只要建立兩個 Channle,第一個是使用者帳號 (string) 第二個是輸出 Result (struct)。完整程式碼範例
in = make(chan string) out = make(chan Result)在 main func 內建立第一個 goroutine
go func(in *chan string) { for { select { case account := <-*in: entry := currency{} // step 1: get current amount err := globalDB.C("bank").Find(bson.M{"account": account}).One(&entry) if err != nil { panic(err) } //step 3: subtract current balance and update back to database entry.Amount = entry.Amount + 50.000 err = globalDB.C("bank").UpdateId(entry.ID, &entry) if err != nil { panic("update error") } out <- Result{ Account: account, Result: entry.Amount, } } } }(&in)上面可以很清楚看到使用到
select
來接受 input channel,並且透過 go
將 for loop 丟到背景執行。所以在每個交易時,將帳號丟到 in
channel 內,就可以開始進行交易,同時間並不會有其他交易。在 handler 內,也是透過此方式來讀取使用者最後存款餘額
wg := sync.WaitGroup{} wg.Add(1) go func(wg *sync.WaitGroup) { in <- account for { select { case result := <-out: fmt.Printf("%+v\n", result) wg.Done() return } } }(&wg) wg.Wait()不過上面這方法,可想而知,只有一個 Queue 幫忙處理交易資料,那假設有幾百萬個交易要同時進行呢,該如何消化更多的交易,就要將上面程式碼改成 Multiple Queue 完整程式碼範例。假設我們有 100 個帳號,開 10 個 Queue 去處理,每一個 Queue 來處理 10 個帳號,也就是說 ID 為 23 號的分給第 3 (23 % 10) 個 Queue,ID 為 59 號則分給第 9 個 Queue。
for i := range in { go func(in *chan string, i int) { for { select { case account := <-*in: out <- Result{ Account: account, Result: entry.Amount, } } } }(&in, i) }其中 channel 要宣告為底下: maxThread 為 10 (可以由開發者任意設定)
in = make([]chan string, maxThread) out = make([]chan Result, maxThread)
Optimistic concurrency control
假設需要擴展服務,執行超過一個服務,就會遇到 Optimistic concurrency control,原因在上述方法只能保證在單一服務內不會同時存取同一筆資料,但是如果是多個服務則還是會發生同時存取或寫入單筆資料。這邊可以用簡單的機制來解決應用層的問題,直接在資料表加上Version
,初始值為 1
,要執行更新時請透過底下語法來更新:
entry.Amount = entry.Amount + 50.000 err = globalDB.C("bank").Update(bson.M{ "version": entry.Version, "_id": entry.ID, }, bson.M{"$set": map[string]interface{}{ "amount": entry.Amount, "version": (entry.Version + 1), }}) if err != nil { goto LOOP }如果資料不存在時,就無法寫入,這樣可以避免同時寫入問題。
效能測試
上述提供了幾種解決方式,但是該選擇哪一種會比較好呢,底下是透過 [vegeta] http 效能檢測工具來實驗看看,底下先整理結果- 使用 sync.Mutex
- 使用 single queue
- 使用 multiple queue
- 使用 optimistic concurrency 解決方案
- 使用 single queue + optimistic concurrency 解決方案
- 使用 multiple queue + optimistic concurrency 解決方案
max Latencies | mean Latencies | user account | |
---|---|---|---|
sync lock | 26.250468944s | 13.171447347s | 1 |
optimistic lock | 5.016707396s | 1.903748023s | 1 |
single queue | 66.078117ms | 763.662µs | 1 |
multiple queue | 49.270982ms | 789.131µs | 100 |
optimistic single queue | 139.045488ms | 1.297197ms | 1 |
optimistic multiple queue | 51.268963ms | 924.951µs | 100 |
multiple queue
,這不是最好的解法,要執行多個服務,請務必使用 optimistic multiple queue