Redis 交易與市場機制實作

在分散式系統中，資料一致性是至關重要的議題。Redis 提供了交易機制以確保多使用者操作資料的正確性，本文將探討 Redis 交易的特性，並以虛擬遊戲市場系統為例，說明如何利用 Redis 的資料結構和交易功能來實作一個高效且穩定的市場機制。此範例包含使用者資訊、庫存管理、市場商品列表以及購買流程等核心功能，並使用 Python 程式碼示範如何操作 Redis。程式碼中使用 WATCH 指令監控資料變更，並結合 MULTI 和 EXEC 指令確保交易的原子性，同時也示範瞭如何處理交易衝突和錯誤。

Redis 交易與市場機制實作

在前一章中，我們探討了 Redis 的主從複製與 Sentinel 自動容錯移轉機制，以確保系統的高用性。本章節將重點放在 Redis 交易（transactions）的概念，以及如何在多使用者環境下確保資料的一致性與正確性。接著，我們將透過一個虛擬遊戲公司的市場機制例項，展示如何利用 Redis 資料結構來設計一個可擴充套件的市場系統。

Redis 交易的必要性

當多個客戶端同時對 Redis 中的同一資料進行操作時，如果沒有適當的控制機制，可能會導致資料不一致或損壞。Redis 提供了交易機制來解決這一問題，雖然它與傳統關聯式資料函式庫的交易有所不同。

傳統資料函式庫與 Redis 交易的比較

在傳統的關聯式資料函式庫中，交易通常遵循以下流程：

1. 使用 BEGIN 指令開始一個交易。
2. 執行一系列的讀寫操作。
3. 使用 COMMIT 提交交易，使變更生效。
4. 或使用 ROLLBACK 回復交易，放棄所有變更。

Redis 的交易機制則是透過 MULTI、EXEC、DISCARD 等指令來實作：

• MULTI：標記一個交易區塊的開始。
• EXEC：執行所有在 MULTI 之後的指令。
• DISCARD：取消交易，放棄執行所有在 MULTI 之後的指令。

Redis 交易的一個關鍵特性是，它會將所有指令延遲到 EXEC 時才執行，這意味著在 EXEC 之前，無法根據讀取的資料做出決策。這種設計雖然簡化了交易的實作，但也帶來了一些挑戰，尤其是在需要根據讀取結果進行進一步操作的場景中。

市場機制的設計與實作

現在，讓我們透過一個具體的例子來展示如何利用 Redis 設計一個可擴充套件的市場系統。假設我們正在為一家虛擬遊戲公司開發一個市場功能，允許玩家在市場上購買和出售虛擬物品。

使用者與庫存的資料結構設計

首先，我們需要設計使用者及其庫存的資料結構：

• 使用者資訊儲存在一個 HASH 中，鍵為 users:<user_id>，儲存使用者的屬性，如名稱、資金等。
• 使用者的函式庫儲存存在一個 SET 中，鍵為 inventory:<user_id>，儲存使用者擁有的物品 ID。

import redis

# 連線到 Redis
r = redis.Redis(host='localhost', port=6379, db=0)

# 設定使用者資訊
def set_user_info(user_id, name, funds):
    key = f"users:{user_id}"
    r.hset(key, "name", name)
    r.hset(key, "funds", funds)

# 新增物品到使用者庫存
def add_item_to_inventory(user_id, item_id):
    key = f"inventory:{user_id}"
    r.sadd(key, item_id)

# 示例用法
set_user_info(17, "Frank", 43)
add_item_to_inventory(17, "ItemM")
add_item_to_inventory(17, "ItemN")

市場資料結構設計

為了實作市場功能，我們需要設計一個有序集合（ZSET）來儲存待售物品：

• 鍵為 market。
• 成員（member）為物品 ID 與賣家 ID 的組合（例如 "ItemA.4"）。
• 分數（score）為物品的售價。

# 在市場上列出物品
def list_item_for_sale(item_id, seller_id, price):
    member = f"{item_id}.{seller_id}"
    r.zadd("market", {member: price})

# 示例用法
list_item_for_sale("ItemA", 4, 35)
list_item_for_sale("ItemC", 7, 48)

處理購買行為

當使用者購買物品時，我們需要執行一系列操作，包括檢查買家的資金是否足夠、轉移資金、轉移物品所有權等。這些操作需要在一個交易中完成，以確保資料的一致性。

def purchase_item(buyer_id, item_id, seller_id, price):
    # 構建交易
    with r.pipeline() as pipe:
        try:
            # 監視相關鍵
            pipe.watch(f"users:{buyer_id}", f"users:{seller_id}", f"inventory:{buyer_id}", f"inventory:{seller_id}", "market")
            
            # 檢查買家資金是否足夠
            buyer_funds = int(pipe.hget(f"users:{buyer_id}", "funds"))
            if buyer_funds < price:
                return False  # 資金不足
            
            # 開始交易
            pipe.multi()
            
            # 轉移資金
            pipe.hincrby(f"users:{buyer_id}", "funds", -price)
            pipe.hincrby(f"users:{seller_id}", "funds", price)
            
            # 轉移物品所有權
            member = f"{item_id}.{seller_id}"
            pipe.zrem("market", member)
            pipe.srem(f"inventory:{seller_id}", item_id)
            pipe.sadd(f"inventory:{buyer_id}", item_id)
            
            # 執行交易
            pipe.execute()
            return True
        except redis.WatchError:
            # 交易失敗，重試或處理錯誤
            return False

# 示例用法
purchase_success = purchase_item(27, "ItemA", 4, 35)
if purchase_success:
    print("購買成功")
else:
    print("購買失敗")

#### 程式碼詳解：

上述程式碼展示瞭如何使用 Redis 的交易機制來確保在多使用者環境下資料的一致性。以下是一些關鍵點的解析：

1. 監視鍵值: 在開始交易之前，使用 WATCH 指令監視相關的鍵值。如果在 EXEC 執行之前這些鍵值被其他客戶端修改，交易將會失敗並傳回錯誤。

2. 檢查條件: 在交易中，首先檢查買家的資金是否足夠。如果不足，直接傳回失敗，不執行後續操作。

3. 交易操作: 在 MULTI 之後，執行一系列的操作，包括轉移資金、更新物品所有權等。這些操作要麼全部成功，要麼全部失敗，從而保證資料的一致性。

4. 錯誤處理: 如果在執行 WATCH 之後到 EXEC 之前的期間內，被監視的鍵值發生了變化，EXEC 將傳回 nil，表示交易失敗。這時，可以選擇重試或根據具體情況處理錯誤。

4.4.2 在市場中列出商品

在列出商品的過程中，我們將使用Redis中的WATCH操作，並結合MULTI和EXEC，有時也會使用UNWATCH或DISCARD。當我們使用WATCH監控某些鍵時，如果在執行EXEC之前，其他客戶端替換、更新或刪除了我們監控的鍵，那麼當我們嘗試執行EXEC時，針對Redis的操作將會失敗並傳回錯誤訊息，此時我們可以選擇重試或中止操作。透過使用WATCH、MULTI/EXEC和UNWATCH/DISCARD，我們可以確保在進行重要操作時，相關資料不會被改變，從而避免資料損壞。

什麼是DISCARD？

如同UNWATCH可以在WATCH之後但在MULTI之前傳送以重置連線一樣，DISCARD也可以在MULTI之後但在EXEC之前傳送以重置連線。也就是說，如果我們已經WATCH了一個或多個鍵，取得了一些資料，然後使用MULTI開始了一個事務，並在其後跟了一組命令，我們可以使用DISCARD來取消WATCH並清除所有已排隊的命令。在我們的例子中，由於我們清楚是否要執行MULTI/EXEC或UNWATCH，因此DISCARD是不必要的。

列出商品的函式實作

讓我們來看看如何在市場中列出商品。為此，我們將商品新增到市場的有序集合（ZSET）中，同時WATCH賣家的庫存以確保商品仍然可以被出售。下面是列出商品的函式實作：

def list_item(conn, itemid, sellerid, price):
    inventory = "inventory:%s" % sellerid
    item = "%s.%s" % (itemid, sellerid)
    end = time.time() + 5
    pipe = conn.pipeline()
    while time.time() < end:
        try:
            pipe.watch(inventory)
            if not pipe.sismember(inventory, itemid):
                pipe.unwatch()
                return None
            pipe.multi()
            pipe.zadd("market:", {item: price})
            pipe.srem(inventory, itemid)
            pipe.execute()
            return True
        except redis.exceptions.WatchError:
            pass
    return False

內容解密：

1. 監控賣家庫存：使用pipe.watch(inventory)來監控賣家的庫存，以確保在事務處理過程中庫存不會被其他客戶端修改。
2. 檢查商品是否存在於庫存中：使用pipe.sismember(inventory, itemid)檢查賣家是否仍然擁有該商品。如果商品不存在於庫存中，則使用pipe.unwatch()停止監控並傳回None。
3. 將商品新增到市場並從庫存中移除：如果商品存在於庫存中，則使用pipe.multi()開始一個事務。在事務中，首先使用pipe.zadd("market:", {item: price})將商品新增到市場的有序集合中，其中商品的鍵是由商品ID和賣家ID組成的，而分數是商品的價格。然後，使用pipe.srem(inventory, itemid)將商品從賣家的庫存集合中移除。
4. 執行事務：使用pipe.execute()來執行事務。如果事務執行成功，則傳回True。
5. 處理WatchError：如果在事務執行過程中發生了WatchError，則重試直到超時（5秒）。

圖示說明

以下是一個Mermaid圖表，用於說明列出商品的流程：

  sequenceDiagram
    participant Client as "客戶端"
    participant Redis as "Redis伺服器"
    Client->>Redis: WATCH inventory:17
    Client->>Redis: sismember('inventory:17', 'ItemM')
    Redis->>Client: 傳回檢查結果
    Client->>Redis: MULTI
    Client->>Redis: zadd('market:', 'ItemM.17', 97)
    Client->>Redis: srem('inventory:17', 'ItemM')
    Client->>Redis: EXEC
    Redis->>Client: 傳回執行結果

圖表翻譯： 此圖表展示了客戶端與Redis伺服器之間的互動過程。首先，客戶端對賣家17的庫存進行WATCH操作，以監控其變化。然後，客戶端檢查商品’ItemM’是否存在於賣家17的庫存中。如果存在，客戶端開始一個事務，將’ItemM.17’新增到市場的有序集合中，價格為97，並將’ItemM’從賣家17的庫存集合中移除。最後，客戶端執行事務，完成商品的上架操作。

4.4.3 購買商品

處理購買商品的操作時，我們首先需要WATCH市場和買家的資訊。然後，我們取得買家的總資金和商品的價格，並驗證買家是否有足夠的錢。如果買家沒有足夠的錢，我們取消事務。如果買家有足夠的錢，我們執行帳戶之間的資金轉移，將商品新增到買家的庫存中，並將商品從市場中移除。在發生WATCH錯誤時，我們重試最多10秒鐘。下面是處理購買商品的函式實作：

def purchase_item(conn, buyerid, itemid, sellerid, lprice):
    buyer = "users:%s" % buyerid
    seller = "users:%s" % sellerid
    item = "%s.%s" % (itemid, sellerid)
    inventory = "inventory:%s" % buyerid
    end = time.time() + 10
    pipe = conn.pipeline()
    while time.time() < end:
        try:
            pipe.watch("market:", buyer)
            price = pipe.zscore("market:", item)
            funds = int(pipe.hget(buyer, "funds"))
            if price != lprice or price > funds:
                pipe.unwatch()
                return None
            pipe.multi()
            pipe.hincrby(seller, "funds", int(price))
            pipe.hincrby(buyer, "funds", int(-price))
            pipe.sadd(inventory, itemid)
            pipe.zrem("market:", item)
            pipe.execute()
            return True
        except redis.exceptions.WatchError:
            pass
    return False

內容解密：

1. 監控市場和買家資訊：使用pipe.watch("market:", buyer)來監控市場和買家的資訊，以確保在事務處理過程中這兩者不會被其他客戶端修改。
2. 檢查商品價格和買家資金：使用pipe.zscore("market:", item)取得商品的價格，使用pipe.hget(buyer, "funds")取得買家的資金。如果商品價格發生變化或買家資金不足，則使用pipe.unwatch()停止監控並傳回None。
3. 執行資金轉移和商品轉移：如果買家資金足夠，則使用pipe.multi()開始一個事務。在事務中，首先使用pipe.hincrby(seller, "funds", int(price))和pipe.hincrby(buyer, "funds", int(-price))來執行賣家和買家之間的資金轉移。然後，使用pipe.sadd(inventory, itemid)將商品新增到買家的庫存集合中，使用pipe.zrem("market:", item)將商品從市場的有序集合中移除。
4. 執行事務：使用pipe.execute()來執行事務。如果事務執行成功，則傳回True。
5. 處理WatchError：如果在事務執行過程中發生了WatchError，則重試直到超時（10秒）。

圖示說明

以下是一個Mermaid圖表，用於說明購買商品的流程：

  sequenceDiagram
    participant Client as "客戶端"
    participant Redis as "Redis伺服器"
    Client->>Redis: WATCH market: 和 users:buyerid
    Client->>Redis: zscore('market:', 'ItemM.17')
    Client->>Redis: hget('users:buyerid', 'funds')
    Redis->>Client: 傳回商品價格和買家資金
    Client->>Redis: MULTI
    Client->>Redis: hincrby('users:sellerid', 'funds', price)
    Client->>Redis: hincrby('users:buyerid', 'funds', -price)
    Client->>Redis: sadd('inventory:buyerid', 'ItemM')
    Client->>Redis: zrem('market:', 'ItemM.17')
    Client->>Redis: EXEC
    Redis->>Client: 傳回執行結果

圖表翻譯： 此圖表展示了客戶端與Redis伺服器之間的互動過程，以完成購買商品的操作。首先，客戶端對市場和買家的資訊進行WATCH操作。然後，客戶端取得商品的價格和買家的資金，並檢查是否能夠繼續購買。如果可以，客戶端開始一個事務，執行資金轉移，將商品新增到買家的庫存中，並將商品從市場中移除。最後，客戶端執行事務，完成購買操作。