本文轉自PythonCaff作者SummerHappy Coding.

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

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

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

寫在開始之前

記住，區塊鏈是一個 不可變的、有序的 被稱為塊的記錄鏈。它們可以包含交易、檔案或任何您喜歡的資料。但重要的是，他們用雜湊 一起被連結在一起。

該指南的目的是什麼? 你可以舒服地閱讀和編寫基礎的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()

@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)

用 Postman 發起一個 GET 請求.

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

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

{
  "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，其中鏈由一致性演算法取代：

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

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