深入解析：如何設計具備交易支援的記憶體鍵值資料庫

在設計具交易支持的內存鍵值數據庫時，並發性和事務處理是兩個主要挑戰。在高併發系統中，如何在不影響性能的情況下確保線程安全，需要謹慎平衡鎖機制。此外，提供事務支持可以讓多個操作以原子性執行，從而避免因競爭條件導致的數據不一致。

那麼，為什麼我們需要事務呢？雖然大部分數據庫都能夠保障**並發安全**，但並非所有都支持**事務**。並發安全意味著多個操作可以同時運行，而不會損壞數據；但事務支持則是確保一系列操作要麼全部成功，要麼完全不應用。這種特性通常被稱為ACID特性，包括原子性、一致性、隔離性與持久性，它們共同確保了資料的完整性與可靠性。

在數據庫中，一個**事務**就是一系列必須作為單一工作單元來處理的操作。例如，在金融交易中，如果其中一步出現錯誤，我們希望整個過程都能回滾，以防止造成資金的不一致。如果沒有這種保障，就可能出現如賬戶餘額錯誤等嚴重問題。

實現這些功能所需的技術材料包括各種鎖定機制或多版本控制。透過這些技術，我們可以在保持高效性能的同時，有效管理複雜的並發場景。因此，對於任何希望構建可靠且高效內存鍵值數據庫的人來說，理解和實施上述概念都是至關重要的。

在一個交易中，如果所有操作都成功，那麼資料庫會將變更**提交**為永久性。如果有任何操作失敗，資料庫則會**回滾**這些變更，以確保一致性。這樣的機制保證了**原子性**，也就是說，交易要麼完全完成，要麼完全撤銷——不應該出現部分更新的情況。

### Redis 事務及其限制

Redis 是一個受歡迎的內存鍵值數據庫，它通過 **MULTI** 和 **EXEC** 命令提供了一定程度上的事務功能。當一個事務開始時，命令會透過 `MULTI` 被排隊進行處理。而 `EXEC` 命令則會按順序執行所有排隊的命令。然而，值得注意的是，Redis 的事務並未提供完整的原子性——如果其中某一條命令失敗，其餘命令還是會繼續執行，而不會回滾之前的變更。

舉例來說，你可以使用以下方式發起一個 Redis 事務：

MULTI

SET key1 "value1"

INCR key2   # 假設 key2 並不是整數，那麼這裡就會失敗

SET key3 "value3"

EXEC

在這種情況下，即使第二條命令因為類型錯誤而失敗，其餘命令仍然會被執行，這便是 Redis 事務的一大局限。因此，在多用戶環境中使用 Redis 時，需要特別考量競爭條件和數據一致性的挑戰。此外，不同資料結構（如列表、集合等）對於支持事務的能力也存在影響，因此選擇適合的資料結構非常重要。

若想克服上述限制，可以考慮使用 Lua 腳本等技術，以提高系統整體效能與可靠性。

如果 `INCR key2` 執行失敗（因為 `key2` 不是一個數字），**Redis 還是會執行 `SET key3 "value3"`**，這樣就會使資料庫處於不一致的狀態。這是因為 Redis 的事務 **缺乏自動回滾機制**。### Redis 中的樂觀鎖：WATCH 為了提升事務的安全性，Redis 引入了通過 `WATCH` 命令實現的 **樂觀鎖**。這個命令 **監控** 某個鍵在執行事務前是否被修改。如果在事務期間該鍵被其他客戶端改動，那麼該事務將 **中止**，避免使用到過時的數據。使用 `WATCH` 的例子如下：

balanceMULTIDECR balance  # 扣除金額EXEC  # 如果 balance 被其他地方修改則會失敗

如果在執行 `EXEC` 前，另一個客戶端更改了 `balance`，那麼這個事務就會失敗，有效防止了不一致的更新。

樂觀鎖基本上是透過維護版本號來確保數據的一致性。在每次讀取和寫入時，它都會檢查所監控鍵的版本號，以判斷該數據是否已經被更改。如果有衝突，就無法完成操作，從而降低了競爭條件發生的可能性。

此外，在高併發環境中，由於減少了對鎖資源的競爭，樂觀鎖通常能提供更好的性能表現。不需要持續佔用資源，可以讓系統保持較高的響應速度。

整體而言，相較於悲觀鎖而言，樂觀鎖具有提高效率及縮短等待時間等優勢，但也可能帶來一定風險，比如頻繁重試可能導致性能下降。因此，在選擇使用哪種方式時，需要根據具體情況進行分析與考量。

### 樂觀鎖的問題
雖然 `WATCH` 有助於維持一致性，但在高併發環境中，這卻引入了**不可預測性**。如果多個客戶端頻繁地修改同一鍵，事務將會不斷**失敗**，導致系統變得容易出錯，因為由於並發更新而使得交易被隨機中止。想像一下，如果多台機器嘗試同時減少相同的餘額鍵。一旦其中一台機器更新了該鍵，所有其他正在監控它的機器必須重新開始它們的交易，使得錯誤處理變得**複雜且低效**。

### 更好的方法：類 SQL 交易
與其依賴樂觀鎖，不如讓我們看看 **SQL 數據庫如何實現更強大的事務模型**：
1. 修改是直到提交前都是彼此隔離的。
2. 如果發生錯誤，一切都會自動回滾。
3. 確保 **原子性** - 要麼所有命令成功執行，要麼完全不應用。

例如，在 PostgreSQL 中，我們可以這樣寫：

BEGIN;

UPDATE accounts SET balance = balance - 100 WHERE id = 1;

UPDATE accounts SET balance = balance + 100 WHERE id = 2;

COMMIT;

如果任何操作失敗（例如餘額不足），數據庫會自動將所有操作回滾，以防止部分更新。為了確保真正的 **可預測性和可靠性的交易**，我計劃實現一個支持 SQL 類型事務的內存數據庫，以確保在故障情況下具有完整的原子性、一致性和回滾能力。

---

## 設計一個安全並發、具備事務功能的內存儲存系統
建立一個具有事務支持且安全並發的鍵值存儲系統對我來說並不是一件簡單的事情。因此，我打算把這個挑戰分解成幾個小步驟。

## 第一步：實施安全並發的內存鍵值存儲
當設計 Go 的鍵值存儲時，首先想到的是 **map[K]V**。然而，Go 的內建映射並不是 **并发安全** 的——對 map 的讀取和寫入如果同時進行就會導致運行時錯誤。Go 提供了 `sync.Map` 作為內建的并发安全映射，但此時我還不太清楚如何在其上實現事務支持。因此，為了靈活起見，我決定使用 **`sync.RWMutex`** 實現一個 **安全并发映射**。

使用 `sync.RWMutex` 而非 `sync.Mutex` 的原因在於其提供了更為精細的併發控制。具體來說，**讀取鎖（RLock）** 允許多個 goroutine 同時進行讀取，而不會互相阻塞；而 **寫入鎖（Lock）** 則確保對數據的修改具有獨占性。這種「多個讀者、單一寫入者」的模式，相較於 `sync.Mutex` 鎖住讀取和寫入操作，能顯著提升性能並改善擴展性。

## 實作：簡易的併發安全鍵值存儲

type Database[K comparable, V any] struct {

    mu   sync.RWMutex

    data map[K]V

}



func (db *Database[K, V]) Get(key K) (V, bool) {

    db.mu.RLock()

    defer db.mu.RUnlock()

    val, exists := db.data[key]

    return val, exists

}



func (db *Database[K, V]) Set(key K, value V) error {

    db.mu.Lock()

    defer db.mu.Unlock()

    db.data[key] = value

    return nil

}

透過上述實作，我們得到了基本操作上具備併發安全性的鍵值存儲。

---

## 第一步：鎖定機制

當衝突事件較少時，樂觀鎖定是一個非常不錯的方法。這種方式能夠有效地管理資源，尤其在高併發環境中，可以顯著減少因為等待資源而造成的效能損失。此外，在設計資料庫系統時，考量 ACID 特性（原子性、一致性、隔離性與持久性）是至關重要的，它們共同保障交易的可靠運行。同時，良好的資料鎖定策略也能降低競爭條件及死鎖問題出現的可能。

不僅如此，不同隔離級別對性能與一致性的影響也是值得深入探討的議題，以便讓開發者在實際應用中選擇最適合自己需求的設定。例如，在一些特定情況下，即使是較低的一致性要求，也可以達到更高的效能表現。因此，理解及運用這些技術將有助於提升系統整體效率和穩定度。

在高度並發的系統中，衝突往往是常態，這使得樂觀鎖變得不太可靠——它可能導致頻繁的交易失敗和重試。## 解決樂觀鎖問題：悲觀鎖不同於假設衝突「不會發生」的樂觀鎖，**悲觀鎖則認為衝突「會發生」**。它透過在修改資源之前進行鎖定來防止數據衝突。我們來比較一下這兩者：雖然**當爭用較少時，樂觀鎖效率較高，但在高併發下卻容易出錯**。舉例來說，Redis 提供了使用 `WATCH` 的**樂觀事務**，但是如果其他進程修改了被監視的鍵，那麼該事務就會失敗，需要重新嘗試。在高流量系統中，這可能造成不可預測且「容易出錯」的體驗。而使用悲觀鎖，我們僅需對正在修改的數據進行鎖定，可以防止衝突，而無需重試。## 實現按鍵鎖定通過使用 **悲觀鎖** ，我們將立即對數據進行鎖定。但是，我們不能在交易期間只對整個數據進行加鎖，而是要針對特定的相關數據進行加鎖。因此，我們引入了 **按鍵擁有獨立鎖** 的概念。這樣可以確保只有特定的鍵被上锁，以便提升併發性。以下是修改後的 `Database` 結構：

type Database[K comparable, V any] struct { dataMu sync.RWMutex data map[K]V lockMu sync.RWMutex locks map[K]*sync.Mutex}

- `dataMu`：保護訪問實際的鍵值存儲；- `lockMu`：保護訪問錶面以防多個 goroutine 同時修改；- `locks`：存儲每個鍵所擁有的獨立鎖（`map[K]*sync.Mutex`），確保僅有特定鍵被加密。在設計一個並發安全的內存鍵值存儲時，可以考慮結合樂觀與悲觀策略，以提高資料的一致性及可用性。此外，選擇合適的数据结构，比如跳表或紅黑樹，也能增強查詢性能。同時，自訂記憶體分配器以減少碎片化，也能提升系統效能。而針對各類交易需求，可調整事務隔離級別及最大併發連線等參數，以達到更佳優化效果。

在初始化階段，`locks` 的映射是空的。我們需要一種方式來**安全地初始化和檢索鎖**。### 第一次迭代：使用全局鎖

func (db *Database[K, V]) getLock(key K) *sync.Mutex {

    db.lockMu.Lock()

    defer db.lockMu.Unlock()

    if _, exists := db.locks[key]; !exists {

        db.locks[key] = &sync.Mutex{}

    }

    return db.locks[key]

}

**優點**：簡單且易於理解。
**缺點**：使用了**全局鎖**，這意味著同時只能有一個讀取者或寫入者能夠訪問鎖。

### 第二次迭代：使用讀取優化的鎖定策略

func (db *Database[K, V]) initLock(key K) *sync.Mutex {

   db.lockMu.Lock()

   defer db.lockMu.Unlock()

   if lock, exists := db.locks[key]; exists {

      return lock

   }

   lock := &sync.Mutex{}

   db.locks[key] = lock

   return lock

}



func (db *Database[K, V]) getLock(key K) *sync.Mutex {

   db.lockMu.RLock()

   lock, exists := db.locks[key]

   if !exists {

      db.lockMu.RUnlock()

      return db.initLock(key)

   }

   defer db.lockMu.RUnlock()

   return lock

}

**優點**：
- 支持多個讀取操作（RLock），同時仍然允許單一寫入操作（Lock）。
- 全局鎖僅在必要時獲取，有效降低了競爭的情況。

在這樣的設計中，我們強調RWMutex原理上允許多個讀取操作同時進行，這對提升系統性能相當重要。特別是在記憶體鍵值資料庫中，讀取通常比寫入頻繁，因此使用RWMutex能有效減少鎖競爭。此外，可以考慮增加一些客製化參數，例如設定最大並發讀者數量或調整超時時間，以更好地適應不同的工作負載，提高系統的靈活性與效率。

我們需要修改設置函數，因為在資料庫外部新增了一個基於每個鍵的鎖定機制。以下是新的設定函數範例：

func (db *Database[K, V]) Set(key K, value V) error {

   // 獲取針對特定鍵的鎖，以防止與事務產生衝突

   lock := db.getLock(key)

   lock.Lock()

   defer lock.Unlock() // 鎖在結束時會自動釋放

   

   db.dataMu.Lock() 

   defer db.dataMu.Unlock() // 保證資料的互斥訪問

   

   db.data[key] = value // 將值設置到資料中

   return nil

}

---

## 第三步：實現類似 SQL 的事務

一個 **類似 SQL 的事務** 包含三個主要步驟：
1. **開始事務**：獲取鎖並創建臨時儲存。
2. **執行操作**：將所有修改應用於臨時儲存。
3. **最終化**：提交（應用變更）或回滾（放棄變更）。

### 事務需求

為了確保正確的事務行為，我們必須強調以下幾點：
1. **一致性**：事務中的數據必須保持隔離且一致。
2. **原子性**：只有在提交時才會應用變更，若回滾則丟棄變更。
3. **單次最終化**：一筆交易只能被提交或回滾一次。

---

## 第一步：開始事務

### 確保一致性（需求 #1）

為了維持一致性，我們需提前鎖住所有相關鍵，以避免後續的數據衝突。

### 延遲修改（需求 #2）

我們不直接立即修改資料庫，而是將原始值複製到一個名為 **temp** 的臨時儲存中。

### 實作範例

type Transaction[K cmp.Ordered, V any] struct {

   db       *Database[K, V]

   locks    []*sync.Mutex

   temp     map[K]V

   txStatus int

}



func (db *Database[K, V]) StartTransaction(keys []K) *Transaction[K, V] {

   tx := &Transaction[K, V]{

      db:     db,

      temp:   make(map[K]V),

      txStatus: 0,

   }

   

   // 為了防止死鎖，我們先排序並去除重複項目 

   slices.Sort(keys)

   keys = slices.Compact(keys)



   for _, key := range keys {

      lock := db.getLock(key)  

      lock.Lock() // 鎖住該鍵以防止其他操作同時進行 

      

      tx.locks = append(tx.locks, lock)



      // 將原始值備份到交易的臨時儲存中 

      if val, exist := db.Get(key); exist {

         tx.temp[key] = val 

      }

    }

    

    return tx // 返回該交易物件以便後續操作使用

}

在這段過程中，我們將進一步探討如何執行事務操作，包括讀取和寫入的具體實作。首先，`Get()` 函數負責從臨時儲存區進行讀取，以確保資料隔離性；而 `Set()` 函數則是將資料寫入到臨時儲存區，以避免直接修改正式的資料庫。

接下來，在提交或回滾事務時，我們需要特別注意幾個要點。在提交之前，必須檢查該事務是否已經被提交或回滾，這樣可以避免重複操作。當準備好提交時，所有在臨時儲存中的資料會被應用到正式的資料庫中，同時也需釋放鎖定，以便其他事務能夠繼續進行。

以下是相關函數的實作範例：

func (tx *Transaction[K, V]) Get(key K) (V, bool) {

    val, exists := tx.temp[key]

    return val, exists

}



func (tx *Transaction[K, V]) Set(key K, value V) error {

    tx.temp[key] = value

    return nil

}

對於事務狀態管理，我們使用常量來表示不同的狀態，包括待處理、已提交及已回滾。在提交過程中，若發現事務狀態為已提交或已回滾，就不再執行任何改動。

const (

   TxPending    = 0

   TxCommitted  = 1

   TxRolledBack = 2

)



func (tx *Transaction[K, V]) Commit() error {

   switch tx.txStatus {

   case TxCommitted:

      return errors.New("transaction already committed")

   case TxRolledBack:

      return errors.New("transaction already rolled back")

   }

   

   tx.txStatus = TxCommitted

   

   // 鎖定資料以防其他操作同時訪問。

   tx.db.dataMu.Lock()

   defer tx.db.dataMu.Unlock()

   

   // 將所有臨時變更應用至正式資料庫。

   for key, value := range tx.temp {

      tx.db.data[key] = value

   }

   

   // 解鎖所有鎖定資源。

   for _, lock := range tx.locks {

      lock.Unlock()

   }

   

   return nil

}

透過上述步驟，我們保證了事務在ACID特性（原子性、一致性、隔離性及持久性）上的可靠實現。值得一提的是，不同的隔離級別，如可重複讀取與讀取已提交，各自有其優缺點，因此根據應用需求調整交易參數，可以有效提升性能與安全性，使得整個系統運作更加穩健。

### 回滾：放棄變更（需求 #2 和 #3）- **因為變更並未應用於實際資料庫，** 所以回滾只需解鎖鎖定的鍵。 - **確保回滾操作只執行一次。**
### 實現

func (tx *Transaction[K, V]) Rollback() {   

    if tx.txStatus != TxPending {      

        return   

    }   

    tx.txStatus = TxRolledBack   

    for _, lock := range tx.locks {      

        lock.Unlock()   

    }

}

## 使用範例

func (h *UserBalanceHandler) TransferBalance(senderDBKey, receiverDBKey string, amount int64) {   

    tx := h.db.StartTransaction([]string{senderDBKey, receiverDBKey})   

    defer tx.Rollback()   



    senderBalance, ok := tx.Get(senderDBKey)   

    if !ok {      

        return   

    }   



    receiverBalance, ok := tx.Get(receiverDBKey)   

    if !ok {      

        return   

    }   



    finalSenderBalance := senderBalance - amount    

    finalReceiverBalance := receiverBalance + amount   



   if finalSenderBalance < 0 {      

       return  

   }   



   err := tx.Set(senderDBKey, finalSenderBalance)    

   if err != nil {      

       return  

   }   



   err = tx.Set(receiverDBKey, finalReceiverBalance)    

   if err != nil {      

       return  

   }  



   err = tx.Commit()    

   if err != nil {      

       return  

   } 

}

## 最終代碼```go
const (
TxPending = 0
TxCommitted = 1
TxRolledBack = 2
)

type Database[K cmp.Ordered, V any] struct {
dataMu sync.RWMutex
data map[K]V
lockMu sync.RWMutex
locks map[K]*sync.Mutex
}

type Transaction[K cmp.Ordered, V any] struct {
db *Database[K, V]
locks []*sync.Mutex
temp map[K]V
txStatus int
}

func NewDatabase[K cmp.Ordered, V any]() *Database[K, V] {
return &Database[K, V]{
data: make(map[K]V),
locks: make(map[K]*sync.Mutex),
}
}

func (db *Database[K, V]) initLock(key K) *sync.Mutex {
db.lockMu.Lock()
defer db.lockMu.Unlock()

if lock, exists := db.locks[key]; exists {
return lock
}

lock := &sync.Mutex{}
db.locks[key] = lock
return lock
}

func (db *Database[K, V]) getLock(key K) *sync.Mutex {
db.lockMu.RLock()
lock ,exists:=db.locks[key]

if !exists {
db.lockMu.RUnlock()
return db.initLock(key)
}

defer db.lockMu.RUnlock()
return lock
}

func (db *Database[K,V]) Get(key K)(V,bool){
// 獲取數據時使用讀鎖防止競爭條件發生。
db.dataMu.RLock()
defer db.dataMu.RUnlock()

val ,exists:=db.data[key]

// 返回值和存在標記。
return val ,exists
}

func (db* Database [K,V]) Set(key K,value V ) error{
// 獲取每個鍵的鎖來防止交易間的衝突。
lock:=db.getLock(key)
lock.Lock()
defer lock.Unlock()

db.dataMu.Lock()
defer db.dataMu.Unlock()

db.data[key]=value
return nil
}

func(db* Database [K,V]) StartTransaction(keys []K)* Transaction [K,V ]{
tx:=&Transaction [K,V ]{
db:db,
temp :make(map [K ]V),
}
// 排序並移除重複項以避免死鎖。
slices.Sort(keys)
keys=slices.Compact(keys)

for _,key:=range keys{
lock=db.getLock(key)
lock.Lock()
tx.locks=append(tx.locks ,lock)

// 將原始值複製到事務臨時存儲中。
if val ,exist:=db.Get(key);exist{
tx.temp[key]=val
}
}
return tx
}

func(tx* Transaction [K,V ]) Get(key K)(V,bool){
val ,exists:=tx.temp[key]
return val ,exists
}

func(tx* Transaction [K,V ]) Set(key K,value V ) error{
tx.temp[key]=value
return nil
}

func(tx* Transaction [K,V ]) Commit()(error){
switch tx.txStatus {
case TxCommitted:
return errors.New("transaction already commited")
case TxRolledBack:
return errors.New("transaction already rolled back")
}
tx.txStatus=TxCommitted

tx.db.dataMu.Lock()
defer tx.db.dataMut.Unlock()

for key,value:=range tx.temp{
// 提交所有臨時數據到實際資料庫中。
// 更新同步狀態之後解鎖所有相關的鎖定物件。
// 確保不會因為多次提交而發生錯誤。
txn.db.data[k]=v
}

for _,lock:=range txn.locks{
// 解開所有所持有的控制權限。
unlock.unlock ()
}
return nil
}

// 回滾方法僅在需要時被調用，以避免對資料庫造成影響。
// 在進行任何更改前必須有一個有效狀態，這樣才能確保事務的一致性與完整性。

func(tx* Transaction [k,v ]) Rollback(){
if txn .txStatus!=TxPending{
// 如果當前事務狀態不是等待中的話，則無需再做處理了。
return
}

txn.txStatus=TxRolledBack

for _,lock=range txn.locks{
// 對每個持有的資源進行解鎖，以釋放佔用資源。
unlock.unlock ()
}