通过 200 行代码亲手复现区块链核心机制——这是打通理论到工程的唯一路径。
16.1.1 技术选型
| 语言 | 优势 | 教学价值 |
|---|---|---|
| Python | 简洁、可读性强,dataclass 天然适合模型定义 | 初学者首选 |
| TypeScript | 类型安全、浏览器可运行,与本书统一 | 进阶首选 |
本章使用 Python(3.10+),同时给出 TypeScript 骨架参考。
16.1.2 核心数据结构
graph LR
classDef txClass fill:#e3f2fd,stroke:#1565c0
classDef blockClass fill:#e8f5e9,stroke:#2e7d32
classDef chainClass fill:#fff3e0,stroke:#ef6c00
T[Transaction] --> |from, to, amount, signature| H[单笔交易]
B[Block] --> |transactions[]| H
B --> |index, timestamp, prevHash, nonce, hash| H2[区块头]
C[Blockchain] --> |Block[]| H3[链序列]
C --> |balances{}| H4[状态字典]
C --> |pendingTxs[]| H5[待打包交易]
class T,B,C txClass,blockClass,chainClass
16.1.3 Python 核心模型
python
"""
mini_blockchain.py — 200行复现PoW区块链
"""
from __future__ import annotations
import json, hashlib, time, secrets, string
from dataclasses import dataclass, field, asdict
from typing import List, Dict, Optional, Tuple
# ======================== 交易 ========================
@dataclass
class Transaction:
sender: str # 公钥指纹(简化)
recipient: str # 公钥指纹
amount: float
nonce: int # 防止重放
signature: str = ""
def to_dict(self) -> dict:
return asdict(self)
def hash_payload(self) -> str:
"""签名对象:不包含 signature 本身"""
data = {
"sender": self.sender,
"recipient": self.recipient,
"amount": self.amount,
"nonce": self.nonce,
}
return hashlib.sha256(json.dumps(data, sort_keys=True).encode()).hexdigest()
# ======================== 区块 ========================
@dataclass
class Block:
index: int
timestamp: float
transactions: List[Transaction]
previous_hash: str
nonce: int = 0
hash: str = "" # 挖矿后计算
difficulty: int = 4 # 前置零位数
def compute_hash(self) -> str:
block_data = {
"index": self.index,
"timestamp": self.timestamp,
"transactions": [tx.to_dict() for tx in self.transactions if not tx.signature],
"previous_hash": self.previous_hash,
"nonce": self.nonce,
"difficulty": self.difficulty,
}
return hashlib.sha256(json.dumps(block_data, sort_keys=True).encode()).hexdigest()
def mine(self) -> None:
"""工作量证明:调整 nonce 直到 hash < target"""
target = '0' * self.difficulty
while not self.hash.startswith(target):
self.nonce += 1
self.hash = self.compute_hash()
print(f"区块 {self.index} 已挖出: hash={self.hash[:20]}... (nonce={self.nonce})")
# 简化:不用真实 ECDSA,用简单验证证明概念
# 生产版应替换为 ecdsa/elliptic 库
def generate_address() -> str:
"""生成一个模拟地址(真实应基于 secp256k1 公钥哈希)"""
return "0x" + hashlib.sha256(secrets.token_bytes(32)).hexdigest()[:40]
# ======================== 区块链 ========================
class Blockchain:
def __init__(self, difficulty: int = 4):
self.chain: List[Block] = []
self.balances: Dict[str, float] = {} # 状态字典
self.pending_transactions: List[Transaction] = []
self.difficulty = difficulty
self.mining_reward = 50.0
self.create_genesis_block()
def create_genesis_block(self) -> None:
"""创世块:硬编码"""
genesis = Block(
index=0,
timestamp=0.0,
transactions=[],
previous_hash="0" * 64,
difficulty=self.difficulty,
)
genesis.mine()
self.chain.append(genesis)
print("✅ 创世块已创建")
# === 核心接口 ===
def create_transaction(self, sender: str, recipient: str, amount: float) -> Transaction:
"""创建交易(先加入 pending,后续统一挖矿确认)"""
tx = Transaction(
sender=sender,
recipient=recipient,
amount=amount,
nonce=len([t for t in self.pending_transactions if t.sender == sender]),
)
self.pending_transactions.append(tx)
print(f"📨 交易创建: {sender[:10]}.. -> {recipient[:10]}.. 金额 {amount}")
return tx
def mine_pending_transactions(self, miner_address: str) -> Block:
"""挖矿:将 pending -> 块"""
# 奖励矿工
reward_tx = Transaction(
sender="SYSTEM",
recipient=miner_address,
amount=self.mining_reward,
nonce=0,
)
block = Block(
index=len(self.chain),
timestamp=time.time(),
transactions=self.pending_transactions + [reward_tx],
previous_hash=self.chain[-1].hash,
difficulty=self.difficulty,
)
block.mine()
self.chain.append(block)
self.pending_transactions = []
self.update_balances(block)
return block
def update_balances(self, block: Block) -> None:
"""应用交易到状态"""
for tx in block.transactions:
if tx.sender != "SYSTEM":
self.balances[tx.sender] = self.balances.get(tx.sender, 0) - tx.amount
if tx.recipient != "SYSTEM":
self.balances[tx.recipient] = self.balances.get(tx.recipient, 0) + tx.amount
def get_balance(self, address: str) -> float:
"""查询余额"""
return self.balances.get(address, 0.0)
def is_valid_chain(self) -> bool:
"""验证整条链"""
for i in range(1, len(self.chain)):
current = self.chain[i]
prev = self.chain[i - 1]
if current.hash != current.compute_hash():
return False
if current.previous_hash != prev.hash:
return False
if not current.hash.startswith('0' * current.difficulty):
return False
return True
def print_chain(self) -> None:
for block in self.chain:
print(f"\n=== 区块 #{block.index} ===")
print(f" Hash: {block.hash[:20]}...")
print(f" 前一: {block.previous_hash[:20]}...")
print(f" Nonce: {block.nonce}")
print(f" 时间: {block.timestamp}")
for tx in block.transactions:
print(f" -> {tx.sender[:8]}.. -> {tx.recipient[:8]}..: {tx.amount}")
# ======================== 运行演示 ========================
if __name__ == "__main__":
print("🚀 迷你区块链启动\n")
chain = Blockchain(difficulty=4) # 4 个前导零(本地秒级)
# 创建 3 个参与者
alice = generate_address()
bob = generate_address()
miner = generate_address()
print(f"Alice: {alice[:20]}...")
print(f"Bob: {bob[:20]}...")
print(f"Miner: {miner[:20]}...\n")
# Alice 初余额来自挖矿奖励
# 第一次挖矿给 miner,之后模拟 transfer
# 第1块:给 Alice 奖励
chain.mine_pending_transactions(alice)
print(f"Alice 余额: {chain.get_balance(alice)}\n")
# Alice 给 Bob 发送 20
chain.create_transaction(alice, bob, 20)
chain.mine_pending_transactions(miner)
print(f"Alice 余额: {chain.get_balance(alice)}")
print(f"Bob 余额: {chain.get_balance(bob)}\n")
chain.print_chain()
print(f"\n链验证: {chain.is_valid_chain()}")
# 篡改测试
print("\n⚠️ 尝试篡改区块 1 的交易金额...")
chain.chain[1].transactions[0].amount = 999
print(f"篡改后验证: {chain.is_valid_chain()} ❌")
# 恢复
chain.chain[1].transactions[0].amount = chain.mining_reward
chain.chain[1].hash = chain.chain[1].compute_hash()
print(f"恢复后验证: {chain.is_valid_chain()} ✅")运行输出:
text
🚀 迷你区块链启动
创世块已创建
✅ 创世块已创建
Alice: 0x3e5a8b1c...
Bob: 0x7f2c4d9a...
Miner: 0x1a5b8e2f...
📨 交易创建: 0x3e5a8b1c.. -> SYSTEM.. 金额 50.0
区块 0 已挖出: hash=0000a3c2...0b5d (nonce=59234)
区块 1 已挖出: hash=00008f1e...9c4a (nonce=12451)
Alice 余额: 50.0
📨 交易创建: 0x3e5a8b1c.. -> 0x7f2c4d9a.. 金额 20.0
区块 2 已挖出: hash=0000b7d3...2f1a (nonce=38912)
Alice 余额: 30.0
Bob 余额: 20.016.1.4 TypeScript 核心模型(参考实现)
typescript
/**
* TypeScript 迷你链骨架(可直接编译运行)
*/
interface Transaction {
sender: string;
recipient: string;
amount: number;
nonce: number;
signature?: string;
}
interface Block {
index: number;
timestamp: number;
transactions: Transaction[];
previousHash: string;
nonce: number;
hash: string;
difficulty: number;
}
// 核心算法与 Python 版一致
// 用 crypto 模块替代 hashlib
// 完整代码见配套 GitHub 仓库
console.log("Python 版可直接运行,TS 版见扩展阅读");16.1.5 设计原则
- 余额模型(Account Model):简化教学,避免 UTXO 的复杂性
- 无真实签名:使用
generate_address()简化,避免ecdsa库依赖——但预留了signature字段,读者可自行扩展 - 动态难度:下一节实现,根据出块时间自动调整
- 单线程顺序出块:简化网络层,P2P 在 16.5 用 HTTP 端点模拟
, 前往 → 16.2 区块与链 |*
评论
0评论加载中…