hyperledger fabric 高吞吐量的存储模型设计

电玩女神 2022-05-26 02:40 151阅读 0赞

## 概述 ##

本文主要解决的是如何设计高并发的fabric链式代码数据模型:  
大家都知道，fabric存到账本里面的数据都是KV方式存贮的，如果使用单个键来实现高并发那是不可能的,原因是: 当许多事务全部进入时，在当在对等体上模拟交易（即创建读取集）并且它已准备好提交到分类账时，另一个交易可能已经更新了相同的值.因此，单个键的简单设计模式中，读取集版本将不再与订购者中的版本匹配，并且大量并行事务将失败。为了解决这个问题，经常更新的值被存储为一系列增量值，这些增量值在必须检索时汇总。通过这种方式，不会经常读取和更新单个行，而是会考虑行的集合。

通过采用这种存储数据的方法，组织可以优化其链码以尽可能快地存储和记录事务，并且可以在选择时将总账记录集中到一个值中，而不会牺牲事务性能。 但是，鉴于Hyperledger Fabric的状态机设计，需要仔细考虑链码的数据模型设计。

上面啰里八嗦地讲了一推，感觉没讲清楚，简单来说就是: 比如有个KV对，利用增量的方式记录K值的变化，如k-uuid01 k-uuid02 k-uuid03 分别记录K对应的V的变化，然后可以定时执行K全量值的计算  
下面我们来看具体的例子，其实这个就是官方的例子，原文地址[high-throughput][]

## 源码 ##

package main
    
    import (
        "fmt"
        "strconv"
    
        "github.com/hyperledger/fabric/core/chaincode/shim"
        sc "github.com/hyperledger/fabric/protos/peer"
    )
    
    type SmartContract struct {
    }
    
    const (
        OK    = 200
        ERROR = 500
    )
    
    func (s *SmartContract) Init(APIstub shim.ChaincodeStubInterface) sc.Response {
        return shim.Success(nil)
    }
    // 该chaincode的方法以及功能
    //  - update 将增量添加到分类帐中的聚合key，则假定所有key都从0开始
    //  - get, 查询分类帐中key的总值
    //  - pruneFast, 删除与key关联的所有行，并用包含聚合值的单行替换它们
    //  - pruneSafe, 与pruneFast相同，就是多了预先计算值并在执行任何破坏性操作之前将其备份
    //  - delete, 删除与key关联的所有行
    
    func (s *SmartContract) Invoke(APIstub shim.ChaincodeStubInterface) sc.Response {
        function, args := APIstub.GetFunctionAndParameters()
        if function == "update" {
            return s.update(APIstub, args)
        } else if function == "get" {
            return s.get(APIstub, args)
        } else if function == "prunefast" {
            return s.pruneFast(APIstub, args)
        } else if function == "prunesafe" {
            return s.pruneSafe(APIstub, args)
        } else if function == "delete" {
            return s.delete(APIstub, args)
        } else if function == "putstandard" {
            return s.putStandard(APIstub, args)
        } else if function == "getstandard" {
            return s.getStandard(APIstub, args)
        }
        return shim.Error("Invalid Smart Contract function name.")
    }
    
    /**
     * Updates the ledger to include a new delta for a particular variable. If this is the first time
     * this variable is being added to the ledger, then its initial value is assumed to be 0. The arguments
     * to give in the args array are as follows:
     *  - args[0] -> name of the variable
     *  - args[1] -> new delta (float)
     *  - args[2] -> operation (currently supported are addition "+" and subtraction "-")
     *
     * @param APIstub The chaincode shim
     * @param args The arguments array for the update invocation
     *
     * @return A response structure indicating success or failure with a message
     */
    func (s *SmartContract) update(APIstub shim.ChaincodeStubInterface, args []string) sc.Response {
        // Check we have a valid number of args
        if len(args) != 3 {
            return shim.Error("Incorrect number of arguments, expecting 3")
        }
    
        // Extract the args
        name := args[0]
        op := args[2]
        _, err := strconv.ParseFloat(args[1], 64)
        if err != nil {
            return shim.Error("Provided value was not a number")
        }
    
        // Make sure a valid operator is provided
        if op != "+" && op != "-" {
            return shim.Error(fmt.Sprintf("Operator %s is unrecognized", op))
        }
    
        // Retrieve info needed for the update procedure
        txid := APIstub.GetTxID()
        compositeIndexName := "varName~op~value~txID"
    
        // Create the composite key that will allow us to query for all deltas on a particular variable
        compositeKey, compositeErr := APIstub.CreateCompositeKey(compositeIndexName, []string{name, op, args[1], txid})
        if compositeErr != nil {
            return shim.Error(fmt.Sprintf("Could not create a composite key for %s: %s", name, compositeErr.Error()))
        }
    
        // Save the composite key index
        compositePutErr := APIstub.PutState(compositeKey, []byte{
       0x00})
        if compositePutErr != nil {
            return shim.Error(fmt.Sprintf("Could not put operation for %s in the ledger: %s", name, compositePutErr.Error()))
        }
    
        return shim.Success([]byte(fmt.Sprintf("Successfully added %s%s to %s", op, args[1], name)))
    }
    
    /**
     * Retrieves the aggregate value of a variable in the ledger. Gets all delta rows for the variable
     * and computes the final value from all deltas. The args array for the invocation must contain the
     * following argument:
     *  - args[0] -> The name of the variable to get the value of
     *
     * @param APIstub The chaincode shim
     * @param args The arguments array for the get invocation
     *
     * @return A response structure indicating success or failure with a message
     */
    func (s *SmartContract) get(APIstub shim.ChaincodeStubInterface, args []string) sc.Response {
        // Check we have a valid number of args
        if len(args) != 1 {
            return shim.Error("Incorrect number of arguments, expecting 1")
        }
    
        name := args[0]
        // Get all deltas for the variable
        deltaResultsIterator, deltaErr := APIstub.GetStateByPartialCompositeKey("varName~op~value~txID", []string{name})
        if deltaErr != nil {
            return shim.Error(fmt.Sprintf("Could not retrieve value for %s: %s", name, deltaErr.Error()))
        }
        defer deltaResultsIterator.Close()
    
        // Check the variable existed
        if !deltaResultsIterator.HasNext() {
            return shim.Error(fmt.Sprintf("No variable by the name %s exists", name))
        }
    
        // Iterate through result set and compute final value
        var finalVal float64
        var i int
        for i = 0; deltaResultsIterator.HasNext(); i++ {
            // Get the next row
            responseRange, nextErr := deltaResultsIterator.Next()
            if nextErr != nil {
                return shim.Error(nextErr.Error())
            }
    
            // Split the composite key into its component parts
            _, keyParts, splitKeyErr := APIstub.SplitCompositeKey(responseRange.Key)
            if splitKeyErr != nil {
                return shim.Error(splitKeyErr.Error())
            }
    
            // Retrieve the delta value and operation
            operation := keyParts[1]
            valueStr := keyParts[2]
    
            // Convert the value string and perform the operation
            value, convErr := strconv.ParseFloat(valueStr, 64)
            if convErr != nil {
                return shim.Error(convErr.Error())
            }
    
            switch operation {
            case "+":
                finalVal += value
            case "-":
                finalVal -= value
            default:
                return shim.Error(fmt.Sprintf("Unrecognized operation %s", operation))
            }
        }
    
        return shim.Success([]byte(strconv.FormatFloat(finalVal, 'f', -1, 64)))
    }
    
    /**
     * Prunes a variable by deleting all of its delta rows while computing the final value. Once all rows
     * have been processed and deleted, a single new row is added which defines a delta containing the final
     * computed value of the variable. This function is NOT safe as any failures or errors during pruning
     * will result in an undefined final value for the variable and loss of data. Use pruneSafe if data
     * integrity is important. The args array contains the following argument:
     *  - args[0] -> The name of the variable to prune
     *
     * @param APIstub The chaincode shim
     * @param args The args array for the pruneFast invocation
     *
     * @return A response structure indicating success or failure with a message
     */
    func (s *SmartContract) pruneFast(APIstub shim.ChaincodeStubInterface, args []string) sc.Response {
        // Check we have a valid number of ars
        if len(args) != 1 {
            return shim.Error("Incorrect number of arguments, expecting 1")
        }
    
        // Retrieve the name of the variable to prune
        name := args[0]
    
        // Get all delta rows for the variable
        deltaResultsIterator, deltaErr := APIstub.GetStateByPartialCompositeKey("varName~op~value~txID", []string{name})
        if deltaErr != nil {
            return shim.Error(fmt.Sprintf("Could not retrieve value for %s: %s", name, deltaErr.Error()))
        }
        defer deltaResultsIterator.Close()
    
        // Check the variable existed
        if !deltaResultsIterator.HasNext() {
            return shim.Error(fmt.Sprintf("No variable by the name %s exists", name))
        }
    
        // Iterate through result set computing final value while iterating and deleting each key
        var finalVal float64
        var i int
        for i = 0; deltaResultsIterator.HasNext(); i++ {
            // Get the next row
            responseRange, nextErr := deltaResultsIterator.Next()
            if nextErr != nil {
                return shim.Error(nextErr.Error())
            }
    
            // Split the key into its composite parts
            _, keyParts, splitKeyErr := APIstub.SplitCompositeKey(responseRange.Key)
            if splitKeyErr != nil {
                return shim.Error(splitKeyErr.Error())
            }
    
            // Retrieve the operation and value
            operation := keyParts[1]
            valueStr := keyParts[2]
    
            // Convert the value to a float
            value, convErr := strconv.ParseFloat(valueStr, 64)
            if convErr != nil {
                return shim.Error(convErr.Error())
            }
    
            // Delete the row from the ledger
            deltaRowDelErr := APIstub.DelState(responseRange.Key)
            if deltaRowDelErr != nil {
                return shim.Error(fmt.Sprintf("Could not delete delta row: %s", deltaRowDelErr.Error()))
            }
    
            // Add the value of the deleted row to the final aggregate
            switch operation {
            case "+":
                finalVal += value
            case "-":
                finalVal -= value
            default:
                return shim.Error(fmt.Sprintf("Unrecognized operation %s", operation))
            }
        }
    
        // Update the ledger with the final value and return
        updateResp := s.update(APIstub, []string{name, strconv.FormatFloat(finalVal, 'f', -1, 64), "+"})
        if updateResp.Status == OK {
            return shim.Success([]byte(fmt.Sprintf("Successfully pruned variable %s, final value is %f, %d rows pruned", args[0], finalVal, i)))
        }
    
        return shim.Error(fmt.Sprintf("Failed to prune variable: all rows deleted but could not update value to %f, variable no longer exists in ledger", finalVal))
    }
    
    /**
     * This function performs the same function as pruneFast except it provides data backups in case the
     * prune fails. The final aggregate value is computed before any deletion occurs and is backed up
     * to a new row. This back-up row is deleted only after the new aggregate delta has been successfully
     * written to the ledger. The args array contains the following argument:
     *  args[0] -> The name of the variable to prune
     *
     * @param APIstub The chaincode shim
     * @param args The arguments array for the pruneSafe invocation
     *
     * @result A response structure indicating success or failure with a message
     */
    func (s *SmartContract) pruneSafe(APIstub shim.ChaincodeStubInterface, args []string) sc.Response {
        // Verify there are a correct number of arguments
        if len(args) != 1 {
            return shim.Error("Incorrect number of arguments, expecting 1 (the name of the variable to prune)")
        }
    
        // Get the var name
        name := args[0]
    
        // Get the var's value and process it getResp := s.get(APIstub, args) if getResp.Status == ERROR { return shim.Error(fmt.Sprintf("Could not retrieve the value of %s before pruning, pruning aborted: %s", name, getResp.Message)) } valueStr := string(getResp.Payload) val, convErr := strconv.ParseFloat(valueStr, 64) if convErr != nil { return shim.Error(fmt.Sprintf("Could not convert the value of %s to a number before pruning, pruning aborted: %s", name, convErr.Error())) } // Store the var's value temporarily
        backupPutErr := APIstub.PutState(fmt.Sprintf("%s_PRUNE_BACKUP", name), []byte(valueStr))
        if backupPutErr != nil {
            return shim.Error(fmt.Sprintf("Could not backup the value of %s before pruning, pruning aborted: %s", name, backupPutErr.Error()))
        }
    
        // Get all deltas for the variable
        deltaResultsIterator, deltaErr := APIstub.GetStateByPartialCompositeKey("varName~op~value~txID", []string{name})
        if deltaErr != nil {
            return shim.Error(fmt.Sprintf("Could not retrieve value for %s: %s", name, deltaErr.Error()))
        }
        defer deltaResultsIterator.Close()
    
        // Delete each row
        var i int
        for i = 0; deltaResultsIterator.HasNext(); i++ {
            responseRange, nextErr := deltaResultsIterator.Next()
            if nextErr != nil {
                return shim.Error(fmt.Sprintf("Could not retrieve next row for pruning: %s", nextErr.Error()))
            }
    
            deltaRowDelErr := APIstub.DelState(responseRange.Key)
            if deltaRowDelErr != nil {
                return shim.Error(fmt.Sprintf("Could not delete delta row: %s", deltaRowDelErr.Error()))
            }
        }
    
        // Insert new row for the final value
        updateResp := s.update(APIstub, []string{name, valueStr, "+"})
        if updateResp.Status == ERROR {
            return shim.Error(fmt.Sprintf("Could not insert the final value of the variable after pruning, variable backup is stored in %s_PRUNE_BACKUP: %s", name, updateResp.Message))
        }
    
        // Delete the backup value
        delErr := APIstub.DelState(fmt.Sprintf("%s_PRUNE_BACKUP", name))
        if delErr != nil {
            return shim.Error(fmt.Sprintf("Could not delete backup value %s_PRUNE_BACKUP, this does not affect the ledger but should be removed manually", name))
        }
    
        return shim.Success([]byte(fmt.Sprintf("Successfully pruned variable %s, final value is %f, %d rows pruned", name, val, i)))
    }

/** * Deletes all rows associated with an aggregate variable from the ledger. The args array * contains the following argument: * - args[0] -> The name of the variable to delete * * @param APIstub The chaincode shim * @param args The arguments array for the delete invocation * * @return A response structure indicating success or failure with a message */
    func (s *SmartContract) delete(APIstub shim.ChaincodeStubInterface, args []string) sc.Response {
        // Check there are a correct number of arguments
        if len(args) != 1 {
            return shim.Error("Incorrect number of arguments, expecting 1")
        }
    
        // Retrieve the variable name
        name := args[0]
    
        // Delete all delta rows
        deltaResultsIterator, deltaErr := APIstub.GetStateByPartialCompositeKey("varName~op~value~txID", []string{name})
        if deltaErr != nil {
            return shim.Error(fmt.Sprintf("Could not retrieve delta rows for %s: %s", name, deltaErr.Error()))
        }
        defer deltaResultsIterator.Close()
    
    
        if !deltaResultsIterator.HasNext() {
            return shim.Error(fmt.Sprintf("No variable by the name %s exists", name))
        }
    
        var i int
        for i = 0; deltaResultsIterator.HasNext(); i++ {
            responseRange, nextErr := deltaResultsIterator.Next()
            if nextErr != nil {
                return shim.Error(fmt.Sprintf("Could not retrieve next delta row: %s", nextErr.Error()))
            }
    
            deltaRowDelErr := APIstub.DelState(responseRange.Key)
            if deltaRowDelErr != nil {
                return shim.Error(fmt.Sprintf("Could not delete delta row: %s", deltaRowDelErr.Error()))
            }
        }
    
        return shim.Success([]byte(fmt.Sprintf("Deleted %s, %d rows removed", name, i)))
    }
    
    
    func f2barr(f float64) []byte {
        str := strconv.FormatFloat(f, 'f', -1, 64)
    
        return []byte(str)
    }
    
    func main() {
        err := shim.Start(new(SmartContract))
        if err != nil {
            fmt.Printf("Error creating new Smart Contract: %s", err)
        }
    }
    
    func (s *SmartContract) putStandard(APIstub shim.ChaincodeStubInterface, args []string) sc.Response {
        name := args[0]
        valStr := args[1]
    
        _, getErr := APIstub.GetState(name)
        if getErr != nil {
            return shim.Error(fmt.Sprintf("Failed to retrieve the statr of %s: %s", name, getErr.Error()))
        }
    
        putErr := APIstub.PutState(name, []byte(valStr))
        if putErr != nil {
            return shim.Error(fmt.Sprintf("Failed to put state: %s", putErr.Error()))
        }
    
        return shim.Success(nil)
    }
    
    func (s *SmartContract) getStandard(APIstub shim.ChaincodeStubInterface, args []string) sc.Response {
        name := args[0]
    
        val, getErr := APIstub.GetState(name)
        if getErr != nil {
            return shim.Error(fmt.Sprintf("Failed to get state: %s", getErr.Error()))
        }
    
        return shim.Success(val)
    }

本文只是简单记录一下，不同的业务场景会有不同的使用方法，以上只是本人抛砖引玉，在此记录记录，望大家多多指导  
参考  
[high-throughput][]

[high-throughput]: https://github.com/hyperledger/fabric-samples/tree/master/high-throughput