2. 程式人生 > 實用技巧 >nacos之配置檔案實時重新整理

nacos之配置檔案實時重新整理

阿新 • • 發佈：2020-11-30

本文是使用 golang 實現 redis 系列的第八篇， 將介紹如何在分散式快取中使用 Try-Commit-Catch 方式來解決分散式一致性問題。

godis 叢集的原始碼在

在上一篇文章中我們使用一致性 hash 演算法將快取中的 key 分散到不同的伺服器節點中，從而實現了分散式快取。隨之而來的問題是：一條指令(比如 MSET)可能需要多個節點同時執行，可能有些節點成功而另一部分節點失敗。

對於使用者而言這種部分成功部分失敗的情況非常難以處理，所以我們需要保證 MSET 操作要麼全部成功要麼全部失敗。

MSET 命令在叢集模式下的問題

於是問題來了 DEL、MSET 等命令所涉及的 key 可能分佈在不同的節點中，在叢集模式下實現這類涉及多個 key 的命令最簡單的方式當然是 For-Each 遍歷 key 並向它們所在的節點發送相應的操作指令。 以 MGET 命令的實現為例：

func MGet(cluster *Cluster, c redis.Connection, args [][]byte) redis.Reply {
    if len(args) < 2 {
        return reply.MakeErrReply("ERR wrong number of arguments for 'mget' command")
    }
    // 從引數列表中取出要讀取的 key
    keys := make([]string, len(args)-1)
    for i := 1; i < len(args); i++ {
        keys[i-1] = string(args[i])
    }

    resultMap := make(map[string][]byte)
    // 計算每個 key 所在的節點，並按照節點分組
    groupMap := cluster.groupBy(keys) 
    // groupMap 的型別為 map[string][]string，key 是節點的地址，value 是 keys 中屬於該節點的 key 列表
    for peer, group := range groupMap {
        // 向每個節點發送 mget 指令，讀取分佈在它上面的 key
        resp := cluster.Relay(peer, c, makeArgs("MGET", group...))
        if reply.IsErrorReply(resp) {
            errReply := resp.(reply.ErrorReply)
            return reply.MakeErrReply(fmt.Sprintf("ERR during get %s occurs: %v", group[0], errReply.Error()))
        }
        arrReply, _ := resp.(*reply.MultiBulkReply)
        // 將每個節點上的結果 merge 到 map 中
        for i, v := range arrReply.Args {
            key := group[i]
            resultMap[key] = v
        }
    }
    result := make([][]byte, len(keys))
    for i, k := range keys {
        result[i] = resultMap[k]
    }
    return reply.MakeMultiBulkReply(result)
}

// 計算 key 所屬的節點，並按節點分組
func (cluster *Cluster) groupBy(keys []string) map[string][]string {
    result := make(map[string][]string)
    for _, key := range keys {
        // 使用一致性 hash 計算所屬節點
        peer := cluster.peerPicker.Get(key)
        // 將 key 加入到相應節點的分組中
        group, ok := result[peer]
        if !ok {
            group = make([]string, 0)
        }
        group = append(group, key)
        result[peer] = group
    }
    return result
}

那麼 MSET 命令的實現能否如法炮製呢？答案是否定的。在上面的程式碼中我們注意到，在向各個節點發送指令時若某個節點讀取失敗則會直接退出整個 MGET 執行過程。

若在執行 MSET 指令時遇到部分節點失敗或超時，則會出現部分 key 設定成功而另一份設定失敗的情況。對於快取使用者而言這種部分成功部分失敗的情況非常難以處理，所以我們需要保證 MSET 操作要麼全部成功要麼全部失敗。

兩階段提交

兩階段提交(2-Phase Commit, 2PC)演算法是解決我們遇到的一致性問題最簡單的演算法。在 2PC 演算法中寫操作被分為兩個階段來執行:

  1. Prepare 階段
  2. 協調者向所有參與者傳送事務內容，詢問是否可以執行事務操作。在 Godis 中收到客戶端 MSET 命令的節點是事務的協調者，所有持有相關 key 的節點都要參與事務。
  3. 各參與者鎖定事務相關 key 防止被其它操作修改。各參與者寫 undo log 準備在事務失敗後進行回滾。
  4. 參與者回覆協調者可以提交。若協調者收到所有參與者的YES回覆，則準備進行事務提交。若有參與者回覆NO或者超時，則準備回滾事務
  5. Commit 階段
    1. 協調者向所有參與者傳送提交請求
    2. 參與者正式提交事務，並在完成後釋放相關 key 的鎖。
    3. 參與者協調者回復ACK，協調者收到所有參與者的ACK後認為事務提交成功。
  6. Rollback 階段
    1. 在事務請求階段若有參與者回覆NO或者超時，協調者向所有參與者發出回滾請求
    2. 各參與者執行事務回滾，並在完成後釋放相關資源。
    3. 參與者協調者回復ACK，協調者收到所有參與者的ACK後認為事務回滾成功。

2PC是一種簡單的一致性協議，它存在一些問題:

  • 單點服務: 若協調者突然崩潰則事務流程無法繼續進行或者造成狀態不一致
  • 無法保證一致性: 若協調者第二階段傳送提交請求時崩潰，可能部分參與者受到COMMIT請求提交了事務，而另一部分參與者未受到請求而放棄事務造成不一致現象。
  • 阻塞: 為了保證事務完成提交，各參與者在完成第一階段事務執行後必須鎖定相關資源直到正式提交，影響系統的吞吐量。

首先我們定義事務的描述結構:

type Transaction struct {
    id      string   // 事務 ID, 由 snowflake 演算法生成
    args    [][]byte // 命令引數
    cluster *Cluster
    conn    redis.Connection

    keys    []string          // 事務中涉及的 key
    undoLog map[string][]byte // 每個 key 在事務執行前的值，用於回滾事務
}

Prepare 階段

先看事務參與者 prepare 階段的操作:

// prepare 命令的格式是: PrepareMSet TxID key1, key2 ...
// TxID 是事務 ID，由協調者決定
func PrepareMSet(cluster *Cluster, c redis.Connection, args [][]byte) redis.Reply {
    if len(args) < 3 {
        return reply.MakeErrReply("ERR wrong number of arguments for 'preparemset' command")
    }
    txId := string(args[1])
    size := (len(args) - 2) / 2
    keys := make([]string, size)
    for i := 0; i < size; i++ {
        keys[i] = string(args[2*i+2])
    }

    txArgs := [][]byte{
        []byte("MSet"),
    } // actual args for cluster.db
    txArgs = append(txArgs, args[2:]...)
    tx := NewTransaction(cluster, c, txId, txArgs, keys) // 建立新事務
    cluster.transactions.Put(txId, tx) // 儲存到節點的事務列表中
    err := tx.prepare() // 準備事務
    if err != nil {
        return reply.MakeErrReply(err.Error())
    }
    return &reply.OkReply{}
}

實際的準備操作在 tx.prepare() 中:

func (tx *Transaction) prepare() error {
    // 鎖定相關 key
    tx.cluster.db.Locks(tx.keys...)

    // 準備 undo log
    tx.undoLog = make(map[string][]byte)
    for _, key := range tx.keys {
        entity, ok := tx.cluster.db.Get(key)
        if ok {
            blob, err := gob.Marshal(entity) // 將修改之前的狀態序列化之後儲存作為 undo log
            if err != nil {
                return err
            }
            tx.undoLog[key] = blob
        } else {
           // 若事務執行前 key 是空的，在回滾時應刪除它
            tx.undoLog[key] = []byte{} 
        }
    }
    tx.status = PreparedStatus
    return nil
}

看看協調者在做什麼:

func MSet(cluster *Cluster, c redis.Connection, args [][]byte) redis.Reply {
    //  解析引數
    argCount := len(args) - 1
    if argCount%2 != 0 || argCount < 1 {
        return reply.MakeErrReply("ERR wrong number of arguments for 'mset' command")
    }

    size := argCount / 2
    keys := make([]string, size)
    valueMap := make(map[string]string)
    for i := 0; i < size; i++ {
        keys[i] = string(args[2*i+1])
        valueMap[keys[i]] = string(args[2*i+2])
    }

    // 找到所屬的節點
    groupMap := cluster.groupBy(keys)
    if len(groupMap) == 1 { // do fast
        // 若所有的 key 都在同一個節點直接執行，不使用較慢的 2pc 演算法
        for peer := range groupMap {
            return cluster.Relay(peer, c, args)
        }
    }

    // 開始準備階段
    var errReply redis.Reply
    txId := cluster.idGenerator.NextId() // 使用 snowflake 演算法決定事務 ID
    txIdStr := strconv.FormatInt(txId, 10)
    rollback := false
    // 向所有參與者傳送 prepare 請求
    for peer, group := range groupMap {
        peerArgs := []string{txIdStr}
        for _, k := range group {
            peerArgs = append(peerArgs, k, valueMap[k])
        }
        var resp redis.Reply
        if peer == cluster.self {
            resp = PrepareMSet(cluster, c, makeArgs("PrepareMSet", peerArgs...))
        } else {
            resp = cluster.Relay(peer, c, makeArgs("PrepareMSet", peerArgs...))
        }
        if reply.IsErrorReply(resp) {
            errReply = resp
            rollback = true
            break
        }
    }
    if rollback {
        // 若 prepare 過程出錯則執行回滾
        RequestRollback(cluster, c, txId, groupMap)
    } else {
        _, errReply = RequestCommit(cluster, c, txId, groupMap)
        rollback = errReply != nil
    }
    if !rollback {
        return &reply.OkReply{}
    }
    return errReply
}

Commit 階段

事務參與者提交本地事務:

func Commit(cluster *Cluster, c redis.Connection, args [][]byte) redis.Reply {
    if len(args) != 2 {
        return reply.MakeErrReply("ERR wrong number of arguments for 'commit' command")
    }
    // 讀取事務資訊
    txId := string(args[1])
    raw, ok := cluster.transactions.Get(txId)
    if !ok {
        return reply.MakeIntReply(0)
    }
    tx, _ := raw.(*Transaction)

    // 在提交成功後解鎖 key
    defer func() {
        cluster.db.UnLocks(tx.keys...)
        tx.status = CommitedStatus
        //cluster.transactions.Remove(tx.id) // cannot remove, may rollback after commit
    }()

    cmd := strings.ToLower(string(tx.args[0]))
    var result redis.Reply
    if cmd == "del" {
        result = CommitDel(cluster, c, tx)
    } else if cmd == "mset" {
        result = CommitMSet(cluster, c, tx)
    }

    // 提交失敗
    if reply.IsErrorReply(result) {
        err2 := tx.rollback()
        return reply.MakeErrReply(fmt.Sprintf("err occurs when rollback:  %v, origin err: %s", err2, result))
    }

    return result
}

// 執行操作
func CommitMSet(cluster *Cluster, c redis.Connection, tx *Transaction) redis.Reply {
    size := len(tx.args) / 2
    keys := make([]string, size)
    values := make([][]byte, size)
    for i := 0; i < size; i++ {
        keys[i] = string(tx.args[2*i+1])
        values[i] = tx.args[2*i+2]
    }
    for i, key := range keys {
        value := values[i]
        cluster.db.Put(key, &db.DataEntity{Data: value})
    }
    cluster.db.AddAof(reply.MakeMultiBulkReply(tx.args))
    return &reply.OkReply{}
}

協調者的邏輯也很簡單:

func RequestCommit(cluster *Cluster, c redis.Connection, txId int64, peers map[string][]string) ([]redis.Reply, reply.ErrorReply) {
    var errReply reply.ErrorReply
    txIdStr := strconv.FormatInt(txId, 10)
    respList := make([]redis.Reply, 0, len(peers))
    for peer := range peers {
        var resp redis.Reply
        if peer == cluster.self {
            resp = Commit(cluster, c, makeArgs("commit", txIdStr))
        } else {
            resp = cluster.Relay(peer, c, makeArgs("commit", txIdStr))
        }
        if reply.IsErrorReply(resp) {
            errReply = resp.(reply.ErrorReply)
            break
        }
        respList = append(respList, resp)
    }
    if errReply != nil {
        RequestRollback(cluster, c, txId, peers)
        return nil, errReply
    }
    return respList, nil
}

Rollback

回滾本地事務:

func Rollback(cluster *Cluster, c redis.Connection, args [][]byte) redis.Reply {
    if len(args) != 2 {
        return reply.MakeErrReply("ERR wrong number of arguments for 'rollback' command")
    }
    txId := string(args[1])
    raw, ok := cluster.transactions.Get(txId)
    if !ok {
        return reply.MakeIntReply(0)
    }
    tx, _ := raw.(*Transaction)
    err := tx.rollback()
    if err != nil {
        return reply.MakeErrReply(err.Error())
    }
    return reply.MakeIntReply(1)
}

func (tx *Transaction) rollback() error {
    for key, blob := range tx.undoLog {
        if len(blob) > 0 {
            entity := &db.DataEntity{}
            err := gob.UnMarshal(blob, entity) // 反序列化事務前的快照
            if err != nil {
                return err
            }
            tx.cluster.db.Put(key, entity) // 寫入事務前的資料
        } else {
            tx.cluster.db.Remove(key) // 若事務開始之前 key 不存在則將其刪除
        }
    }
    if tx.status != CommitedStatus {
        tx.cluster.db.UnLocks(tx.keys...)
    }
    tx.status = RollbackedStatus
    return nil
}

協調者的邏輯與 commit 類似:

func RequestRollback(cluster *Cluster, c redis.Connection, txId int64, peers map[string][]string) {
    txIdStr := strconv.FormatInt(txId, 10)
    for peer := range peers {
        if peer == cluster.self {
            Rollback(cluster, c, makeArgs("rollback", txIdStr))
        } else {
            cluster.Relay(peer, c, makeArgs("rollback", txIdStr))
        }
    }
}

相關推薦

nacos之配置檔案實時重新整理

當初為了解決nacos配置檔案實時重新整理問題，搜尋了很多資料，仍無效，最後不經意間的嘗試卻解決了這個問題。

golang常用庫之配置檔案解析庫-viper使用詳解

golang常用庫：gorilla/mux-http路由庫使用 golang常用庫：配置檔案解析庫-viper使用 golang常用庫：操作資料庫的orm框架-gorm基本使用

Kafka學習之配置檔案詳解

　　kafka的配置檔案包含server.properties、proceducer.properties、consumer.properties, 分別為 broker、producer、consumer三個不同的配置。

Qt基礎之配置檔案

　　在專案開發中，配置檔案通常分為三種：ini檔案、xml檔案和json檔案，個人認為三種檔案的區別在於：ini檔案記錄方便、格式簡單但不便於擴充套件；xml檔案和json檔案擴充套件性強，能夠記錄更復雜的配置檔案，但格

K8s之配置檔案kube config生成

K8s之配置檔案kubeconfig生成 在開啟了 TLS 的叢集中，每當與叢集互動的時候少不了的是身份認證，使用 kubeconfig（即證書） 和 token 兩種認證方式是最簡單也最通用的認證方式。

自動化測試框架Python+selenium+unittest系列 之 配置檔案讀取(Python)

我們在編寫指令碼的過程中，可能涉及一些敏感資訊，例如：使用者資訊、資料庫資訊、測試資料等，如果這些資料暴露在指令碼中，無疑是不安全的。

SpringBoot 動態載入配置檔案及重新整理Bean

需求 系統遇到這樣一個需求，線上環境在配置檔案發生變動時，可以不用經過重啟，通過重新整理介面的方式得到配置檔案的載入，主要目的是為了迅速部署，避免因手動重啟，出現數據或任務丟失的問題

pytest之配置檔案pytest.ini

　　pytest.ini檔案是pytest的主配置檔案，可以改變pytest的執行方式，它是一個固定的檔案pytest.ini檔案，讀取配置資訊，按指定的方式去執行。

7 flask之配置檔案

flask配置檔案 flask中的配置檔案是一個flask.config.Config物件（繼承字典）,預設配置為：

MyBatis之配置檔案解析_類型別名及處理

JdbcType mybatis關注兩種型別：資料庫使用的型別及Java型別，Java型別通過別名處理，資料庫使用的型別用JdbcType表示，相關介面與類定義在包org.apache.ibatis.type中

MyBatis之配置檔案解析_型別處理器的處理

型別處理器介面TypeHandler JDBC執行SQL語句時，使用PreparedStatement設定引數值的方法是setXxx方法，Xxx表示一種型別，所以設定引數值需要根據引數的型別使用不同的方法，mybatis中用這個型別對應的TypeHandler物件

MyBatis之配置檔案解析_databaseIdProvider及多資料庫支援

Mybatis多資料庫支援：Configuration中有一個字串屬性databaseId，mybatis試圖通過這個屬性解決多資料庫配置問題。DatabaseIdProvider介面及實現類，它能根據DataSource獲得資料庫的標識id，mybatis提供了一個預設的

Nginx之配置檔案詳解（4）

1、 nginx.conf配置檔案 Nginx配置檔案主要分成四個部分：main（全域性設定）、server（主機設定）、upstream（上游伺服器設定，主要為反向代理、負載均衡相關配置）和location（URL匹配特定位置後的設定），每部分包

springboot基礎之配置檔案和屬性繫結功能

條件裝配：Spring中的@Conditional註解及其衍生註解的說明 在實際的專案開發中，經常遇到根據不同的環境做出不同的配置的需求，springboot中提供了將不同的配置寫在指定的環境配置檔案中，伺服器啟動時回讀取指定的配

C#之XML操作（XML配置檔案）一起學習一起進步

在開發程式中，很多超輕量級不重要資料寫在XML裡，比較靈活。 1,XML的讀取。

mybatis原始碼配置檔案解析之五：解析mappers標籤（解析XML對映檔案）

在上篇文章中分析了mybatis解析<mappers>標籤，《mybatis原始碼配置檔案解析之五：解析mappers標籤》重點分析瞭如何解析<mappers>標籤中的<package>子標籤的過程。mybatis解析<mappers>標籤

mybatis原始碼配置檔案解析之五：解析mappers標籤流程圖

前面幾篇部落格分析了mybatis解析mappers標籤的過程，主要分為解析package和mapper子標籤。補充一張解析的總體過程流程圖，畫的不好，多多諒解，感謝。

MySQL資料庫配置檔案之優先順序

目錄1.資料庫配置方式1）預編譯2）配置檔案3）命令列2.資料庫配置檔案1）配置檔案2）配置檔案讀取順序3）配置檔案優先順序4）配置檔案生效順序測試3.配置生效順序4.mysql配置優先順序總結5.配置檔案的作用1）作用2）

Spring-Boot配置檔案之application.properties常用配置

1、配置埠和專案名訪問 #指定springboot內嵌容器啟動的埠，預設使用tomcat容器時在8080埠

Python+Selenium中級篇之4-封裝一個自己的類-瀏覽器引擎類/Python讀取配置檔案內容

封裝一個自己的類-瀏覽器引擎類 前一篇文章我們知道了，如何去封裝幾個簡單的Selenium方法到我們自定義的類，這次我們編寫一個類，叫瀏覽器引擎類，通過更改一個字串的值，利用if語句去判斷和控制啟動那個瀏覽器。這