作者 | Daniel van Flymen

來源 | Python學習開發

你是否會和我一樣，對加密數字貨幣底層的區塊鏈技術非常感興趣，特別想了解他們的運行機制。

但是學習區塊鏈技術并非一帆風順，我看多了大量的視頻教程還有各種課程，最終的感覺就是真正可用的實戰課程太少。

我喜歡在實踐中學習，尤其喜歡以代碼為基礎去了解整個工作機制。如果你我一樣喜歡這種學習方式，當你學完本教程時，你將會知道區塊鏈技術是如何工作的。

寫在開始之前

記住，區塊鏈是一個 不可變的、有序的 被稱為塊的記錄鏈。它們可以包含交易、文件或任何您喜歡的數據。但重要的是，他們用哈希 一起被鏈接在一起。

如果你不熟悉哈希，這里是一個解釋.

該指南的目的是什么？ 你可以舒服地閱讀和編寫基礎的 Python，因為我們將通過 HTTP 與區塊鏈進行討論，所以你也要了解 HTTP 的工作原理。

我需要準備什么？ 確定安裝了 Python 3.6+ (還有 pip) ，你還需要安裝 Flask、 Requests 庫:

`pip install Flask==0.12.2 requests==2.18.4`

對了，你還需要一個支持 HTTP 的客戶端，比如 Postman 或者 cURL，其他也可以。

Step 1: 創建一個區塊鏈

打開你最喜歡的文本編輯器或者 IDE, 我個人比較喜歡 PyCharm. 新建一個名為 blockchain.py 的文件。我們將只用這一個文件就可以。但是如果你還是不太清楚，你也可以參考 源碼.

描述區塊鏈

我們要創建一個 Blockchain 類 ，他的構造函數創建了一個初始化的空列表（要存儲我們的區塊鏈），并且另一個存儲交易。下面是我們這個類的實例:

blockchain.py

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 類負責管理鏈式數據，它會存儲交易并且還有添加新的區塊到鏈式數據的 Method。讓我們開始擴充更多 Method

塊是什么樣的？

每個塊都有一個 索引，一個 時間戳（Unix時間戳），一個事務列表， 一個 校驗 (稍后詳述) 和 前一個塊的散列 。

下面是一個 Block 的例子 ：

blockchain.py

block = {
 'index': 1,
 'timestamp': 1506057125.900785,
 'transactions': [
 {
 'sender': "8527147fe1f5426f9dd545de4b27ee00",
 'recipient': "a77f5cdfa2934df3954a5c7c7da5df1f",
 'amount': 5,
 }
 ],
 'proof': 324984774000,
 'previous_hash': "2cf24dba5fb0a30e26e83b2ac5b9e29e1b161e5c1fa7425e73043362938b9824"
}

在這一點上，一個 區塊鏈 的概念應該是明顯的 - 每個新塊都包含在其內的前一個塊的 散列 。這是至關重要的，因為這是 區塊鏈 不可改變的原因：如果攻擊者損壞 區塊鏈 中較早的塊，則所有后續塊將包含不正確的哈希值。

這有道理嗎？如果你還沒有想通，花點時間仔細思考一下 - 這是區塊鏈背后的核心理念

添加交易到區塊

我們將需要一個添加交易到區塊的方式。我們的 new_transaction() 方法的責任就是這個， 并且它非常的簡單:

blockchain.py

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() 方法添加了交易到列表，它返回了交易將被添加到的區塊的索引 --- 講開采下一個這對稍后對提交交易的用戶有用。

創建新的區塊

當我們的 Blockchain 被實例化后，我們需要將 創世 區塊（一個沒有前導區塊的區塊）添加進去進去。我們還需要向我們的起源塊添加一個 證明，這是挖礦的結果 (或工作證明)。我們稍后會詳細討論挖礦。

除了在構造函數中創建 創世 區塊外，我們還會補全 new_block() 、 new_transaction() 和 hash() 函數：

blockchain.py

import hashlib
import json
from time import time
class Blockchain(object):
 def __init__(self):
 self.current_transactions = []
 self.chain = []
 # 創建創世區塊
 self.new_block(previous_hash=1, proof=100)
 def new_block(self, proof, previous_hash=None):
 """
 創建一個新的區塊到區塊鏈中
 :param proof: <int> 由工作證明算法生成的證明
 :param previous_hash: (Optional) <str> 前一個區塊的 hash 值
 :return: <dict> 新區塊
 """
 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):
 """
 創建一筆新的交易到下一個被挖掘的區塊中
 :param sender: <str> 發送人的地址
 :param recipient: <str> 接收人的地址
 :param amount: <int> 金額
 :return: <int> 持有本次交易的區塊索引
 """
 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):
 """
 給一個區塊生成 SHA-256 值
 :param block: <dict> Block
 :return: <str>
 """
 # 我們必須確保這個字典（區塊）是經過排序的，否則我們將會得到不一致的散列
 block_string = json.dumps(block, sort_keys=True).encode()
 return hashlib.sha256(block_string).hexdigest()

上面的代碼應該是直白的 --- 為了讓代碼清晰，我添加了一些注釋和文檔說明。我們差不多完成了我們的區塊鏈。但在這個時候你一定很疑惑新的塊是怎么被創建、鍛造或挖掘的。

工作量證明算法

使用工作量證明（PoW）算法，來證明是如何在區塊鏈上創建或挖掘新的區塊。PoW 的目標是計算出一個符合特定條件的數字，這個數字對于所有人而言必須在計算上非常困難，但易于驗證。這是工作證明背后的核心思想。

我們將看到一個簡單的例子幫助你理解：

假設一個整數 x 乘以另一個整數 y 的積的 Hash 值必須以 0 結尾，即 hash (x * y) = ac23dc...0。設 x = 5，求 y ？用 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。因為，生成的 Hash 值結尾必須為 0。

hash(5 * 21) = 1253e9373e...5e3600155e860

在比特幣中，工作量證明算法被稱為 Hashcash ，它和上面的問題很相似，只不過計算難度非常大。這就是礦工們為了爭奪創建區塊的權利而爭相計算的問題。通常，計算難度與目標字符串需要滿足的特定字符的數量成正比，礦工算出結果后，就會獲得一定數量的比特幣獎勵（通過交易）。

驗證結果，當然非常容易。

實現工作量證明

讓我們來實現一個相似 PoW 算法。規則類似上面的例子：

找到一個數字 P ，使得它與前一個區塊的 proof 拼接成的字符串的 Hash 值以 4 個零開頭。

blockchain.py

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"

衡量算法復雜度的辦法是修改零開頭的個數。使用 4 個來用于演示，你會發現多一個零都會大大增加計算出結果所需的時間。

現在 Blockchain 類基本已經完成了，接下來使用 HTTP requests 來進行交互。

Step 2: Blockchain 作為 API 接口

我們將使用 Python Flask 框架，這是一個輕量 Web 應用框架，它方便將網絡請求映射到 Python 函數，現在我們來讓 Blockchain 運行在基于 Flask web 上。

我們將創建三個接口：

• /transactions/new 創建一個交易并添加到區塊
• /mine 告訴服務器去挖掘新的區塊
• /chain 返回整個區塊鏈

創建節點

我們的 “Flask 服務器” 將扮演區塊鏈網絡中的一個節點。我們先添加一些框架代碼：

blockchain.py

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 = 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 接口，返回整個區塊鏈。
• 第 40--41 行：服務器運行端口 5000 。

發送交易

發送到節點的交易數據結構如下：

{
 "sender": "my address",
 "recipient": "someone else's address",
 "amount": 5
}

因為我們已經有了添加交易的方法，所以基于接口來添加交易就很簡單了。讓我們為添加事務寫函數:

blockchain.py

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

挖礦

挖礦正是神奇所在，它很簡單，做了一下三件事：

1. 計算工作量證明 PoW
2. 通過新增一個交易授予礦工（自己）一個幣
3. 構造新區塊并將其添加到鏈中

blockchain.py

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 類方法進行交互。到此，我們的區塊鏈就算完成了，我們來實際運行下.

Step 3: 運行區塊鏈

你可以使用 cURL 或 Postman 去和 API 進行交互

啟動 Server：

$ python blockchain.py
* Running on http://127.0.0.1:5000/ (Press CTRL+C to quit)

讓我們通過請求 http://localhost:5000/mine （ GET ）來進行挖礦：

用 Postman 發起一個 GET 請求.

創建一個交易請求，請求 http://localhost:5000/transactions/new （POST）, 如圖

如果不是使用 Postman，則用一下的 cURL 語句也是一樣的：

$ curl -X POST -H "Content-Type: application/json" -d '{
 "sender": "d4ee26eee15148ee92c6cd394edd974e",
 "recipient": "someone-other-address",
 "amount": 5
}' "http://localhost:5000/transactions/new"

在挖了兩次礦之后，就有 3 個塊了，通過請求 http://localhost:5000/chain 可以得到所有的塊信息

{
 "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
}

Step 4: 一致性（共識）

我們已經有了一個基本的區塊鏈可以接受交易和挖礦。但是區塊鏈系統應該是分布式的。既然是分布式的，那么我們究竟拿什么保證所有節點有同樣的鏈呢？這就是一致性問題，我們要想在網絡上有多個節點，就必須實現一個一致性的算法

注冊節點

在實現一致性算法之前，我們需要找到一種方式讓一個節點知道它相鄰的節點。每個節點都需要保存一份包含網絡中其它節點的記錄。因此讓我們新增幾個接口：

1. /nodes/register 接收 URL 形式的新節點列表.
2. /nodes/resolve 執行一致性算法，解決任何沖突，確保節點擁有正確的鏈.

我們修改下 Blockchain 的 init 函數并提供一個注冊節點方法：

blockchain.py

...
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 來儲存節點，這是一種避免重復添加節點的簡單方法.

實現共識算法

就像先前講的那樣，當一個節點與另一個節點有不同的鏈時，就會產生沖突。為了解決這個問題，我們將制定最長的有效鏈條是最權威的規則。換句話說就是：在這個網絡里最長的鏈就是最權威的。我們將使用這個算法，在網絡中的節點之間達成共識。

blockchain.py

...
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 中，一個用于添加相鄰節點，另一個用于解決沖突：

blockchain.py

@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。注冊一個新節點：

然后我在節點 2 上挖掘了一些新的塊，以確保鏈條更長。之后，我在節點 1 上調用 GET /nodes/resolve，其中鏈由一致性算法取代：

這是一個包，去找一些朋友一起，以幫助測試你的區塊鏈。

我希望本文能激勵你創造更多新東西。我之所以對數字貨幣入迷，是因為我相信區塊鏈會很快改變我們看待事物的方式，包括經濟、政府、檔案管理等。