Python初探,构建你的第一个简单区块链应用

时间: 2026-02-26 0:45 阅读数: 5人阅读

区块链技术作为近年来备受瞩目的创新,以其去中心化、不可篡改和透明可追溯的特性,在金融、供应链、数字版权等多个领域展现出巨大潜力,虽然区块链的底层原理复杂,但我们可以借助Python这种简洁易学的编程语言,亲手搭建一个简单的区块链应用,从而更直观地理解其核心概念。

本文将带你一步步用Python实现一个基础的区块链,包括区块结构、链的构建、工作量证明(PoW)以及简单的共识机制。

准备工作:环境与基础概念

在开始编码前,确保你已经安装了Python(推荐3.6及以上版本),我们将主要使用Python内置的hashlib库来进行哈希运算,json库用于数据序列化,以及time库来记录时间戳。

核心概念回顾:

  1. 区块(Block):区块链的基本单元,包含区块头和区块体,区块头包含前一个区块的哈希值、当前区块的哈希值、时间戳、目标难度以及一个随机数(用于工作量证明),区块体存储实际的数据(例如交易信息)。
  2. 链(Chain):由多个区块按顺序连接而成,每个区块都通过其包含的“前一个区块哈希值”指向前一个区块,形成一条不可逆的链条。
  3. 哈希(Hash):将任意长度的输入数据转换为固定长度的输出值的函数,区块链中常用SHA-256哈希算法,它能确保数据的微小改动都会导致哈希值发生巨大变化,从而保证数据的完整性。
  4. 工作量证明(Proof of Work, PoW):一种共识机制,要求节点(矿工)通过大量的计算工作找到一个符合特定条件的随机数(Nonce),使得当前区块头的哈希值满足一定的前导零数量,这可以有效防止恶意篡改,因为篡改数据需要重新计算后续所有区块的工作量,成本极高。

构建区块结构

我们定义一个Block类来表示区块链中的每一个区块。

import hashlib
import json
from time import time
class Block:
    def __init__(self, index, previous_hash, transactions, timestamp=None, nonce=0):
        self.index = index              # 区块号
        self.previous_hash = previous_hash  # 前一个区块的哈希
        self.timestamp = timestamp or time()  # 时间戳
        self.transactions = transactions    # 区块数据(交易列表)
        self.nonce = nonce              # 工作量证明的随机数
    def compute_hash(self):
        """
        计算区块的哈希值
        """
        # 将区块内容序列化为JSON字符串,并确保键的顺序一致,以保证哈希一致性
        block_string = json.dumps({
            "index": self.index,
            "previous_hash": self.previous_hash,
            "timestamp": self.timestamp,
            "transactions": self.transactions,
            "nonce": self.nonce
        }, sort_keys=True).encode()
        return hashlib.sha256(block_string).hexdigest()

构建区块链

我们创建一个Blockchain类来管理整条链,它需要包含创世区块(Genesis Block,即第一个区块)、添加新区块的方法,以及验证链的有效性的方法。

class Blockchain:
    def __init__(self):
        self.chain = []
        self.current_transactions = []  # 存储当前待打包的交易
        self.difficulty = 4  # 工作量难度,即哈希值前导零的个数
        self.create_genesis_block()
    def create_genesis_block(self):
        """
        创建创世区块
        """
        genesis_block = Block(
            index=1,
            previous_hash="0",  # 创世区块没有前一个区块,哈希值为"0"
            transactions=[{"message": "Genesis Block"}],
            timestamp=time()
        )
        # 创世区块不需要工作量证明,或者可以简单挖矿
        genesis_block.hash = genesis_block.compute_hash()
        self.chain.append(genesis_block)
    @property
    def last_block(self):
        """
        返回链上的最后一个区块
        """
        return self.chain[-1]
    def add_transaction(self, sender, recipient, amount):
        """
        添加新的交易到待打包列表
        """
        self.current_transactions.append({
            "sender": sender,
            "recipient": recipient,
            "amount": amount
        })
        # 返回该交易将被打包的区块的索引
        return self.last_block['index'] + 1
    def proof_of_work(self, block):
        """
        工作量证明:找到一个nonce,使得区块的哈希值满足前difficulty个零
        """
        block.nonce = 0
        computed_hash = block.compute_hash()
        while not computed_hash.startswith('0' * self.difficulty):
            block.nonce += 1
            computed_hash = block.compute_hash()
        return computed_hash
    def add_block(self, block, proof):
        """
        添加一个新区块到链中
        """
        previous_hash = self.last_block.hash
        if previous_hash != block.previous_hash:
            return False
        if not self.is_valid_proof(block, proof):
            return False
        block.hash = proof
        self.chain.append(block)
        self.current_transactions = []  # 重置当前交易列表
        return True
    def is_valid_proof(self, block, block_hash):
        """
        验证工作量证明是否有效
        """
        return block_hash.startswith('0' * self.difficulty) and block_hash == block.compute_hash()
    def is_chain_valid(self):
        """
        验证整个区块链的有效性
        """
        for i in range(1, len(self.chain)):
            current_block = self.chain[i]
            previous_block = self.chain[i-1]
            if current_block.previous_hash != previous_block.hash:
                return False
            if not self.is_valid_proof(current_block, current_block.hash):
                return False
        return True

简单的交互与测试

我们可以实例化我们的区块链,并进行一些简单的操作,比如添加交易、挖矿(创建新区块)以及验证链的有效性。

if __name__ == "__main__":
    # 创建一个区块链实例
    my_blockchain = Blockchain()
    print("创世区块已创建!")
    print(f"创世区块: {my_blockchain.chain[0].__dict__}")
    # 添加一些交易
    print("\n添加交易...")
    my_blockchain.add_transaction("Alice", "Bob", 10)
    my_blockchain.add_transaction("Bob", "Charlie", 5)
随机配图
# 挖矿,创建新区块 print("\n开始挖矿...") last_block = my_blockchain.last_block new_block = Block( index=last_block.index + 1, previous_hash=last_block.hash, transactions=my_blockchain.current_transactions ) proof = my_blockchain.proof_of_work(new_block) my_blockchain.add_block(new_block, proof) print(f"新区块已添加到链中!") print(f"新区块: {new_block.__dict__}") # 打印整个区块链 print("\n当前区块链:") for block in my_blockchain.chain: print(f"区块 {block.index}:") print(f" 前一个区块哈希: {block.previous_hash}") print(f" 区块哈希: {block.hash}") print(f" 交易: {block.transactions}") print(f" 时间戳: {block.timestamp}") print("-" * 30) # 验证区块链 print("\n验证区块链...") if my_blockchain.is_chain_valid(): print("区块链有效!") else: print("区块链无效!") # 尝试篡改一个区块 print("\n尝试篡改区块1的交易...") my_blockchain.chain[1].transactions[0] = {"message": "篡改后的交易"} if my_blockchain.is_chain_valid(): print("篡改后区块链仍然有效(这不应该发生!)") else: print("篡改被检测到,区块链无效!")

代码解析与扩展

  1. Block:封装了区块的属性和方法,特别是compute_hash()用于计算区块的哈希值。
  2. Blockchain
    • create_genesis_block():创建并添加创世区块。
    • add_transaction():模拟添加交易,这些交易会暂存current_transactions中,直到被打包进新区块。
    • proof_of_work():核心的挖矿过程,通过不断调整nonce来寻找满足难度条件的哈希值。
    • add_block():将经过挖矿验证的新区块添加到链中,并检查其与前一个区块的连接是否正确。
    • is_valid_proof():验证某个区块的哈希值是否符合工作量证明的要求。
    • is_chain_valid():遍