区块链实现代码详细分析（Python）
生活随笔
收集整理的這篇文章主要介紹了
区块链实现代码详细分析（Python）
小編覺(jué)得挺不錯(cuò)的,現(xiàn)在分享給大家,幫大家做個(gè)參考.
代碼
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, requestclass Blockchain(object):def __init__(self):...self.nodes = set()# 用 set 來(lái)儲(chǔ)存節(jié)點(diǎn)，避免重復(fù)添加節(jié)點(diǎn)....self.chain = []self.current_transactions = []# 創(chuàng)建創(chuàng)世區(qū)塊self.new_block(previous_hash=1, proof=100)def reister_node(self, address):"""在節(jié)點(diǎn)列表中添加一個(gè)新節(jié)點(diǎn):param address::return:"""prsed_url = urlparse(address)self.nodes.add(prsed_url.netloc)def valid_chain(self, chain):"""確定一個(gè)給定的區(qū)塊鏈?zhǔn)欠裼行?param chain::return:"""last_block = chain[0]current_index = 1while current_index < len(chain):block = chain[current_index]print('{}'.format(last_block))print('{}'.format(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 Falselast_block = blockcurrent_index += 1return Truedef ressolve_conflicts(self):"""共識(shí)算法:return:"""neighbours = self.nodesnew_chain = None# 尋找最長(zhǎng)鏈條max_length = len(self.chain)# 獲取并驗(yàn)證網(wǎng)絡(luò)中的所有節(jié)點(diǎn)的鏈for node in neighbours:response = requests.get('http://{}/chain'.format(node))if response.status_code == 200:length = response.json()['length']chain = response.json()['chain']# 檢查長(zhǎng)度是否長(zhǎng)，鏈?zhǔn)欠裼行?/span>if length > max_length and self.valid_chain(chain):max_length = lengthnew_chain = chain# 如果發(fā)現(xiàn)一個(gè)新的有效鏈比當(dāng)前的長(zhǎng)，就替換當(dāng)前的鏈if new_chain:self.chain = new_chainreturn Truereturn Falsedef new_block(self, proof, previous_hash=None):"""創(chuàng)建一個(gè)新的塊并將其添加到鏈中:param proof: 由工作證明算法生成證明:param previous_hash: 前一個(gè)區(qū)塊的hash值:return: 新區(qū)塊"""block = {'index': len(self.chain) + 1,'timestamp': time(),'transactions': self.current_transactions,'proof': proof,'previous_hash': previous_hash or self.hash(self.chain[-1]),}# 重置當(dāng)前交易記錄self.current_transactions = []self.chain.append(block)return blockdef new_transaction(self, sender, recipient, amount):# 將新事務(wù)添加到事務(wù)列表中"""Creates a new transaction to go into the next mined Block:param sender:發(fā)送方的地址:param recipient:收信人地址:param amount:數(shù)量:return:保存該事務(wù)的塊的索引"""self.current_transactions.append({'sender': sender,'recipient': recipient,'amount': amount,})return self.last_block['index'] + 1@staticmethoddef hash(block):"""給一個(gè)區(qū)塊生成 SHA-256 值:param block::return:"""# 必須確保這個(gè)字典（區(qū)塊）是經(jīng)過(guò)排序的，否則將會(huì)得到不一致的散列block_string = json.dumps(block, sort_keys=True).encode()return hashlib.sha256(block_string).hexdigest()@propertydef last_block(self):# 返回鏈中的最后一個(gè)塊return self.chain[-1]def proof_of_work(self, last_proof):# 工作算法的簡(jiǎn)單證明proof = 0while self.valid_proof(last_proof, proof) is False:proof += 1return proof@staticmethoddef valid_proof(last_proof, proof):# 驗(yàn)證證明guess = ('{}{}'.format(last_proof, proof)).encode()guess_hash = hashlib.sha256(guess).hexdigest()return guess_hash[:4] == "0000"# 實(shí)例化節(jié)點(diǎn) app = Flask(__name__)# 為該節(jié)點(diǎn)生成一個(gè)全局惟一的地址 node_identifier = str(uuid4()).replace('-', '')# 實(shí)例化Blockchain類 blockchain = Blockchain()# 進(jìn)行挖礦請(qǐng)求 @app.route('/mine', methods=['GET']) def mine():# 運(yùn)行工作算法的證明來(lái)獲得下一個(gè)證明。last_block = blockchain.last_blocklast_proof = last_block['proof']proof = blockchain.proof_of_work(last_proof)# 必須得到一份尋找證據(jù)的獎(jiǎng)賞。blockchain.new_transaction(sender="0",recipient=node_identifier,amount=1,)# 通過(guò)將其添加到鏈中來(lái)構(gòu)建新的塊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# 創(chuàng)建交易請(qǐng)求 @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# 創(chuàng)建一個(gè)新的事物index = blockchain.new_transaction(values['sender'], values['recipient'], values['amount'])response = {'message': 'Transaction will be added to Block {}'.format(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# 添加節(jié)點(diǎn) @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", 400for 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), 200if __name__ == '__main__':app.run(host='0.0.0.0', port=5000)類 BlockChain分析:
構(gòu)造函數(shù)：
- 創(chuàng)建一個(gè)變量nodes來(lái)存儲(chǔ)所有的節(jié)點(diǎn)(保證節(jié)點(diǎn)之間互不相同)
- 創(chuàng)建一個(gè)變量chain來(lái)存儲(chǔ)鏈
- 調(diào)用new_block函數(shù)創(chuàng)建創(chuàng)世塊(設(shè)置初始hash值(創(chuàng)世塊的previous hash), 檢驗(yàn)值設(shè)置為100)
hash函數(shù)(static method):
- 將整個(gè)塊的轉(zhuǎn)成一個(gè)string,，然后再通過(guò)這個(gè)string生成hash值，并轉(zhuǎn)成16進(jìn)制
- 值得稱贊的是，這里要確保通過(guò)json將dict轉(zhuǎn)成string的過(guò)程中，需要保證keys的順序。只有這樣，才能確保整個(gè)hash映射到整個(gè)塊的時(shí)候，得到的結(jié)果具有唯一性
last_block函數(shù)(通過(guò)@property操作符變成了一個(gè)屬性)
- 會(huì)返回這個(gè)blockChain中最后一個(gè)塊
valid_proof證明函數(shù)
- 傳進(jìn)來(lái)的兩個(gè)檢驗(yàn)值，一個(gè)是last_proof 還有一個(gè)就是proof。
- 在直接放在一起之后，通過(guò)哈希映射之中的sha256映射(再轉(zhuǎn)16進(jìn)制)
- 最后，通過(guò)判斷上面得到的最終的哈希值來(lái)判斷前4位是不是都是0
- 如果4個(gè)0開(kāi)頭，那么就是合法，否者就是不合法的
proof_of_work函數(shù)
- 通過(guò)傳進(jìn)來(lái)的那個(gè)函數(shù)中的參數(shù)(也就是前一個(gè)的proof值)
- 通過(guò)循環(huán)來(lái)逐步找到對(duì)應(yīng)的當(dāng)前proof使得上面的valid_proof函數(shù)返回的是一個(gè)true值
reister_node()函數(shù)
- 傳進(jìn)來(lái)的參數(shù)為address,表示地址。
- 這個(gè)參數(shù)必須是網(wǎng)頁(yè)url，或者是對(duì)應(yīng)的ip地址。
- 經(jīng)過(guò)解析之后，節(jié)點(diǎn)的命名就是一個(gè)ip地址或者是網(wǎng)頁(yè)地址（網(wǎng)絡(luò)層面上跟ip地址等價(jià)）
valid_chain()函數(shù)
- 這里會(huì)傳進(jìn)來(lái)一個(gè)chain 可以理解為一個(gè)列表之類的可遍歷的對(duì)象。
- 然后判斷后一個(gè)塊的previous_hash是不是真的就是前一個(gè)塊做了hash的結(jié)果
- 同時(shí)也需要判斷，這個(gè)檢驗(yàn)值是否合法
- 直到這些都滿足之后，才算是ok的
new_block函數(shù)：
- 創(chuàng)建一個(gè)新的塊。語(yǔ)言層面上，其實(shí)就是一個(gè)字典。每個(gè)塊存儲(chǔ)的交易都是整個(gè)類一開(kāi)始就公有的
- index設(shè)置為鏈長(zhǎng)度+1
- ‘previous_hash’:previous_hash or self.hash(self.chain[-1]) 這個(gè)地方有點(diǎn)意思。
- 如果是創(chuàng)世塊，這里就會(huì)直接得到對(duì)應(yīng)的結(jié)果
ressolve_conflicts函數(shù)：
- 先循環(huán)檢驗(yàn)每個(gè)節(jié)點(diǎn)。來(lái)進(jìn)行判斷
- 要求子節(jié)點(diǎn)存儲(chǔ)的長(zhǎng)度要小于總長(zhǎng)度
- 要求子節(jié)點(diǎn)存儲(chǔ)的鏈也必須是合法的
- 如果有合法的，并且子節(jié)點(diǎn)上的鏈長(zhǎng)度更長(zhǎng)那就復(fù)制給main服務(wù)器
- 如果發(fā)生過(guò)改變，就返回True
- else return false
總結(jié)
以上是生活随笔為你收集整理的区块链实现代码详细分析（Python）的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。
- 上一篇： plt.acorr()函数使用实例-Va
- 下一篇： Python dataframe修改列顺