您现在的位置：程式師世界 >> 編程語言 > >> 更多編程語言 >> Python

【畢設教程】python區塊鏈實現 - proof of work工作量證明共識算法

編輯：Python

文章目錄

0 前言
1 區塊鏈基礎
- 1.1 比特幣內部結構
- 1.2 實現的區塊鏈數據結構
- 1.3 注意點
- 1.4 區塊鏈的核心-工作量證明算法
- - 1.4.1 拜占庭將軍問題
  - 1.4.2 解決辦法
  - 1.4.3 代碼實現
2 快速實現一個區塊鏈
- 2.1 什麼是區塊鏈
- 2.2 一個完整的快包含什麼
- 2.3 什麼是挖礦
- 2.4 工作量證明算法：
- 2.5 實現代碼
3 最後

0 前言

Hi，大家好，這裡是丹成學長，今天向同學們介紹如何構建一個區塊鏈系統作為畢設，區塊鏈的原理

🧿 選題指導, 項目分享：

https://gitee.com/dancheng-senior/project-sharing-1/blob/master/%E6%AF%95%E8%AE%BE%E6%8C%87%E5%AF%BC/README.md

1 區塊鏈基礎

學長以比特幣的結構向大家詳解區塊鏈的組成部分

1.1 比特幣內部結構

previous hash(前一個區塊的hash)
merkle root(默克爾樹根節點,內部存儲交易數據)
timestamp(當前區塊生成的時間)
nonce(曠工計算hash值次數)

1.2 實現的區塊鏈數據結構

index 當前第幾個區塊
timestamp 該區塊創建時的時間戳
data 交易信息
previousHash 前一個區塊的hash
hash 當前區塊的hash

1.3 注意點

第一個區塊叫做創世區塊(genesis block),區塊鏈創建的時候默認生產的這裡用的是單純的鏈表,不是用默克爾樹存儲

示例代碼

from hashlib import sha256
//區塊schema
class Block:
def __init__(self,index,timestamp,data,previousHash=""):
self.index = index
self.timestamp = timestamp
self.data = data
self.previousHash = previousHash
self.hash = self.calculateHash()
//計算當前區塊的hash值
def calculateHash(self):
plainData = str(self.index)+str(self.timestamp)+str(self.data)
return sha256(plainData.encode('utf-8')).hexdigest()
def __str__(self):
return str(self.__dict__)
//區塊鏈schema
class BlockChain:
//初始化的時候 創建 創世區塊
def __init__(self):
self.chain = [self.createGenesisBlock()]
//構建創世區塊
def createGenesisBlock(self):
return Block(0,"01/01/2018","genesis block","0")
//獲取最後一個區塊
def getLatestBlock(self):
return self.chain[len(self.chain)-1]
//往區塊鏈裡面添加區塊
def addBlock(self,newBlock):
newBlock.previousHash = self.getLatestBlock().hash
newBlock.hash = newBlock.calculateHash()
self.chain.append(newBlock)
def __str__(self):
return str(self.__dict__)
//校驗區塊鏈是不是有效的 有沒有人被篡改
def chainIsValid(self):
for index in range(1,len(self.chain)):
currentBlock = self.chain[index]
previousBlock = self.chain[index-1]
if (currentBlock.hash != currentBlock.calculateHash()):
return False
if previousBlock.hash != currentBlock.previousHash:
return False
return True
myCoin = BlockChain()
myCoin.addBlock(Block(1,"02/01/2018","{amount:4}"))
myCoin.addBlock(Block(2,"03/01/2018","{amount:5}"))
#print block info 打印區塊鏈信息
print("print block info ####:")
for block in myCoin.chain:
print(block)
#check blockchain is valid 檢查區塊鏈是不是有效的
print("before tamper block,blockchain is valid ###")
print(myCoin.chainIsValid())
#tamper the blockinfo 篡改區塊2的數據
myCoin.chain[1].data = "{amount:1002}"
print("after tamper block,blockchain is valid ###")
print(myCoin.chainIsValid())

輸出結果

print block info ####:
{
'index': 0, 'timestamp': '01/01/2018', 'data': 'genesis block', 'previousHash': '0', 'hash': 'd8d21e5ba33780d5eb77d09d3b407ceb8ade4e5545ef951de1997b209d91e264'}
{
'index': 1, 'timestamp': '02/01/2018', 'data': '{amount:4}', 'previousHash': 'd8d21e5ba33780d5eb77d09d3b407ceb8ade4e5545ef951de1997b209d91e264', 'hash': '15426e32db30f4b26aa719ba5e573f372f41e27e4728eb9e9ab0bea8eae63a9d'}
{
'index': 2, 'timestamp': '03/01/2018', 'data': '{amount:5}', 'previousHash': '15426e32db30f4b26aa719ba5e573f372f41e27e4728eb9e9ab0bea8eae63a9d', 'hash': '75119e897f21c769acee6e32abcefc5e88e250a1f35cc95946379436050ac2f0'}
before tamper block,blockchain is valid ###
True
after tamper block,blockchain is valid ###
False

1.4 區塊鏈的核心-工作量證明算法

上面學長介紹了區塊鏈的基本結構,我在之前的基礎上來簡單實現一下工作量證明算法(proof of work),在介紹pow之前先思考一下為什麼要工作量證明算法,或者再往前想一步為什麼比特幣如何解決信任的問題?

1.4.1 拜占庭將軍問題

比特幣出現之前就有了拜占庭將軍問題,主要思想是,如何在分布式系統環境裡去相信其他人發給你的信息？

一組拜占庭將軍分別各率領一支軍隊共同圍困一座城市。為了簡化問題，將各支軍隊的行動策略限定為進攻或撤離兩種。因為部分軍隊進攻部分軍隊撤離可能會造成災難性後果，因此各位將軍必須通過投票來達成一致策略，即所有軍隊一起進攻或所有軍隊一起撤離。因為各位將軍分處城市不同方向，他們只能通過信使互相聯系。在投票過程中每位將軍都將自己投票給進攻還是撤退的信息通過信使分別通知其他所有將軍，這樣一來每位將軍根據自己的投票和其他所有將軍送來的信息就可以知道共同的投票結果而決定行動策略
系統的問題在於，將軍中可能出現叛徒，他們不僅可能向較為糟糕的策略投票，還可能選擇性地發送投票信息。假設有9位將軍投票，其中1名叛徒。8名忠誠的將軍中出現了4人投進攻，4人投撤離的情況。這時候叛徒可能故意給4名投進攻的將領送信表示投票進攻，而給4名投撤離的將領送信表示投撤離。這樣一來在4名投進攻的將領看來，投票結果是5人投進攻，從而發起進攻；而在4名投撤離的將軍看來則是5人投撤離。這樣各支軍隊的一致協同就遭到了破壞。

1.4.2 解決辦法

拜占庭將軍問題主要問題是,中間人可以攔截消息,進行修改;上述的那些士兵可以理解成比特幣中的一些節點,不是所有節點拿到消息後都是可以直接處理的,先去解決一個數學問題,就是工作量證明,只有擁有特定的計算能力解決了問題之後才能去修改或者校驗(驗證,打包,上鏈)。

上圖就是簡單的工作量證明算法流程,一串數字後面有個x,x之前的數可以理解成交易數據,然後需要找到一個x,讓整個數的hash值的開頭有n個0,如果hash是很均勻的話,那麼生成的hash值每一位為0或者1都是等可能的,所以前n個都為0的概率就是2的n次方/2的hash值位數,上圖給出了如果hash值是5個bit的情況下的所有可能

1.4.3 代碼實現

from hashlib import sha256
import time
class Block:
def __init__(self,index,timestamp,data,previousHash=""):
self.index = index
self.timestamp = timestamp
self.data = data
self.previousHash = previousHash
self.nonce = 0 //代表當前計算了多少次hash計算
self.hash = self.calculateHash()
def calculateHash(self):
plainData = str(self.index)+str(self.timestamp)+str(self.data)+str(self.nonce)
return sha256(plainData.encode('utf-8')).hexdigest()
#挖礦 difficulty代表復雜度 表示前difficulty位都為0才算成功
def minerBlock(self,difficulty):
while(self.hash[0:difficulty]!=str(0).zfill(difficulty)):
self.nonce+=1
self.hash = self.calculateHash()
def __str__(self):
return str(self.__dict__)
class BlockChain:
def __init__(self):
self.chain = [self.createGenesisBlock()]
self.difficulty = 5
def createGenesisBlock(self):
return Block(0,"01/01/2018","genesis block")
def getLatestBlock(self):
return self.chain[len(self.chain)-1]
#添加區塊前需要 做一道計算題,坐完後才能把區塊加入到鏈上
def addBlock(self,newBlock):
newBlock.previousHash = self.getLatestBlock().hash
newBlock.minerBlock(self.difficulty)
self.chain.append(newBlock)
def __str__(self):
return str(self.__dict__)
def chainIsValid(self):
for index in range(1,len(self.chain)):
currentBlock = self.chain[index]
previousBlock = self.chain[index-1]
if (currentBlock.hash != currentBlock.calculateHash()):
return False
if previousBlock.hash != currentBlock.previousHash:
return False
return True
myCoin = BlockChain()
# 下面打印了每個區塊挖掘需要的時間 比特幣通過一定的機制控制在10分鐘出一個塊 
# 其實就是根據當前網絡算力 調整我們上面difficulty值的大小,如果你在
# 本地把上面代碼difficulty的值調很大你可以看到很久都不會出計算結果
startMinerFirstBlockTime = time.time()
print("start to miner first block time :"+str(startMinerFirstBlockTime))
myCoin.addBlock(Block(1,"02/01/2018","{amount:4}"))
print("miner first block time completed" + ",used " +str(time.time()-startMinerFirstBlockTime) +"s")
startMinerSecondBlockTime = time.time()
print("start to miner first block time :"+str(startMinerSecondBlockTime))
myCoin.addBlock(Block(2,"03/01/2018","{amount:5}"))
print("miner second block time completed" + ",used " +str(time.time()-startMinerSecondBlockTime) +"s\n")
#print block info
print("print block info ####:\n")
for block in myCoin.chain:
print("\n")
print(block)
#check blockchain is valid
print("before tamper block,blockchain is valid ###")
print(myCoin.chainIsValid())
#tamper the blockinfo
myCoin.chain[1].data = "{amount:1002}"
print("after tamper block,blockchain is valid ###")
print(myCoin.chainIsValid())

輸出

2 快速實現一個區塊鏈

2.1 什麼是區塊鏈

區塊鏈是一個不可變得，有序的被稱之為塊的記錄鏈，它們可以包含交易、文件或者任何你喜歡的數據，但最重要的是，它們用hash連接在一起。

2.2 一個完整的快包含什麼

一個索引，一個時間戳，一個事物列表，一個校驗，一個前快的散鏈表

2.3 什麼是挖礦

挖礦其實非常簡單就做了以下三件事：

1、計算工作量證明poW
2、通過新增一個交易賦予礦工（自已）一個幣
3、構造新區塊並將其添加到鏈中

2.4 工作量證明算法：

使用該算法來證明是如何在區塊上創建和挖掘新的區塊，pow的目標是計算出一個符合特定條件的數字，這個數字對於所有人而言必須在計算上非常困難，但易於驗證，這就是工作證明背後的核心思想計算難度與目標字符串需要滿足的特定字符串成正比。

2.5 實現代碼

import hashlib
import json
import requests
from textwrap import dedent
from time import time
from uuid import uuid4
from urllib.parse import urlparse
from flask import Flask, jsonify, request
class Blockchain(object):
def __init__(self):
...
self.nodes = set()
# 用 set 來儲存節點，避免重復添加節點.
...
self.chain = []
self.current_transactions = []
#創建創世區塊
self.new_block(previous_hash=1,proof=100)
def reister_node(self,address):
""" 在節點列表中添加一個新節點 :param address: :return: """
prsed_url = urlparse(address)
self.nodes.add(prsed_url.netloc)
def valid_chain(self,chain):
""" 確定一個給定的區塊鏈是否有效 :param chain: :return: """
last_block = chain[0]
current_index = 1
while current_index<len(chain):
block = chain[current_index]
print(f'{
last_block}')
print(f'{
block}')
print("\n______\n")
# 檢查block的散列是否正確
if block['previous_hash'] != self.hash(last_block):
return False
# 檢查工作證明是否正確
if not self.valid_proof(last_block['proof'], block['proof']):
return False
last_block = block
current_index += 1
return True
def ressolve_conflicts(self):
""" 共識算法 :return: """
neighbours = self.nodes
new_chain = None
# 尋找最長鏈條
max_length = len(self.chain)
# 獲取並驗證網絡中的所有節點的鏈
for node in neighbours:
response = requests.get(f'http://{
node}/chain')
if response.status_code == 200:
length = response.json()['length']
chain = response.json()['chain']
# 檢查長度是否長，鏈是否有效
if length > max_length and self.valid_chain(chain):
max_length = length
new_chain = chain
# 如果發現一個新的有效鏈比當前的長，就替換當前的鏈
if new_chain:
self.chain = new_chain
return True
return False
def new_block(self,proof,previous_hash=None):
""" 創建一個新的塊並將其添加到鏈中 :param proof: 由工作證明算法生成證明 :param previous_hash: 前一個區塊的hash值 :return: 新區塊 """
block = {

'index':len(self.chain)+1,
'timestamp':time(),
'transactions':self.current_transactions,
'proof':proof,
'previous_hash':previous_hash or self.hash(self.chain[-1]),
}
# 重置當前交易記錄
self.current_transactions = []
self.chain.append(block)
return block
def new_transaction(self,sender,recipient,amount):
# 將新事務添加到事務列表中
""" Creates a new transaction to go into the next mined Block :param sender:發送方的地址 :param recipient:收信人地址 :param amount:數量 :return:保存該事務的塊的索引 """
self.current_transactions.append({

'sender':sender,
'recipient':recipient,
'amount':amount,
})
return self.last_block['index'] + 1
@staticmethod
def hash(block):
""" 給一個區塊生成 SHA-256 值 :param block: :return: """
# 必須確保這個字典（區塊）是經過排序的，否則將會得到不一致的散列
block_string = json.dumps(block,sort_keys=True).encode()
return hashlib.sha256(block_string).hexdigest()
@property
def last_block(self):
# 返回鏈中的最後一個塊
return self.chain[-1]
def proof_of_work(self,last_proof):
# 工作算法的簡單證明
proof = 0
while self.valid_proof(last_proof,proof)is False:
proof +=1
return proof
@staticmethod
def valid_proof(last_proof,proof):
# 驗證證明
guess = f'{
last_proof}{
proof}'.encode()
guess_hash = hashlib.sha256(guess).hexdigest()
return guess_hash[:4] =="0000"
# 實例化節點
app = Flask(__name__)
# 為該節點生成一個全局惟一的地址
node_identifier = str(uuid4()).replace('-','')
# 實例化Blockchain類
blockchain = Blockchain()
# 進行挖礦請求
@app.route('/mine',methods=['GET'])
def mine():
# 運行工作算法的證明來獲得下一個證明。
last_block = blockchain.last_block
last_proof = last_block['proof']
proof = blockchain.proof_of_work(last_proof)
# 必須得到一份尋找證據的獎賞。
blockchain.new_transaction(
sender="0",
recipient=node_identifier,
amount=1,
)
# 通過將其添加到鏈中來構建新的塊
previous_hash = blockchain.hash(last_block)
block = blockchain.new_block(proof,previous_hash)
response = {

'message': "New Block Forged",
'index': block['index'],
'transactions': block['transactions'],
'proof': block['proof'],
'previous_hash': block['previous_hash'],
}
return jsonify(response), 200
# 創建交易請求
@app.route('/transactions/new',methods=['POST'])
def new_transactions():
values = request.get_json()
# 檢查所需要的字段是否位於POST的data中
required = ['seder','recipient','amount']
if not all(k in values for k in request):
return 'Missing values',400
#創建一個新的事物
index = blockchain.new_transaction(values['sender'], values['recipient'], values['amount'])
response = {
'message': f'Transaction will be added to Block {
index}'}
return jsonify(response), 201
# 獲取所有快信息
@app.route('/chain',methods=['GET'])
def full_chain():
response = {

'chain':blockchain.chain,
'length':len(blockchain.chain),
}
return jsonify(response),200
# 添加節點
@app.route('/nodes/register',methods=['POST'])
def register_nodes():
values = request.get_json()
nodes = values.get('nodes')
if nodes is None:
return "Error: Please supply a valid list of nodes", 400
for node in nodes:
blockchain.register_node(node)
response = {

'message': 'New nodes have been added',
'total_nodes': list(blockchain.nodes),
}
return jsonify(response), 201
# 解決沖突
@app.route('/nodes/resolve', methods=['GET'])
def consensus():
replaced = blockchain.resolve_conflicts()
if replaced:
response = {

'message': 'Our chain was replaced',
'new_chain': blockchain.chain
}
else:
response = {

'message': 'Our chain is authoritative',
'chain': blockchain.chain
}
return jsonify(response), 200
if __name__ == '__main__':
app.run(host='0.0.0.0',port=5000)

代碼弄好啟動你的項目以後打開Postman 完成以下操作