基于hyperledger联盟链的汽车轨迹追溯系统

Posted 2022-01-31 计划是动力

基于hyperledger联盟链的汽车轨迹追溯系统

• 一、区块链面对大数据的数据链改方案
• 二、基于nodejs的infuxdb时序数据库后台搭建
• 三、链码修订，改为Datahash上链

一、区块链面对大数据的数据链改方案

区块链是面向安全性的档案式管理数据存储方案，不适合处理大规模的数据业务，我们需要引入结构相似的influxdb 时序数据库进行大流量数据处理，同时上链的数据改为轨迹json字符串的hash，在取数据的时候进行区块链上轨迹hash比对

具体思想可以参照链接
处理大数据的区块链链改方案

二、基于nodejs的infuxdb时序数据库后台搭建

（具体环境搭建在docker专栏搭建influxdb教程里）
在controller文件夹里创建influxdb.js，搭建influxdb交互模块

const Influx = require('influx'); //导包
// 定义数据库连接和数据格式，创建client
const client = new Influx.InfluxDB(
  database: '******',
  username: '******',
  password: '******',
  hosts: [ host: 'xxx.xxx.xxx.xxx' ],
  schema: [
    
      measurement: 'trails', //类似于数据表的概念
      fields:  //数据表的字段，定义类型，FLOAT/INTEGER/STRING/BOOLEAN
			//   trid:Influx.FieldType.STRING,
            //   userid:Influx.FieldType.STRING,
              jsonStr:Influx.FieldType.STRING,
      , // tag 也是里面的字段，是自带索引光环。查询速度杠杠的。
      tags: ['trid','userid']
    
  ]
);
// 插入数据
function replaceAll(string, search, replace) 
    return string.split(search).join(replace);
  

function trim(str) 
    return str.replace(/^\\s\\s*/, '').replace(/\\s\\s*$/, ''); 
  
function writePoints(Trid, Userid, JsonStr)
    client.writePoints([
        
          measurement: 'trails',
          fields: 
            jsonStr: JsonStr
          ,
          tags: 
            trid:Trid,
            userid:Userid
          ,
        
      ])

async function queryPointsByUserid(userid)
  return await client
  .query(
    `
    SELECT * FROM "trails" WHERE "userid" = '$userid'
  `
  )

async function queryPointsByTrid(trid)
  return await client
  .query(
    `
    SELECT * FROM "trails" WHERE "trid" = '$trid'
  `
  )

exports.writePoints = writePoints;
exports.queryPointsByUserid = queryPointsByUserid;
exports.queryPointsByTrid = queryPointsByTrid;

封装写入和条件查询方法

三、链码修订，改为Datahash上链

package main

import(
	"fmt"
	"encoding/json"
	"bytes"
	"github.com/hyperledger/fabric/core/chaincode/shim"
	"github.com/hyperledger/fabric/protos/peer"
)

type TrailChaincode struct 
    

type Trail struct 
	Username string  `json:"username"`
	Userid string  `json:"userid"`
	Devid string `json:"devid"`
	Devstr string`json:"devstr"`
	Trid string `json:"trid"`
	TranHash string `json:"tranhash"`
	DataHash string `json:"datahash"`
 
func (t *TrailChaincode) Init(stub shim.ChaincodeStubInterface) peer.Response 
	return shim.Success(nil)


func (t *TrailChaincode) Invoke(stub shim.ChaincodeStubInterface) peer.Response 

	fn , args := stub.GetFunctionAndParameters()

	if fn == "insertTrail" 
		return t.insertTrail(stub,args)
	 else if fn == "searchAllTrail" 
		return t.searchAllTrail(stub,args)
	 else if fn == "searchOneTrail" 
		return t.searchOneTrail(stub,args)
	 else if fn == "deleteTrail" 
		return t.deleteTrail(stub,args)
	 else if fn == "updateTrail" 
		return t.updateTrail(stub,args)
	 else if fn == "searchTrailByid" 
		return t.searchTrailByid(stub,args)
	 else if fn == "getHistoryForTrail" 
		return t.getHistoryForTrail(stub,args)
	 
	return shim.Error("Invoke 调用方法有误！")


func (t *TrailChaincode) insertTrail(stub shim.ChaincodeStubInterface , args []string) peer.Response
	Trid := args[0]
	jsonStr := args[1]
	err := stub.PutState(Trid, []byte(jsonStr))
	if err != nil 
		shim.Error(err.Error())
	
	return shim.Success([]byte("insert 写入账本成功！"))

func (t *TrailChaincode) updateTrail(stub shim.ChaincodeStubInterface , args []string) peer.Response
	Trid := args[0]
	jsonStr := args[1]
	err := stub.PutState(Trid, []byte(jsonStr))
	if err != nil 
		shim.Error(err.Error())
	
	return shim.Success([]byte("insert 更新账本成功！"))

//查询所有的轨迹
func (t *TrailChaincode) deleteTrail(stub shim.ChaincodeStubInterface , args []string) peer.Response

	Trid := args[0]
	err := stub.DelState(Trid)
	if err != nil 
		shim.Error(err.Error())
	
	return shim.Success([]byte("delete 删除账本成功！"))

//删除指定轨迹
func (t *TrailChaincode) searchAllTrail(stub shim.ChaincodeStubInterface, args []string) peer.Response

	fmt.Println("start searchAllTrail")
	// 获取所有用户的票数
	resultIterator, err := stub.GetStateByRange("","")
	if err != nil 
		return shim.Error("查询所有的轨迹失败！")
	
	defer resultIterator.Close()

	var buffer bytes.Buffer
	buffer.WriteString("[")

	isWritten := false

	for resultIterator.HasNext() 
		queryResult , err := resultIterator.Next()
		if err != nil 
			return shim.Error(err.Error())
		

		if isWritten == true 
			buffer.WriteString(",")
		

		buffer.WriteString(string(queryResult.Value))
		isWritten = true
	

	buffer.WriteString("]")

	fmt.Printf("查询结果：\\n%s\\n",buffer.String())
	fmt.Println("end searchAllTrail")
	return shim.Success(buffer.Bytes())


//查询所有的轨迹
func (t *TrailChaincode) searchOneTrail(stub shim.ChaincodeStubInterface, args []string) peer.Response

	fmt.Println("start searchOneTrail")
	trid := args[0]
	result, err := stub.GetState(trid)

	if err != nil 
		return shim.Error("查询所有的轨迹失败！")
	

	return shim.Success(result)


func SliceRemove(s []Trail, index int) []Trail 
 
	return append(s[:index], s[index+1:]...)
	 

//通过电话号查所有轨迹
func (t *TrailChaincode) searchTrailByid(stub shim.ChaincodeStubInterface, args []string) peer.Response

	idpar := args[0]
	var revMsg Trail
	fmt.Println("start searchTrailByid")
	var trails []Trail;
	// 获取所有用户的票数
	resultIterator, err := stub.GetStateByRange("","")
	if err != nil 
		return shim.Error("查询所有的轨迹失败！")
	
	defer resultIterator.Close()

	// var buffer bytes.Buffer
	// buffer.WriteString("[")

	// isWritten := false

	for resultIterator.HasNext() 
		queryResult , err := resultIterator.Next()

		if err != nil 
			return shim.Error(err.Error())
		

		// if isWritten == true 
		// 	buffer.WriteString(",")
		// 

	    err1 := json.Unmarshal([]byte(queryResult.Value),&revMsg)
		// shim.Error(revMsg)
		if err1 != nil 
			return shim.Error(err.Error())
		

		if revMsg.Userid == idpar 
			// str, err2 := json.Marshal(revMsg)
			// if err2 != nil 
			// 	return shim.Error(err.Error())
			// 
			// buffer.WriteString(string(str))
			trails = append(trails, revMsg)  
		    // isWritten = true
		
		//  else  isWritten = false 
	

	// buffer.WriteString("]")
	//---------------------------------------------------
	// if len(trails) != 0 
	// 	trails = SliceRemove(trails,len(trails)-1)
	// 
	//---------------------------------------------------
	fmt.Printf("- searchTrailByid returning:\\n%s", trails)

	//change to array of bytes
	trailsAsBytes, _ := json.Marshal(trails)     //convert to array of bytes
	return shim.Success(trailsAsBytes)


func (t *TrailChaincode) getHistoryForTrail(stub shim.ChaincodeStubInterface, args []string) peer.Response
	type AuditHistory struct 
		TxId    string   `json:"txId"`
		Value   Trail   `json:"value"`
	
	var history []AuditHistory
	var trail Trail

	if len(args) != 1 
		return shim.Error("参数个数错误，应为1!")
	

	trid := args[0]
	fmt.Printf("- 开始历史回溯: %s\\n", trid)

	// Get History
	resultsIterator, err := stub.GetHistoryForKey(trid)
	if err != nil 
		return shim.Error(err.Error())
	
	defer resultsIterator.Close()

	for resultsIterator.HasNext() 
		historyData, err := resultsIterator.Next()
		if err != nil 
			return shim.Error(err.Error())
		

		var tx AuditHistory
		tx.TxId = historyData.TxId                     //copy transaction id over
		json.Unmarshal(historyData.Value, &trail)     //un stringify it aka JSON.parse()
		if historyData.Value == nil                   //trail has been deleted
			var emptyTrail Trail
			tx.Value = emptyTrail                 //copy nil trail
		 else 
			json.Unmarshal(historyData.Value, &trail) //un stringify it aka JSON.parse()
			tx.Value = trail                      //copy trail over
		
		history = append(history, tx)              //add this tx to the list
	
	fmt.Printf("- getHistoryForTrail returning:\\n%s", history)

	//change to array of bytes
	historyAsBytes, _ := json.Marshal(history)     //convert to array of bytes
	return shim.Success(historyAsBytes)

func main()
	err := shim.Start(new(TrailChaincode))
	if err != nil 
		fmt.Println("Trail chaincode start err")