前言
最近在用 Go 寫業(yè)務(wù)的時(shí)碰到了并發(fā)更新數(shù)據(jù)的場(chǎng)景，由于該業(yè)務(wù)并發(fā)度不高，只是為了防止出現(xiàn)并發(fā)時(shí)數(shù)據(jù)異常。
所以自然就想到了樂(lè)觀鎖的解決方案。
實(shí)現(xiàn)
樂(lè)觀鎖的實(shí)現(xiàn)比較簡(jiǎn)單，相信大部分有數(shù)據(jù)庫(kù)使用經(jīng)驗(yàn)的都能想到。
UPDATE `table` SET `amount`=100,`version`=version+1 WHERE `version` = 1 AND `id` = 1
需要在表中新增一個(gè)類似于 version 的字段，本質(zhì)上我們只是執(zhí)行這段 SQL，在更新時(shí)比較當(dāng)前版本與數(shù)據(jù)庫(kù)版本是否一致。
如上圖所示：版本一致則更新成功，并且將版本號(hào)+1；如果不一致則認(rèn)為出現(xiàn)并發(fā)沖突，更新失敗。
這時(shí)可以直接返回失敗，讓業(yè)務(wù)重試；當(dāng)然也可以再次獲取最新數(shù)據(jù)進(jìn)行更新嘗試。
我們使用的是 gorm 這個(gè) orm 庫(kù)，不過(guò)我查閱了官方文檔卻沒(méi)有發(fā)現(xiàn)樂(lè)觀鎖相關(guān)的支持，看樣子后續(xù)也不打算提供實(shí)現(xiàn)。
不過(guò)借助 gorm 實(shí)現(xiàn)也很簡(jiǎn)單：
type Optimistic struct {
Id int64 `gorm:"column:id;primary_key;AUTO_INCREMENT" json:"id"`
UserId string `gorm:"column:user_id;default:0;NOT NULL" json:"user_id"` // 用戶ID
Amount float32 `gorm:"column:amount;NOT NULL" json:"amount"` // 金額
Version int64 `gorm:"column:version;default:0;NOT NULL" json:"version"` // 版本
}
func TestUpdate(t *testing.T) {
dsn := "root:abc123@/test?charset=utf8&parseTime=True&loc=Local"
db, err := gorm.Open(mysql.Open(dsn), &gorm.Config{})
var out Optimistic
db.First(&out, Optimistic{Id: 1})
out.Amount = out.Amount + 10
column := db.Model(&out).Where("id", out.Id).Where("version", out.Version).
UpdateColumn("amount", out.Amount).
UpdateColumn("version", gorm.Expr("version+1"))
fmt.Printf("#######update %v line \n", column.RowsAffected)
}
這里我們創(chuàng)建了一張 t_optimistic 表用于測(cè)試，生成的 SQL 也滿足樂(lè)觀鎖的要求。
不過(guò)考慮到這類業(yè)務(wù)的通用性，每次需要樂(lè)觀鎖更新時(shí)都需要這樣硬編碼并不太合適。對(duì)于業(yè)務(wù)來(lái)說(shuō)其實(shí) version 是多少壓根不需要關(guān)心，只要能滿足并發(fā)更新時(shí)的準(zhǔn)確性即可。
因此我做了一個(gè)封裝，最終使用如下：
var out Optimistic
db.First(&out, Optimistic{Id: 1})
out.Amount = out.Amount + 10
if err = UpdateWithOptimistic(db, &out, nil, 0, 0); err != nil {
fmt.Printf("%+v \n", err)
}
- 這里的使用場(chǎng)景是每次更新時(shí)將
amount金額加上10。
這樣只會(huì)更新一次，如果更新失敗會(huì)返回一個(gè)異常。
當(dāng)然也支持更新失敗時(shí)執(zhí)行一個(gè)回調(diào)函數(shù)，在該函數(shù)中實(shí)現(xiàn)對(duì)應(yīng)的業(yè)務(wù)邏輯，同時(shí)會(huì)使用該業(yè)務(wù)邏輯嘗試更新 N 次。
func BenchmarkUpdateWithOptimistic(b *testing.B) {
dsn := "root:abc123@/test?charset=utf8&parseTime=True&loc=Local"
db, err := gorm.Open(mysql.Open(dsn), &gorm.Config{})
if err != nil {
fmt.Println(err)
return
}
b.RunParallel(func(pb *testing.PB) {
var out Optimistic
db.First(&out, Optimistic{Id: 1})
out.Amount = out.Amount + 10
err = UpdateWithOptimistic(db, &out, func(model Lock) Lock {
bizModel := model.(*Optimistic)
bizModel.Amount = bizModel.Amount + 10
return bizModel
}, 3, 0)
if err != nil {
fmt.Printf("%+v \n", err)
}
})
}
以上代碼的目的是：
將 amount 金額 +10，失敗時(shí)再次依然將金額+10，嘗試更新 3 次；經(jīng)過(guò)上述的并行測(cè)試，最終查看數(shù)據(jù)庫(kù)確認(rèn)數(shù)據(jù)并沒(méi)有發(fā)生錯(cuò)誤。
面向接口編程
下面來(lái)看看具體是如何實(shí)現(xiàn)的；其實(shí)真正核心的代碼也比較少：
func UpdateWithOptimistic(db *gorm.DB, model Lock, callBack func(model Lock) Lock, retryCount, currentRetryCount int32) (err error) {
if currentRetryCount > retryCount {
return errors.WithStack(NewOptimisticError("Maximum number of retries exceeded:" + strconv.Itoa(int(retryCount))))
}
currentVersion := model.GetVersion()
model.SetVersion(currentVersion + 1)
column := db.Model(model).Where("version", currentVersion).UpdateColumns(model)
affected := column.RowsAffected
if affected == 0 {
if callBack == nil && retryCount == 0 {
return errors.WithStack(NewOptimisticError("Concurrent optimistic update error"))
}
time.Sleep(100 * time.Millisecond)
db.First(model)
bizModel := callBack(model)
currentRetryCount++
err := UpdateWithOptimistic(db, bizModel, callBack, retryCount, currentRetryCount)
if err != nil {
return err
}
}
return column.Error
}
具體步驟如下：
- 判斷重試次數(shù)是否達(dá)到上限。
- 獲取當(dāng)前更新對(duì)象的版本號(hào)，將當(dāng)前版本號(hào) +1。
- 根據(jù)版本號(hào)條件執(zhí)行更新語(yǔ)句。
- 更新成功直接返回。
- 更新失敗
affected == 0時(shí)，執(zhí)行重試邏輯。- 重新查詢?cè)搶?duì)象的最新數(shù)據(jù)，目的是獲取最新版本號(hào)。
- 執(zhí)行回調(diào)函數(shù)。
- 從回調(diào)函數(shù)中拿到最新的業(yè)務(wù)數(shù)據(jù)。
- 遞歸調(diào)用自己執(zhí)行更新，直到重試次數(shù)達(dá)到上限。
這里有幾個(gè)地方值得說(shuō)一下；由于 Go 目前還不支持泛型，所以我們?nèi)绻胍@取 struct 中的 version 字段只能通過(guò)反射。
考慮到反射的性能損耗以及代碼的可讀性，有沒(méi)有更”優(yōu)雅“的實(shí)現(xiàn)方式呢？
于是我定義了一個(gè) interface:
type Lock interface {
SetVersion(version int64)
GetVersion() int64
}
其中只有兩個(gè)方法，目的則是獲取 struct 中的 version 字段；所以每個(gè)需要樂(lè)觀鎖的 struct 都得實(shí)現(xiàn)該接口，類似于這樣：
func (o *Optimistic) GetVersion() int64 {
return o.Version
}
func (o *Optimistic) SetVersion(version int64) {
o.Version = version
}
這樣還帶來(lái)了一個(gè)額外的好處：
一旦該結(jié)構(gòu)體沒(méi)有實(shí)現(xiàn)接口，在樂(lè)觀鎖更新時(shí)編譯器便會(huì)提前報(bào)錯(cuò)，如果使用反射只能是在運(yùn)行期間才能進(jìn)行校驗(yàn)。
所以這里在接收數(shù)據(jù)庫(kù)實(shí)體的便可以是 Lock 接口，同時(shí)獲取和重新設(shè)置 version 字段也是非常的方便。
currentVersion := model.GetVersion()
model.SetVersion(currentVersion + 1)
類型斷言
當(dāng)并發(fā)更新失敗時(shí)affected == 0，便會(huì)回調(diào)傳入進(jìn)來(lái)的回調(diào)函數(shù)，在回調(diào)函數(shù)中我們需要實(shí)現(xiàn)自己的業(yè)務(wù)邏輯。
err = UpdateWithOptimistic(db, &out, func(model Lock) Lock {
bizModel := model.(*Optimistic)
bizModel.Amount = bizModel.Amount + 10
return bizModel
}, 2, 0)
if err != nil {
fmt.Printf("%+v \n", err)
}
但由于回調(diào)函數(shù)的入?yún)⒅荒苤朗且粋€(gè) Lock 接口，并不清楚具體是哪個(gè) struct，所以在執(zhí)行業(yè)務(wù)邏輯之前需要將這個(gè)接口轉(zhuǎn)換為具體的 struct。
這其實(shí)和 Java 中的父類向子類轉(zhuǎn)型非常類似，必須得是強(qiáng)制類型轉(zhuǎn)換，也就是說(shuō)運(yùn)行時(shí)可能會(huì)出問(wèn)題。
在 Go 語(yǔ)言中這樣的行為被稱為類型斷言；雖然叫法不同，但目的類似。其語(yǔ)法如下：
x.(T)
x:表示 interface
T:表示 向下轉(zhuǎn)型的具體 struct
所以在回調(diào)函數(shù)中得根據(jù)自己的需要將 interface 轉(zhuǎn)換為自己的 struct，這里得確保是自己所使用的 struct ，因?yàn)槭菑?qiáng)制轉(zhuǎn)換，編譯器無(wú)法幫你做校驗(yàn)，具體能否轉(zhuǎn)換成功得在運(yùn)行時(shí)才知道。
總結(jié)
有需要的朋友可以在這里獲取到源碼及具體使用方式:
https://github.com/crossoverJie/gorm-optimistic
最近工作中使用了幾種不同的編程語(yǔ)言，會(huì)發(fā)現(xiàn)除了語(yǔ)言自身的語(yǔ)法特性外大部分知識(shí)點(diǎn)都是相同的；
比如面向?qū)ο?、?shù)據(jù)庫(kù)、IO操作等；所以掌握了這些基本知識(shí)，學(xué)習(xí)其他語(yǔ)言自然就能觸類旁通了。
