只需三步,教你搭建一個區塊鏈程式
區塊鏈,大家或許都不陌生,或多或少都對它有一些瞭解。不過,這些瞭解可能都是支離破碎的。當問及其中一些概念是如何實現的,你可能就「蒙圈」了。那想了解其中的實現細節怎麼辦呢?
這篇文章可以手把手的教你搭建一個區塊鏈程式,讓你從技術層面,詳細瞭解區塊鏈的實現細節。我相信你把這篇文章的程式碼跑一遍,你會有一種「哦!原來是這麼實現的」想法。下面,我們就來試試吧。
要完全理解區塊鏈不是件容易的事,起碼對我而言。為了搞懂區塊鏈,我看過大量的視訊、研究過各種教程和為數不多的幾個案例,整個過程虐心之極。
所以,我決定在實踐中學習。在實踐中學習的一大好處是它能逼著你去理解區塊鏈最底層的原理,並且容易讓人堅持下去。如果你也想試試這個方法的話,建議你好好讀完這篇文章,跟著步驟一步步地去操作。這樣,你不僅可以親自開發出一個功能完備的區塊鏈,同時也搞清楚了區塊鏈的機制到底是什麼?
準備工作
在開始之前,我們需要做些準備工作,搞清楚一些問題。
什麼是區塊鏈?區塊鏈是由不可變的、有順序記錄的區塊組成。他們可以包含交易資料、檔案資料或者其他你想要記錄的資料。不過最重要的是這些區塊通過雜湊錶鏈接在一起。
什麼是雜湊?雜湊函式是一個輸入值函式,從該輸入建立一個確定輸入值的輸出值。更多解釋可以點選下邊這個連結:
https://learncryptography.com/hash-functions/what-are-hash-functions
這篇文章適合誰,首先Python程式設計師,你只要能輕鬆地讀寫一些基本的Python程式碼就可以了;第二是HTTP程式設計師,因為我們接下來講到的區塊鏈,是構建在HTTP上面的,這需要你起碼要了解HTTP請求的工作原理。
我需要做什麼?首先要確保安裝了Python 3.6以上的環境和Flask,此外還需要安裝一個碉堡的Requests庫。版本資訊如下:
pip install Flask==0.12.2 requests==2.18.4
哦對了,你還需要一個HTTP客戶端,比如 Postman 或者 cURL
在哪裡下載完整程式碼,請猛戳:
https://github.com/dvf/blockchain 。
下面跟著我一步一步來操作吧。
第一步:建立區塊鏈
開啟你常用的編輯器,我個人比較喜歡PyCharm。建立一個新的檔案,命名為 blockchain.py。整個專案,我們都只會用到這一個檔案。有不清楚的地方,可以參考原始碼。
表示一個區塊鏈
我們將建立一個 Blockchain 類,它的建構函式裡建立了一個初始為空的列表(用於儲存我們的區塊鏈),和一個儲存交易的列表。下邊是這個類的程式碼:
class Blockchain(object): def __init__(self): self.chain = [] self.current_transactions = [] def new_block(self): # Creates a new Block and adds it to the chain pass def new_transaction(self): # Adds a new transaction to the list of transactions pass @staticmethod def hash(block): # Hashes a Block pass @property def last_block(self): # Returns the last Block in the chain pass
Blockchain引數的作用是管理區塊鏈,也用於儲存交易資訊和新增區塊的方式。
區塊到底長什麼樣?
每一個區塊包含一個索引、一個時間戳、一個交易列表、一個證明(之後更多)和前一個區塊的雜湊值。
以下是一個區塊的例子:
block = { 'index': 1, 'timestamp': 1506057125.900785, 'transactions': [ { 'sender': "8527147fe1f5426f9dd545de4b27ee00", 'recipient': "a77f5cdfa2934df3954a5c7c7da5df1f", 'amount': 5, } ], 'proof': 324984774000, 'previous_hash': "2cf24dba5fb0a30e26e83b2ac5b9e29e1b161e5c1fa7425e73043362938b9824" }
程式碼左滑可檢視未顯示部分(下同)
到這裡,區塊鏈的原理就很容易理解了:每一個區塊包含它自己本身的一些變數,以及前一個區塊的雜湊值。這一點非常重要,因為雜湊值保證了區塊鏈不可篡改的特性。如果一個區塊受到攻擊雜湊值變了,那麼後面的所有區塊的雜湊值都會為之改變。
你可能想,我還是不太理解。沒關係,先接著往下看。
在區塊上新增交易
那麼,我們怎麼在區塊上新增交易呢?可以使用new_transaction()引數。使用方法簡單、直接,如下面程式碼所示:
class Blockchain(object): ... def new_transaction(self, sender, recipient, amount): """ Creates a new transaction to go into the next mined Block :param sender: <str> Address of the Sender :param recipient: <str> Address of the Recipient :param amount: <int> Amount :return: <int> The index of the Block that will hold this transaction """ self.current_transactions.append({ 'sender': sender, 'recipient': recipient, 'amount': amount, }) return self.last_block['index'] + 1
在 new_transaction() 新增交易資訊到列表中後,它會返回下一個將被開採區塊的索引號,交易資訊將被打包到這個區塊上。這對稍後提交交易的使用者有用。
建立新區塊
在區塊鏈建立完成後,我們需要建立一個創世區塊(也就是區塊鏈上的第一個區塊)。當然,創世區塊也需要被證明,這需要通過PoW的挖礦機制。後邊我們會更多的介紹挖礦,這兒就不做過多的介紹了。
除了在建構函式中建立創始區塊,我們還需要用new_block()、new_transaction() 和 hash()引數對其進行完善。程式碼如下:
import hashlib import json from time import time class Blockchain(object): def __init__(self): self.current_transactions = [] self.chain = [] # Create the genesis block self.new_block(previous_hash=1, proof=100) def new_block(self, proof, previous_hash=None): """ Create a new Block in the Blockchain :param proof: <int> The proof given by the Proof of Work algorithm :param previous_hash: (Optional) <str> Hash of previous Block :return: <dict> New Block """ block = { 'index': len(self.chain) + 1, 'timestamp': time(), 'transactions': self.current_transactions, 'proof': proof, 'previous_hash': previous_hash or self.hash(self.chain[-1]), } # Reset the current list of transactions 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: <str> Address of the Sender :param recipient: <str> Address of the Recipient :param amount: <int> Amount :return: <int> The index of the Block that will hold this transaction """ self.current_transactions.append({ 'sender': sender, 'recipient': recipient, 'amount': amount, }) return self.last_block['index'] + 1 @property def last_block(self): return self.chain[-1] @staticmethod def hash(block): """ Creates a SHA-256 hash of a Block :param block: <dict> Block :return: <str> """ # We must make sure that the Dictionary is Ordered, or we'll have inconsistent hashes block_string = json.dumps(block, sort_keys=True).encode() return hashlib.sha256(block_string).hexdigest()
為了方便大家理解,我在上面程式碼中加了一些註釋。到這裡,我們就已經對區塊鏈屬性有一個全面的瞭解了。不過我們還是要知道,區塊鏈是怎麼建立、怎麼開發,以及跟礦工有什麼關係。
關於工作量證明(PoW)
工作證明演算法(PoW)的作用,是對區塊鏈上建立或開發新的區塊的證明。其背後的核心是:找到一串解決某個數學問題的數字,這個數字必須符合兩個條件: 第一,難找;第二,很容易被驗證(而且是很容易被任何人驗證)。
我們來舉個非常簡單的例子來幫助大家理解。
我們來看一下這個例子,某個整數 x 乘以另外一個數 y ,得到的結果的雜湊值必須是以 0 結尾。可以簡單表示為:hash(x * y) = ac23dc...0。所以,我們的目標是找到滿足這個條件的一個 y 值。為了方便理解,我們暫定x=5。下面我們就用Python來做這樣一個運算:
from hashlib import sha256 x = 5 y = 0# We don't know what y should be yet... while sha256(f'{x*y}'.encode()).hexdigest()[-1] != "0": y += 1 print(f'The solution is y = {y}')
最終,計算結果是 y=21。因此,生成的以 0 結尾的雜湊值是:
hash(5 * 21) = 1253e9373e...5e3600155e860
在比特幣中,PoW演算法被稱為Hashcash,原理跟上面例子差不多。礦工們為了能建立一個新區塊,鉚足勁兒做著上面的數學題(只有勝出者才能新增區塊)。一般而言,證明的難度取決於字串中搜索的字元數量,先找到正確數字的曠工就能夠在每筆交易中獲得比特幣作為獎勵。
系統能夠很容易驗證他們的解決方案。
實現基本的工作量證明
下面在我們剛剛建立好的區塊鏈上,來實現一個相似的工作量證明演算法。規則與上邊那個簡單的例子相似:
找到一個數字 p ,它和前邊一個區塊的解決數字進行雜湊,生成前4位為 0 的雜湊值。
下面是具體的Python程式碼實現:
import hashlib import json from time import time from uuid import uuid4 class Blockchain(object): ... def proof_of_work(self, last_proof): """ Simple Proof of Work Algorithm: - Find a number p' such that hash(pp') contains leading 4 zeroes, where p is the previous p' - p is the previous proof, and p' is the new proof :param last_proof: <int> :return: <int> """ proof = 0 while self.valid_proof(last_proof, proof) is False: proof += 1 return proof @staticmethod def valid_proof(last_proof, proof): """ Validates the Proof: Does hash(last_proof, proof) contain 4 leading zeroes? :param last_proof: <int> Previous Proof :param proof: <int> Current Proof :return: <bool> True if correct, False if not. """ guess = f'{last_proof}{proof}'.encode() guess_hash = hashlib.sha256(guess).hexdigest() return guess_hash[:4] == "0000"
我們可以通過修改雜湊值前 0 的數量,來調整演算法的難度,一般來說,4位已經是足夠了。每在雜湊值前多加一個0,計算所花費的時間將呈指數倍增加。
到這兒,我們的類基本寫好了。下面,我們準備通過 HTTP 請求與其互動。
第二步:建立 API
我們打算使用 Python 的 Flask 框架,它是一個輕型框架,可以很容易實現端點到Python函式的對映。這樣,我們就可以使用 HTTP 請求通過網頁訪問我們的區塊鏈了。
我們用以下三個方法建立:
- /transactions/new 為一個區塊建立一個新的交易;
- /mine 告訴我們的伺服器開採一個新的區塊;
- /chain 返回完整的 Blockchain 類。
搭建 Flask 框架
我們的伺服器會在區塊鏈網路中形成單個節點。下面來建立一些樣板程式碼:
import hashlib import json from textwrap import dedent from time import time from uuid import uuid4 from flask import Flask class Blockchain(object): ... # Instantiate our Node app = Flask(__name__) # Generate a globally unique address for this node node_identifier = str(uuid4()).replace('-', '') # Instantiate the Blockchain blockchain = Blockchain() @app.route('/mine', methods=['GET']) def mine(): return "We'll mine a new Block" @app.route('/transactions/new', methods=['POST']) def new_transaction(): return "We'll add a new transaction" @app.route('/chain', methods=['GET']) def full_chain(): response = { 'chain': blockchain.chain, 'length': len(blockchain.chain), } return jsonify(response), 200 if __name__ == '__main__': app.run(host='0.0.0.0', port=5000)
我們來簡單的解釋一下上邊的程式碼:
- 第15行: 例項化我們的節點;載入 Flask 框架。
- 第18行:為我們的節點建立一個隨機名稱。
- 第21行:例項化 Blockchain 類。
- 第24-26行:建立 /mine 端點,這是一個GET請求。
- 第28-30行:建立 /transactions/new 端點,這是一個 POST 請求,我們將用它來發送資料。
- 第32-38行:建立 /chain 端點,它是用來返回整個 Blockchain 類。
- 第40-41行:設定伺服器執行埠為 5000。
交易節點
交易的請求是什麼形式呢?下面我們看看使用者傳送到伺服器的一段請求程式碼:
{ "sender": "my address", "recipient": "someone else's address", "amount": 5 }
因為我們已經寫好了將交易打包到區塊上的程式碼,剩下的部分就簡單了。只需要呼叫這個方法,從而實現新增交易的功能。下面是具體程式碼實現:
import hashlib import json from textwrap import dedent from time import time from uuid import uuid4 from flask import Flask, jsonify, request ... @app.route('/transactions/new', methods=['POST']) def new_transaction(): values = request.get_json() # Check that the required fields are in the POST'ed data required = ['sender', 'recipient', 'amount'] if not all(k in values for k in required): return 'Missing values', 400 # Create a new Transaction index = blockchain.new_transaction(values['sender'], values['recipient'], values['amount']) response = {'message': f'Transaction will be added to Block {index}'} return jsonify(response), 201
挖礦節點
挖礦節點是整個過程中最有趣的部分,它必須要達到三個目的:
- 計算工作量證明;
- 通過打包交易獎勵礦工一個幣;
- 通過將新塊新增到鏈中來偽造新塊。
import hashlib import json from time import time from uuid import uuid4 from flask import Flask, jsonify, request ... @app.route('/mine', methods=['GET']) def mine(): # We run the proof of work algorithm to get the next proof... last_block = blockchain.last_block last_proof = last_block['proof'] proof = blockchain.proof_of_work(last_proof) # We must receive a reward for finding the proof. # The sender is "0" to signify that this node has mined a new coin. blockchain.new_transaction( sender="0", recipient=node_identifier, amount=1, ) # Forge the new Block by adding it to the chain 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
這兒要注意一下,開採區塊的接收者是我們節點的地址。在這裡完成的大部分工作只是與Blockchain類中的方法進行互動。下面可以開始與我們的區塊鏈互動啦。
第三步:實現與 Blockchain 類互動
你可以使用普通的 cURL 或者 Postman 通過網路和剛才生成的 API 進行互動。
啟動伺服器:
$ python blockchain.py * Running on http://127.0.0.1:5000/ (Press CTRL+C to quit)
通過向以下地址傳送請求,我們可以嘗試一下挖礦。
http://localhost:5000/mine

使用 Postman 傳送 GET 請求
下面我們通過向下面連結傳送post請求,來建立一個新的交易:
http://localhost:5000/transactions/new 。
請求中要包含我們的交易結構。

使用 Postman 傳送一個 POST 請求
如果你用的是cURL,則可以通過下面程式碼來實現。
$ curl -X POST -H "Content-Type: application/json" -d '{ "sender": "d4ee26eee15148ee92c6cd394edd974e", "recipient": "someone-other-address", "amount": 5 }' "http://localhost:5000/transactions/new"
完成上面步驟之後,需要重啟下伺服器。這時候,我挖出了2個區塊,獲得了3個幣的獎勵。這裡,我們還可以像以下地址傳送請求,來對整條鏈進行檢查。
{ "chain": [ { "index": 1, "previous_hash": 1, "proof": 100, "timestamp": 1506280650.770839, "transactions": [] }, { "index": 2, "previous_hash": "c099bc...bfb7", "proof": 35293, "timestamp": 1506280664.717925, "transactions": [ { "amount": 1, "recipient": "8bbcb347e0634905b0cac7955bae152b", "sender": "0" } ] }, { "index": 3, "previous_hash": "eff91a...10f2", "proof": 35089, "timestamp": 1506280666.1086972, "transactions": [ { "amount": 1, "recipient": "8bbcb347e0634905b0cac7955bae152b", "sender": "0" } ] } ], "length": 3 }
第四步:形成共識
終於寫到共識了,共識機制是我認為區塊鏈中最有意思的部分。 在上面的步驟中,我們已經建立完成了一個簡單的區塊鏈,並且能夠實現交易、挖礦等基本功能。 不過,區塊鏈上的節點應該是分散的。 如果它們是分散的,我們究竟如何確保它們記錄的都是同一條鏈? 這就叫共識問題。如果我們的網路中需要多個節點,我們必須實現共識演算法。
註冊新節點
在我們實現共識演算法之前,需要解決一個問題:在同一個網路上,讓其中一個節點知道它的相鄰節點有哪些。每一個節點需要網路上的其他節點進行註冊。因此,我們將需要更多的節點:
- /nodes/register 接受URL形式的新節點列表。
- /nodes/resolve 實現我們的共識演算法,它可以解決任何爭議,保證節點具有正確的鏈。
下面,我們需要修改Blockchain類的結構,以及找到註冊節點實現的方法。
... from urllib.parse import urlparse ... class Blockchain(object): def __init__(self): ... self.nodes = set() ... def register_node(self, address): """ Add a new node to the list of nodes :param address: <str> Address of node. Eg. 'http://192.168.0.5:5000' :return: None """ parsed_url = urlparse(address) self.nodes.add(parsed_url.netloc)
這兒需要注意一下,set()函式用來儲存節點列表。 它是確保新增的新節點具有冪等性的方法,意思是無論我們使用這個方法新增特定節點多少次,它都只會出現一次。
實現共識演算法
先前提到的,當某個節點與另一個節點的記錄不一致時,會「打架」。為了解決這個衝突,我們需要制定一個規則,即最長而有效的鏈是最有權威性的。換句話說,網路上最長的鏈就是事實。使用這個演算法,我們就可以在我們的網路上達成共識。
... import requests class Blockchain(object) ... def valid_chain(self, chain): """ Determine if a given blockchain is valid :param chain: <list> A blockchain :return: <bool> True if valid, False if not """ 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") # Check that the hash of the block is correct if block['previous_hash'] != self.hash(last_block): return False # Check that the Proof of Work is correct if not self.valid_proof(last_block['proof'], block['proof']): return False last_block = block current_index += 1 return True def resolve_conflicts(self): """ This is our Consensus Algorithm, it resolves conflicts by replacing our chain with the longest one in the network. :return: <bool> True if our chain was replaced, False if not """ neighbours = self.nodes new_chain = None # We're only looking for chains longer than ours max_length = len(self.chain) # Grab and verify the chains from all the nodes in our network for node in neighbours: response = requests.get(f'http://{node}/chain') if response.status_code == 200: length = response.json()['length'] chain = response.json()['chain'] # Check if the length is longer and the chain is valid if length > max_length and self.valid_chain(chain): max_length = length new_chain = chain # Replace our chain if we discovered a new, valid chain longer than ours if new_chain: self.chain = new_chain return True return False
其中,valid_chain() 方法是負責校驗這一條鏈是否是有效的,怎麼校驗呢?遍歷每一個區塊,驗證它們的雜湊值和工作量證明。
resolve_conflicts() 是遍歷我們所有相鄰節點的方法,會下載它們的鏈,然後使用上述方法去驗證它們。如果找到一個有效的鏈,其長度大於我們的鏈,就將我們的鏈條替換為該鏈。
下面我們在API中,註冊兩個節點,一個用於新增相鄰節點,另一個用於解決衝突:
@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
此時,你可以使用不同機器(或使用同一臺機器的不同埠)啟動不同的節點。 我是使用的同一臺機器,在另外一個埠上建立了另一個節點,並將其註冊到當前節點。 因此,我有兩個節點:http:// localhost:5000 和 http:// localhost:5001。
然後,我在第二個節點上挖出了一些新的區塊,以確保第二個節點的鏈條比第一個節點的鏈條更長。 之後,我在第一個節點上呼叫 GET / nodes / resolve,使其中鏈通過共識演算法被第二個節點的鏈條取代:

在工作的共識演算法
好啦,你已經成功建立好了一個區塊鏈程式,快去叫上你的朋友們來測試一下吧!