加载中…
正文 字体大小:

从零开始创建一个区块链应用(Python版)

(2018-04-01 18:52:11)
标签:

区块链

python

分类: 区块链

2018年什么最火?非区块链莫属!

一时间网上各种介绍区块链的文章层出不穷,但大多数都是从概念层面进行解释的,本文则从技术层面讲解,如何从零开始创建一个区块链应用。


本文使用Python开发,读者需要了解一些Python语言的基础知识。


首先创建一个Python文件blockchain.py,所有源代码都保存在该文件中。

接着创建一个 Blockchain 类,在构造函数中创建两个列表,一个用于储存区块链,另一个用于储存交易。

---------------------------------------------------------------------------------------------------

class Blockchain(object):

    def __init__(self):

        self.chain = [] # 存储区块链

        self.current_transactions = [] # 存储交易

        self.init_chain() # 初始化区块链

---------------------------------------------------------------------------------------------------

接下来我们创建一个区块,每个区块包含属性:索引(index)、Unix 时间戳(timestamp)、交易列表(transactions)、工作量证明(proof)和前一个区块的 hash 值(previous_hash)。

---------------------------------------------------------------------------------------------------

def new_block(self, proof, previous_hash=None):

    """

    创建新区块

    :param proof: 工作量

    :param previous_hash: 前一个区块的 hash 值

    :return: 新区块

    """

    block = {

        'index': len(self.chain) + 1, # 索引

        'timestamp': time(), # 时间戳

        'transactions': self.current_transactions, # 交易列表

        'proof': proof, # 工作量证明

        'previous_hash': previous_hash or self.hash(self.chain[-1]), # 前一个区块的 hash 值

    }


    self.current_transactions = [] # 清空当前的交易列表

    self.chain.append(block) # 新区块添加到链尾

    return block


def hash(block):

    """

    生成块的 SHA-256 hash值

    :param block: Block

    :return:

    """

    # 确保块字典是排好序的, 否则不能得到一致性hash值

    block_string = json.dumps(block, sort_keys=True).encode()

    return hashlib.sha256(block_string).hexdigest()

---------------------------------------------------------------------------------------------------

每一个区块都包含前一个区块的 hash 值,这是关键的一点,它保障了区块链的不可变性。如果攻击者破坏了前面的某个区块,那么后面所有区块的 hash 值都会变得不正确。


接下来我们加入交易模块

---------------------------------------------------------------------------------------------------

def new_transaction(self, sender, recipient, amount):

    """

    生成新的交易记录, 将其加入到下一个区块中

    :param sender: 发送者

    :param recipient: 接受者

    :param amount: 数量

    :return: 记录此交易的区块索引

    """

    self.current_transactions.append({

        'sender': sender,

        'recipient': recipient,

        'amount': amount,

    })

    return self.last_block['index'] + 1

---------------------------------------------------------------------------------------------------

理解工作量证明

新的区块依赖工作量证明算法(PoW)来构造。PoW 的目标是找出一个符合特定条件的数字,这个数字很难计算出来,但容易验证。这就是工作量证明的核心思想。

在比特币中,使用称为 Hashcash(https://en.wikipedia.org/wiki/Hashcash) 的工作量证明算法,矿工们为了争夺创建区块的权利而争相计算结果。通常,计算难度与目标字符串需要满足的特定字符的数量成正比,矿工算出结果后,会获得比特币奖励。当然,一旦计算出来,很容易验证这个结果。


实现我们自己的工作量证明

我们的工作量证明非常简单,就是验证用户输入的成语是否有效以及该成语开始汉字和区块链最后一个区块中成语的结束汉字是否相同(即成语接龙规则)。

---------------------------------------------------------------------------------------------------

def valid_proof(last_proof, proof):

    """

    验证工作量证明

    :param last_proof: 前一个工作量

    :param proof: 当前工作量

    :return: True:有效 , False:无效.

    """

    params = {

        'q': proof,

        't': 'ChengYu'

    }

    # 验证是否是成语

    url = "http://chengyu.t086.com/chaxun.php?" + urlencode(params, encoding="GBK")

    try:

        r = requests.get(url, timeout=10)

        r.raise_for_status()

        r.encoding = "GBK"

        content = r.text

    except:

        content = None


    if content is None or content.find("没有找到与您搜索相关的成语") > -1 or content.find("搜索词太长") > -1:

        return False


 

    # 验证是否满足成语接龙的规则

    return last_proof[-1] == proof[0]

---------------------------------------------------------------------------------------------------

到目前为止,Blockchain 类已经基本完成了,接下来我们创建用户挖矿的界面。

我们使用 Python Flask 框架,这是一个轻量级 Web 应用框架,它方便将网络请求映射到 Python 函数,现在我们来让 Blockchain 运行在 Flask Web 上。


创建节点

我们的“Flask 服务器”将扮演区块链网络中的一个节点,首先我们添加一些框架代码:

---------------------------------------------------------------------------------------------------

from flask import Flask, jsonify, request, session, render_template

# 创建我们的节点

app = Flask(__name__)

app.config['SECRET_KEY'] = os.urandom(32)

blockchain = Blockchain()


欢迎页

@app.route('/')

def index():

    return render_template('index.html')


挖矿页

@app.route('/chain', methods=['POST'])

def chain():

    phone = request.form['phone']

    if phone is None or phone == "":

        return '缺少参数[phone]', 400

    session["phone"] = phone

    blockchain.resolve_conflicts()


    proofs = [block["proof"] for block in blockchain.chain]


    return render_template('chain.html', phone=phone, proofs=proofs)


挖矿

@app.route('/mine', methods=['POST'])

def mine():

   一致性算法解决冲突,后面会讲到

    blockchain.resolve_conflicts()


    last_block = blockchain.last_block

    last_proof = last_block['proof']


    proof = request.form['answer'].strip()

    if last_proof[-1] != proof[0]:

        proof_chain = ""

        for block in blockchain.chain:

            proof_chain += block["proof"] + " -> "

        response = {

            'message': "非常遗憾!你晚了一步,已经有其他用户回答了这个成语",

            'proof_chain': proof_chain

        }

        return jsonify(response), 200


    if blockchain.valid_proof(last_proof, proof):

        blockchain.new_transaction(

            sender="*",  # "*" 表示新挖出的币

            recipient=session["phone"],

            amount=1,

        )

        # 在区块链的末尾加入新块

        block = blockchain.new_block(proof)

        blockchain.write()


        proof_chain = ""

        for block in blockchain.chain:

            proof_chain += block["proof"] + " -> "


        response = {

            'message': "恭喜你接龙成功!",

            'proof_chain': proof_chain

        }

    else:

        response = {

            'message': "你输入的不是成语,请重新输入..."

        }


    return jsonify(response), 200

---------------------------------------------------------------------------------------------------


这是节点的欢迎页

从零开始创建一个区块链应用(Python版)

在欢迎页我们需要输入手机号作为个人的身份认证。


接下来是挖矿页

在这个页面你将会看到所有区块的工作量证明,输入你将要接龙的成语进行挖矿。

从零开始创建一个区块链应用(Python版)


挖矿成功后,你将会拥有该成语(即工作量证明)。

从零开始创建一个区块链应用(Python版)


至此,我们的区块链应用已经可以挖矿了,但是区块链系统是分布式的。既然是分布式的,那么我们究竟该如何保证所有节点拥有同样的链呢?这就是一致性问题,我们要想在网络上有多个节点,就必须实现一个一致性算法。

在一致性算法中我们规定最长且有效的链才是最终链,其它链将会被抛弃。

---------------------------------------------------------------------------------------------------

def resolve_conflicts(self):

    """

    一致性算法解决冲突

    使用网络中最长的链.

    :return: True:链被取代, False:链未被取代

    """

    new_chain = self.chain

    # 寻找全网链长最大的区块链

    max_length = len(self.chain)

    replaced = False

    # 遍历全网所有节点获取区块链,在中本聪的《比特币:一种点对点的电子现金系统》论文中没有给出该如何获取全网所有节点,我们这里采用遍历区域网的方式实现

    for i in range(1, 256):

        try:

            response = requests.get(f'http://192.168.0.{i}:5000/gainBlockChain', timeout=30)

            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

                    replaced = True

                # 链长等于当前最大且有效的区块链, 对比最后一个block的创建时间

                elif length == max_length and self.valid_chain(chain):

                    if float(chain[-1]["timestamp"]) < float(new_chain[-1]["timestamp"]):

                        new_chain = chain

                        replaced = True

        except Exception as e:

            print(e)


    # 如果发现新的链长大于自己且有效的区块链则替换自己的

    if replaced:

        self.chain = new_chain

        self.write()

    return replaced


获取节点的区块链

@app.route('/gainBlockChain', methods=['GET'])

def full_chain():

    response = {

        'chain': blockchain.chain,

        'length': len(blockchain.chain),

    }

    return jsonify(response), 200


def valid_chain(self, chain):

    """

    验证区块链的有效性

    :param chain: 待验证的区块链

    :return: True:有效, False:无效

    """

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

        # 验证块的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


data_dir = "/opt/blockchain"

def write(self):

    '''

    将区块链信息写入本地磁盘

    :return:

    '''

    f = open(data_dir + "/data.txt", "w", encoding="utf-8")

    for block in self.chain:

        f.write(str(block["index"]))

        f.write("#")

        f.write(block["proof"])

        f.write("#")

        f.write(str(block["previous_hash"]))

        f.write("#")

        f.write(str(block["timestamp"]))

        f.write("#")

        transactions = block["transactions"]

        if transactions and len(transactions) > 0:

            f.write(transactions[-1]["sender"])

            f.write("#")

            f.write(transactions[-1]["recipient"])

            f.write("#")

            f.write(str(transactions[-1]["amount"]))

        else:

            f.write("*#*#1")

        f.write("\n")

    f.close()

---------------------------------------------------------------------------------------------------

在创建新的节点时,我们会调用resolve_conflicts()方法来解决冲突。


这是创建节点时初始化区块链的方法,里面包含了创世块的创建,这是非常关键的一步。

---------------------------------------------------------------------------------------------------

def init_chain(self):

    # 读取本地存储的区块链

    if len(self.chain) == 0:

        try:

            f = open(data_dir + "/data.txt", "r", encoding="utf-8")

            for line in f.readlines():

                items = line.split("#")

                self.current_transactions = []

                transaction = {

                    'sender': items[4],

                    'recipient': items[5],

                    'amount': int(items[6]),

                }

                self.current_transactions.append(transaction)

                block = {

                    'index': int(items[0]),

                    'timestamp': float(items[3]),

                    'transactions': self.current_transactions,

                    'proof': items[1],

                    'previous_hash': items[2],

                }

                self.chain.append(block)

            f.close()

        except:

            pass


    # 遍历全网下载链长最长且有效的区块链

    self.resolve_conflicts()


    # 创建创世快

    if len(self.chain) == 0:

        self.new_block(proof="海阔天空", previous_hash="*")

        # self.new_transaction(sender="*", recipient="*", amount=1)

        if not os.path.exists(data_dir):

            os.mkdir(data_dir)

        self.write()

---------------------------------------------------------------------------------------------------

好啦,我们的区块链应用可以挖矿了,你可以邀请小伙伴们一起来测试了。


尚未解决的问题点:

1. 时间戳服务器: 在中本聪的《比特币:一种点对点的电子现金系统》论文中给出了时间戳服务器的概念,不过在我们的实现代码中,仅采用了各个节点自身的时间生成的时间戳,这是一个需要解决的问题点。

2. 区块链全网所有节点: 在我们的实现代码中,我们是采用遍历区域网的方式实现的,而在真实的场景下,需要记录所有的节点用于实现一致性算法。


源代码: https://github.com/haidawyl/blockchain/blob/master/blockchain-python/blockchainV2.py

参考: http://blog.51cto.com/wuwenhua/2084615


0

阅读 评论 收藏 转载 喜欢 打印举报
前一篇:搭建SolrCloud
  • 评论加载中,请稍候...
发评论

    发评论

    以上网友发言只代表其个人观点,不代表新浪网的观点或立场。

    < 前一篇搭建SolrCloud
      

    新浪BLOG意见反馈留言板 电话:4006900000 提示音后按1键(按当地市话标准计费) 欢迎批评指正

    新浪简介 | About Sina | 广告服务 | 联系我们 | 招聘信息 | 网站律师 | SINA English | 会员注册 | 产品答疑

    新浪公司 版权所有