从0开始构建区块链(Blockchain）

Posted 2021-04-29 Cocoa开发者社区

本公众号内容均为本号转发，已尽可能注明出处。因未能核实来源或转发内容图片有权利瑕疵的，请及时联系本号，本号会第一时间进行修改或删除。 QQ : 3442093904 

区块链现在有多火看看比特币的价格也就知道了，不过作为一个有逼格的程序员，我们不应只关注到币的价值(为啥我没去买比特币，T-T)，而应该去关注技术的本质，这个号称“第四次工业革命”的区块链技术。不过很多人估计对这个技术不太了解，包括我自己。既然不懂不如自己动手撸一个，实践出真知。在翻阅文章的时候刚好找到了这篇，下面让我们自己动手搭建一个简单的区块链。

简单的说，区块链就是一个不可变、有序的链(chain)结构，链中保存着称之为块(block)的记录，这些记录可以是交易，文件或是任意你想要的数据。其中重要的是它们通过哈希链接在一起。

如果你不懂哈希，可以查看一下，不过作为开发者应该都懂吧。。。

准备

这里我们会以python实现一个区块链，因此需要首先安装python，这里我是用的是python2.7版本,需要安装flask以及Requests库。

pip install Flask requests

同时还需要一个http客户端，比如postman或curl。

1、创建区块链

新建一个blockchain.py文件，我们这里只使用这一个文件。

区块链表示

我们首先创建一个Blockchain类，并且在构造函数中创建两个空的list，一个用于储存我们的区块链，另一个用于储存交易。

Blockchain类的定义如下：

class Blockchain(object):
    def __init__(self):
        self.chain = []
        self.current_transactions = []

    def new_block(self):
        # 创建一个新的块，并添加到链中
        pass

    def new_transaction(self):
        # 添加一笔新的交易到transactions中
        pass

    @staticmethod
    def hash(block):
        # 对一个块做hash
        pass

    @property
    def last_block(self):】
        # 返回链中的最后一个块
        pass

我们的Blockchain类用来管理链，它会存储交易信息，以及一些添加新的块到链中的辅助方法，让我们开始实现这些方法。

块(Block)长啥样？

每个块都包含如下属性：索引(index)，时间戳(Unix时间)，交易列表(transactions)，工作量证明(proof，稍后解释)以及前一个块的hash值，下面是一个区块的例子：

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

现在，链的概念应该很清楚了--每个新的区块都包含了上一个区块的hash值。这很重要，因为它保障了整个区块链的不可变性：如果攻击者毁坏了一个之前的块，那么后续所有的块的hash值都将错误。没有理解？停下来细想一下，这是区块链背后的核心思想

添加交易到区块

我们需要一个添加交易到区块的地方，实现一下new_transaction方法：

class Blockchain(object):
    ...

    def new_transaction(self, sender, recipient, amount):
        """
        添加一笔新的交易到transactions中
        :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

在new_transaction方法向列表中添加一个交易记录后，它返回该记录将被添加到的区块的索引--下一个待挖掘的区块。这在后面用户提交交易记录的时候会有用。

创建一个新的区块

当我们的Blockchain实例化之后，我们需要为它创建一个创世块(第一个区块，没有前区块，类似于链表的head)，并且给该创世块添加一个工作量证明(proof of work)，工作量证明是挖矿的结果。我们后面在来讨论挖矿。

除了在我们的构造函数中创建一个创世块之外，我们也要实现new_block(), new_transaction() 和 hash()方法：

import hashlib
import json
from time import time

class Blockchain(object):
    def __init__(self):
        self.chain = []
        self.current_transactions = []

        # 创建创世块
        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]),
        }

        # 重置存储交易信息的list
        self.current_transactions = []

        self.chain.append(block)
        return block

    def new_transaction(self, sender, recipient, amount)):
        """
        添加一笔新的交易到transactions中
        :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>
        """

        # 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()

上面的代码应该比较直观，我们差不多完成了区块链的表示了。不过你应该比较好奇新的区块是怎么创建或者挖出来的。

理解工作量证明(proof of work,pow)

工作量证明(POW),简单的说就是一份证明，该证明可以用来确认你做过一定量的工作。工作量证明的目标就是找到一个能解决一个问题的数字，这个数字不好找但是很容易证明(该数字就是问题的答案)，这就是工作量证明背后的核心思想(俗称挖矿)。

我们举一个简单的例子，方便理解：

假设有一个问题，一个整数x乘以整数y，对积做hash，hash值必须以0结尾，即hash(x * y) = ac23dc…0。我们假设X的值为5，求y的值大小？用Python实现如下：

from hashlib import sha256
x = 5
y = 0  # y是未知数
while sha256(str(x*y).encode()).hexdigest()[-1] != "0":
    y += 1
print('The solution is y = ' + str(y))

可以看到如下的结果：

The solution is y = 21

在比特币中使用的工作量证明算法叫做，它和我们上面的简单例子相差不大。矿工们为了获得能够创建新的区块权利，需要使用该算法参与计算竞争。通常，问题的难度取决于需要在字符串中查找的字符个数，矿工算出结果后，会获得相应的奖励币。

tips:比特币网络中任何一个节点，如果想生成一个新的区块并写入区块链，必须能够解出比特币网络给出的工作量证明问题。这道题关键的三个要素是工作量证明函数、区块及难度值。工作量证明函数是这道题的计算方法，区块决定了这道题的输入数据，难度值决定了这道题的所需要的计算量，谁最先解出问题，谁就能生成新的区块并写入区块链。

工作量证明简单实现

为我们的区块链实现一个相似的算法，规则跟上述的例子差不多：
找到一个数字p，该数字与前一个块的proof值的hash值以4个0开头

import hashlib
import json

from time import time
from uuid import uuid4

class Blockchain(object):
    ...

    def proof_of_work(self, last_proof):
            """
            简单工作量证明(POW)算法:
             - 找到一个数字p'，使得hash(pp')值的开头包含4个0, p是上一个块的proof,  p'是新的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):
            """
            验证Proof: hash(last_proof, proof)值开头是否包含4个0?
            :param last_proof: <int> 上一个Proof
            :param proof: <int> 当前Proof
            :return: <bool>
            """

            guess = (str(last_proof)+str(proof)).encode()
            guess_hash = hashlib.sha256(guess).hexdigest()
            return guess_hash[:4] == "0000"

为了调整算法的难度，我们可以修改需要匹配的0的个数。但4已经足够了，你会发现增加一个数字会大大增加计算的时间。

我们的Blockchain类差不多完成了，接下来可以开始用http请求进行交互了。

2、Blockchain作为api

我们将使用Python Flask框架，这是一个轻量级的Web应用框架，它能方便的将请求映射到 Python函数，这允许我们通过http请求跟区块链进行交互。

我们会创建3个方法：

/transactions/new 创建一个新的交易，并添加到区块中
/mine 告诉服务器服挖掘一个新的块
/chain 返回整个区块链

设置flask

我们的服务器会作为区块链网络中的一个节点，让我们添加一下代码：

import hashlib
import json
from textwrap import dedent
from time import time
from uuid import uuid4

from flask import Flask


class Blockchain(object):
    ...


# 实例化我们的节点
app = Flask(__name__)

# 为这个节点生成一个全局的唯一地址
node_identifier = str(uuid4()).replace('-', '')

# 实例化区块链
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)

代码相对简单，就不做说明了。

交易请求

用户发给服务器的交易请求如下：

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

我们在Blockchain类中已经定义了添加交易的方法，因此接下来的事情比较简单：

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

    # 检查需要的字段是不是都有
    required = ['sender', 'recipient', 'amount']
    if not all(k in values for k in required):
        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), 200

挖矿

挖矿是里面的神奇所在，而且很简单，只需要做如下三件事：

1 计算工作量证明
2 增加一个交易，授予矿工（自己）一个币
3 构造新区块并将其添加到链中

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():
    # 运行工作量证明算法，获取下一个proof
    last_block = blockchain.last_block
    last_proof = last_block['proof']
    proof = blockchain.proof_of_work(last_proof)

    # 由于找到了proof，我们获得一笔奖励
    # 发送者为"0", 表明是该节点挖出来的新币
    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

3、与我们的区块链交互

你可以通过cURL或Postman去跟API交互。
启动服务器：

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

让我们试着通过Get请求来挖一个块：

mine.png

让我们通过POST请求创建一笔新的交易，post body包含交易的结构信息：

new.png

我重启服务器然后挖了两个块之后，现在总共有三个块(包含创世块)，我们可以通过请求获取所有的块信息。

chain.png

4、一致性

我们已经有了一个可以接受交易以及挖矿的基础区块链。但是区块链系统应该是去中心化的，既然区块链是去中心化，我们该如何确保所有的节点都有同样的链呢？这就是一致性问题，如果想要在我们的区块链网络中运行多个节点，我们就需要实现一个一致性算法。

注册新节点

在实现一致性算法之前，我们需要找到一种方法让一个节点知道它相邻的节点。我们网络中的每个节点都需要保存一份其它节点的记录信息。因此让我们增加一些接口：

1、/nodes/register 接收以url表示的新节点列表
2、/nodes/resolve 实现一致性算法，解决冲突，确保每个节点都有正确的链信息

我们需要修改Blockchain的构造函数，添加一个注册节点的方法：

...
from urlparse import urlparse
...


class Blockchain(object):
    def __init__(self):
        ...
        self.nodes = set()
        ...

    def register_node(self, address):
        """
        添加一个新的节点到节点列表中
        :param address: <str> 节点地址：比如'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):
        """
        确定一个给定的区块链是否有效
        :param chain: <list> 区块链
        :return: <bool> True 有效, False 无效
        """

        last_block = chain[0]
        current_index = 1

        while current_index < len(chain):
            block = chain[current_index]
            # 检查block的hash值是否正确
            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 resolve_conflicts(self):
        """
        一致性算法，通过将我们的链替换成网络中最长的链来解决冲突
        :return: <bool> True 我们的链被取代, 否则为False
        """

        neighbours = self.nodes
        new_chain = None

        # 我们只查看比我们链长的节点
        max_length = len(self.chain)

        # Grab and verify the chains from all the nodes in our network
        for node in neighbours:
            response = requests.get('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

valid_chain()方法主要负责检查链是否有效，历遍链中的所有块，检查块的hash值以及工作量证明是否正确。

resolve_conflicts()负责解决冲突，通过历遍附近节点，通过valid_chain方法验证该节点的链是否正确，如果该节点链正确且该链比我们节点的链长度长，则用该链替换我们的链。

让我们添加两个方法，一个用来注册附近节点，一个用来解决冲突：

@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

你可以在不同的机器上运行节点，这里我们在一台机器上启动不同端口来代表不同的节点。假设我们有两个节点：
和 。

运行两个节点，并将第二节点注册到第一个节点中：

register.png

然后在第二个节点(5001端口)挖了一些块，确保该节点的链长度比第一个节点长。然后在第一个节点调用GET /nodes/resolve请求，可以看到节点1的链通过一致性算法被替换掉了。

resolve.png

嗯。。现在可以邀请你的朋友来测试你的区块链了。可以到这里查看

参考：