Go并發(fā)編程 — I/O聚合優(yōu)化（動(dòng)畫(huà)講解）
背景提要
在存儲(chǔ)系統(tǒng)中，在確保功能不受損的前提下，盡量的減少讀寫(xiě)I/O的次數(shù)是優(yōu)化的一個(gè)重要方向，也就是聚合I/O的場(chǎng)景。讀寫(xiě)操作雖然都有聚合I/O的需求，但各自的重點(diǎn)和實(shí)現(xiàn)方法卻有所不同。接下來(lái)，我們將分別探討讀和寫(xiě)請(qǐng)求的聚合優(yōu)化方法。
讀請(qǐng)求的聚合
以讀操作中，緩存優(yōu)化是一種常見(jiàn)的優(yōu)化手段。具體做法是將讀取的數(shù)據(jù)存儲(chǔ)在內(nèi)存中，并通過(guò)一個(gè)唯一的Key來(lái)索引這些數(shù)據(jù)。當(dāng)讀請(qǐng)求來(lái)到時(shí)，如果該Key在緩存中沒(méi)有命中，那么就需要從后端存儲(chǔ)獲取。用戶請(qǐng)求直接穿透到后端存儲(chǔ)，如果并發(fā)很大，這可能是一個(gè)很大的風(fēng)險(xiǎn)。
例如，對(duì)于 Key：“test”，如果緩存中沒(méi)有相應(yīng)的數(shù)據(jù)，并且突然出現(xiàn)大量并發(fā)讀取請(qǐng)求，每個(gè)請(qǐng)求都會(huì)發(fā)現(xiàn)緩存未命中。如果這些請(qǐng)求全部直接訪問(wèn)后端存儲(chǔ)，可能會(huì)給后端存儲(chǔ)帶來(lái)巨大壓力。
為了應(yīng)對(duì)這種情況，我們其實(shí)可以只允許一個(gè)讀請(qǐng)求去后端讀取數(shù)據(jù)，而其他并發(fā)請(qǐng)求則等待這個(gè)請(qǐng)求的結(jié)果。這就是讀請(qǐng)求聚合的基本原理。
在Go語(yǔ)言中，可以使用singleflight 這類第三方庫(kù)完成上述需求。singleflight的設(shè)計(jì)理念是“單一請(qǐng)求執(zhí)行”，即針對(duì)同一個(gè)Key，在多個(gè)并發(fā)請(qǐng)求中只允許一個(gè)請(qǐng)求訪問(wèn)后端。
01 - 讀請(qǐng)求聚合的使用姿勢(shì)
下面是一個(gè)使用 singleflight 的示例，展現(xiàn)了如何通過(guò)傳入特定的Key和閉包函數(shù)來(lái)聚合并發(fā)請(qǐng)求。
package main
import (
// ...
"golang.org/x/sync/singleflight"
)
func main() {
var g singleflight.Group
var wg sync.WaitGroup
// 模擬多個(gè) goroutine 并發(fā)請(qǐng)求相同的資源
for i := 0; i < 5; i++ {
wg.Add(1)
go func(idx int) {
defer wg.Done()
v, err, shared := g.Do("objectkey", func() (interface{}, error) {
fmt.Printf("協(xié)程ID:%v 正在執(zhí)行...\n", idx)
time.Sleep(2 * time.Second)
return "objectvalue", nil
})
if err != nil {
log.Fatalf("err:%v", err)
}
fmt.Printf("協(xié)程ID:%v 請(qǐng)求結(jié)果: %v, 是否共享結(jié)果: %v\n", idx, v, shared)
}(i)
}
wg.Wait()
}在這個(gè)例子中，多個(gè)Goroutine并發(fā)地請(qǐng)求Key為“objectkey”的資源。通過(guò)singleflight，我們確保只有一個(gè)Goroutine去執(zhí)行實(shí)際的數(shù)據(jù)加載操作，而其他請(qǐng)求則等待這個(gè)操作的結(jié)果。接下來(lái)，我們將探討 singleflight 的原理。
02 - singleflight的原理
singleflight 庫(kù)提供了一個(gè)Group結(jié)構(gòu)體，用于管理不同的請(qǐng)求，意圖在內(nèi)部實(shí)現(xiàn)聚合的效果。定義如下：
type Group struct {
mu sync.Mutex // 互斥鎖，包含下面的映射表
m map[string]*call // 正在執(zhí)行請(qǐng)求的映射表
}Group結(jié)構(gòu)的核心就是這個(gè)map結(jié)構(gòu)。每個(gè)正在執(zhí)行的請(qǐng)求被封裝在 call 結(jié)構(gòu)中，定義如下：
type call struct {
wg sync.WaitGroup // 用于同步并發(fā)的請(qǐng)求
val interface{} // 用于存放執(zhí)行的結(jié)果
err error // 存放執(zhí)行的結(jié)果
dups int // 用于計(jì)數(shù)聚合的請(qǐng)求
// ...其他字段用于處理特殊情況和提高容錯(cuò)性
}Group結(jié)構(gòu)的Do方法實(shí)現(xiàn)了聚合去重的核心邏輯，代碼實(shí)現(xiàn)如下所示：
func (g *Group) Do(key string, fn func() (interface{}, error)) (v interface{}, err error, shared bool) {
g.mu.Lock()
if g.m == nil {
g.m = make(map[string]*call)
}
// 用 map 結(jié)構(gòu)，來(lái)判斷是否已經(jīng)有對(duì)應(yīng) Key 正在執(zhí)行的請(qǐng)求
if c, ok := g.m[key]; ok {
c.dups++
// 如果有對(duì)應(yīng) Key 的請(qǐng)求正在執(zhí)行，那么等待結(jié)果即可。
g.mu.Unlock()
c.wg.Wait()
// ...
return c.val, c.err, true
}
// 創(chuàng)建一個(gè)代表執(zhí)行請(qǐng)求的結(jié)構(gòu)，和 Key 關(guān)聯(lián)起來(lái)，存入map中
c := new(call)
c.wg.Add(1)
g.m[key] = c
g.mu.Unlock()
g.doCall(c, key, fn) // 真正執(zhí)行請(qǐng)求
return c.val, c.err, c.dups > 0
}
func (g *Group) doCall(c *call, key string, fn func() (interface{}, error)) {
defer func() {
// ...省略異常處理
c.wg.Done()
}()
func() {
// 真正執(zhí)行請(qǐng)求
c.val, c.err = fn()
}()
// ...
}通過(guò)上述代碼，singleflight的Group結(jié)構(gòu)體利用map記錄了正在執(zhí)行的請(qǐng)求，關(guān)聯(lián)了請(qǐng)求的Key和執(zhí)行體。當(dāng)新的請(qǐng)求到來(lái)時(shí)，先檢查是否有相同Key的正在執(zhí)行的請(qǐng)求，如果有，則等待起結(jié)果，從而避免重復(fù)執(zhí)行相同的請(qǐng)求。
動(dòng)畫(huà)示意圖：
圖片
對(duì)于讀操作，singleflight通過(guò)這種方式有效地減少了重復(fù)工作。然而，對(duì)于寫(xiě)操作，處理邏輯會(huì)有所不同，它需要額外的機(jī)制來(lái)保證數(shù)據(jù)落盤(pán)的時(shí)序。
寫(xiě)請(qǐng)求的聚合
我們先回憶一下寫(xiě)操作的姿勢(shì)。首先通過(guò)Write系統(tǒng)調(diào)用來(lái)寫(xiě)入數(shù)據(jù)，默認(rèn)情況下此時(shí)數(shù)據(jù)可能僅駐留在PageCache中，為了確保數(shù)據(jù)安全落盤(pán)，此時(shí)我們需要手動(dòng)調(diào)用一次 Sync 系統(tǒng)調(diào)用。
然而，Sync操作的成本相當(dāng)大，并且它除了數(shù)據(jù)，還會(huì)同步元數(shù)據(jù)等其他信息到磁盤(pán)上。對(duì)于性能影響巨大。并且，在機(jī)械盤(pán)的場(chǎng)景下，串行化的執(zhí)行Sync是更好的實(shí)踐。
因此，我們面臨的一個(gè)問(wèn)題是：如果在不犧牲數(shù)據(jù)安全性的前提下，能否減少Sync的次數(shù)呢？
對(duì)于同一個(gè)文件的寫(xiě)操作，合并Sync操作是可行的。
文件的Sync會(huì)將當(dāng)前時(shí)刻文件在內(nèi)存中的全部數(shù)據(jù)一次性同步到磁盤(pán)。無(wú)論之前執(zhí)行過(guò)多少次Write調(diào)用，一次Sync就能全部刷盤(pán)。這正是聚合寫(xiě)請(qǐng)求以優(yōu)化性能的關(guān)鍵所在。
01 - 寫(xiě)聚合的原理
假設(shè)對(duì)同一個(gè)文件寫(xiě)了三次數(shù)據(jù)，每一次都是Write+Sync的操作。那么在合適的時(shí)機(jī)，三次Sync調(diào)用可以優(yōu)化成一次。如下圖所示：
圖片
請(qǐng)求 C 的 Sync 操作是在所有請(qǐng)求的 Write 之后才發(fā)起的，所以它必定能保證在此之前的所有變更的數(shù)據(jù)都安全落盤(pán)。這就是寫(xiě)操作聚合的根本原理。
接下來(lái)我們來(lái)思考兩個(gè)問(wèn)題。
問(wèn)題一：有童鞋可能會(huì)問(wèn)，讀寫(xiě)聚合優(yōu)化感覺(jué)有一點(diǎn)相似？那能否用 singleflight 聚合寫(xiě)操作呢？
例如，當(dāng)并發(fā)調(diào)用 Sync 的時(shí)候，如果發(fā)現(xiàn)有正在執(zhí)行的Sync，能否共享這次Sync請(qǐng)求呢？
答案是：不可以。使用singleflight來(lái)優(yōu)化寫(xiě)無(wú)法保證數(shù)據(jù)的安全性。
我們必須要保證的是，Sync操作一定要在Write完成之后發(fā)起。只要兩者存在并發(fā)的可能性，那么Sync就不能保證攜帶了這次Write操作的數(shù)據(jù)，也就無(wú)法保證安全性。
示意圖：
圖片
還是以上面的圖為例來(lái)說(shuō)明，當(dāng)請(qǐng)求 B 完成 Write 操作后，看到請(qǐng)求 A 已經(jīng)發(fā)起了 Sync 操作。此時(shí)它是無(wú)法判斷請(qǐng)求 A 的 Sync 操作是否包含了請(qǐng)求 B 的數(shù)據(jù)。從圖示我們也很清晰的看到，請(qǐng)求B的 Write 和請(qǐng)求 A 的 Sync 在時(shí)間上存在重疊。
因此，當(dāng)Write完成后，如果發(fā)現(xiàn)有一個(gè)Sync正在執(zhí)行，我們不能簡(jiǎn)單地復(fù)用這個(gè)Sync。我們需要啟動(dòng)一個(gè)新的Sync操作。
問(wèn)題二：那么聚合的時(shí)機(jī)在哪里呢？
對(duì)于讀請(qǐng)求的聚合，其時(shí)機(jī)相對(duì)直觀：一旦發(fā)現(xiàn)有針對(duì)同一個(gè) Key 的請(qǐng)求，就可以等待這次的結(jié)果并復(fù)用該結(jié)果。但寫(xiě)請(qǐng)求的聚合時(shí)機(jī)則不是，它的聚合時(shí)機(jī)是在等待中遇到“志同道合“的請(qǐng)求。
讓我們通過(guò)一個(gè)具體例子來(lái)說(shuō)明（注意，以下所有的請(qǐng)求都是針對(duì)相同的文件）：
- t0 時(shí)刻：A 執(zhí)行了 Write，并嘗試發(fā)起Sync，由于此時(shí)沒(méi)有其他請(qǐng)求在執(zhí)行，A 便執(zhí)行真正的Sync操作。
- t1 時(shí)刻：B 執(zhí)行了 Write，發(fā)現(xiàn)已經(jīng)有請(qǐng)求在Sync了（即A），因此進(jìn)入等待狀態(tài)，直到A完成。
- t2 時(shí)刻：C 執(zhí)行了 Write，發(fā)現(xiàn)已經(jīng)有請(qǐng)求在Sync了（即A），因此進(jìn)入等待狀態(tài)，直到A完成。
- t3 時(shí)刻：D 執(zhí)行了 Write，發(fā)現(xiàn)已經(jīng)有請(qǐng)求在Sync了（即A），因此進(jìn)入等待狀態(tài)，直到A完成。
- t4 時(shí)刻：A 的Sync操作終于完成。A隨即通知 B、C、D 三位，告知它們可以進(jìn)行Sync請(qǐng)求了。
- t5 時(shí)刻：從B、C、D中選擇一個(gè)來(lái)執(zhí)行一次Sync操作。假設(shè)B被選中，則C、D請(qǐng)求則等待B完成Sync即可。B發(fā)起的Sync操作一定包含了B，C，D三者寫(xiě)的數(shù)據(jù)，確保了安全性。
- t6：B 的Sync操作完成，C、D被通知操作已完成。如此一來(lái)，B、C、D三者的數(shù)據(jù)都確保落盤(pán)。
正如上述所演示，寫(xiě)操作的聚合是在等待前一次Sync操作完成期間收集到的請(qǐng)求。本來(lái)需要4次Sync操作，現(xiàn)在僅需2次Sync就可以確保數(shù)據(jù)的安全性。
在高并發(fā)的場(chǎng)景下，這種聚合方式的效益尤為顯著。下面，我們將探討這種策略的具體代碼實(shí)現(xiàn)。
02 - 寫(xiě)聚合的代碼實(shí)現(xiàn)
實(shí)現(xiàn)寫(xiě)操作聚合的關(guān)鍵在于確保數(shù)據(jù)安全的時(shí)序前提下進(jìn)行聚合。以下是一種典型和實(shí)現(xiàn)方式，它是對(duì) sync.Cond 和 sync.Once 的巧妙應(yīng)用。首先，我們定義一個(gè)負(fù)責(zé)聚合的結(jié)構(gòu)體，如下：
// SyncJob 用于管理一個(gè)文件的 Sync 任務(wù)
type SyncJob struct {
*sync.Cond // 聚合 Sync 的關(guān)鍵
holding int32 // 記錄聚合的個(gè)數(shù)
lastErr error // 記錄執(zhí)行 Sync 結(jié)果
syncPoint *sync.Once // 確保同一時(shí)間只有一個(gè) Sync 執(zhí)行
syncFunc func(interface{}) error // 實(shí)際執(zhí)行 Sync 的函數(shù)
}
// SyncJob 的構(gòu)建函數(shù)
func NewSyncJob(fn func(interface{}) error) *SyncJob {
return &SyncJob{
Cond: sync.NewCond(&sync.Mutex{}),
syncFunc: fn,
syncPoint: &sync.Once{},
}
}接下來(lái)，我們?yōu)?SyncJob 定義一個(gè)執(zhí)行聚合的方法，如下：
func (s *SyncJob) Do(job interface{}) error {
s.L.Lock()
if s.holding > 0 {
// 如果有請(qǐng)求在前面，則等待前一次請(qǐng)求完成。
// 等待的過(guò)程中，會(huì)有"志同道合"之人
s.Wait()
}
// 準(zhǔn)備要下發(fā)請(qǐng)求了，增加計(jì)數(shù)
s.holding += 1
syncPoint := s.syncPoint
s.L.Unlock()
// "志同道合"的人一起來(lái)到這里，此時(shí)已經(jīng)滿足 Write 和 Sync 的時(shí)序關(guān)系。
// 使用 sync.Once 確保只有請(qǐng)求者執(zhí)行同步操作。
syncPoint.Do(func() {
// 執(zhí)行實(shí)際的 Sync 操作
s.lastErr = s.syncFunc(job)
s.L.Lock()
// holding 展示本批次有多少個(gè)請(qǐng)求
fmt.Printf("holding:%v\n", s.holding)
// 本次請(qǐng)求執(zhí)行完成，重置計(jì)數(shù)器，準(zhǔn)備下一輪聚合
s.holding = 0
s.syncPoint = &sync.Once{}
// 喚醒下一批的請(qǐng)求
s.Broadcast()
s.L.Unlock()
})
return s.lastErr
}在這里，我們使用了一個(gè)Go的 sync.Cond 來(lái)阻塞和通知等待中的請(qǐng)求，并通過(guò) sync.Once 確保同步操作同一時(shí)間、同一批只有一個(gè)在執(zhí)行。
- 其實(shí)在這個(gè)場(chǎng)景下，從代碼實(shí)現(xiàn)來(lái)講，sync.Cond 也可以使用 Go 的 Channel 來(lái)實(shí)現(xiàn)相同的效果，用 Ch← 來(lái)阻塞，用 close(Ch) 來(lái)通知。效果是一樣的，感興趣的童鞋可以改造試試。
現(xiàn)在讓我們來(lái)看看這段代碼的實(shí)際運(yùn)行效果：
func main() {
file, err := os.OpenFile("hello.txt", os.O_RDWR, 0700)
if err != nil {
log.Fatal(err)
}
defer file.Close()
// 初始化 Sync 聚合服務(wù)
syncJob := NewSyncJob(func(interface{}) error {
fmt.Printf("do sync...\n")
time.Sleep(time.Second())
return file.Sync()
})
wg := sync.WaitGroup{}
for i := 0; i < 10; i++ {
wg.Add(1)
go func() {
defer wg.Done()
// 執(zhí)行寫(xiě)操作 write ...
fmt.Printf("write...\n")
// 觸發(fā) sync 操作
syncJob.Do(file)
}()
}
wg.Wait()
}通過(guò)上述代碼，我們講對(duì)文件寫(xiě)入操作后的 Sync 調(diào)用進(jìn)行有效的聚合。童鞋們可以多次運(yùn)行程序，觀察其行為。可以通過(guò)觀察打印的 holding 字段獲悉每一批聚合的請(qǐng)求是多少個(gè)。
思考：從效果來(lái)講，上面的代碼無(wú)論怎么跑，最少要執(zhí)行兩次 Sync。你知道是為什么嗎？
動(dòng)畫(huà)示意圖：
圖片
總結(jié)
上面介紹了讀寫(xiě)聚合優(yōu)化的兩種實(shí)現(xiàn)。讀和寫(xiě)的聚合是有區(qū)別的。
- 讀操作，核心是一個(gè) map，只要有相同Key的讀取正在執(zhí)行，那么等待這份正在執(zhí)行的請(qǐng)求的結(jié)果也是符合預(yù)期的。同步等待則用的是 sync.WaitGroup 來(lái)實(shí)現(xiàn)。
- 寫(xiě)操作，核心是要先保證數(shù)據(jù)安全性。它必須保證 Sync 操作在 Write 操作之后。因此當(dāng)發(fā)現(xiàn)有正在執(zhí)行的Sync操作，那么就等待這次完成，然后必須重新開(kāi)啟一輪的 Sync 操作，等待的過(guò)程也是聚合的時(shí)機(jī)。我們可以使用 sync.Cond（或者 Channel ）來(lái)實(shí)現(xiàn)阻塞和喚醒，使用 sync.Once 來(lái)保證同一時(shí)間單個(gè)執(zhí)行。
