我早期接触区块链的时候总觉得这玩意儿被神化得太厉害了。后来花了几天时间用Python把一条极简区块链从头撸了一遍才彻底想明白所谓区块链核心不过就是三件事——把数据放进结构体、给结构体算一个独一无二的指纹、再把一串结构体按指纹顺序串起来。这篇文章就是我那次从零开始“拆玩具”的完整记录包含区块结构、SHA-256哈希、工作量证明、Flask接口以及我一路上踩过的坑。它适合刚学完Python基础、想看看“区块链到底怎么用代码实现”的读者你不需要懂密码学或分布式系统跟着把代码敲一遍概念自然就通了。1. 先想清楚一个区块里到底装什么1.1 把账本切成“块”区块这个词听起来玄乎你完全可以把它想成一页账本。手里有一笔接一笔的交易记录为了让别人查起来方便、又能防止事后篡改就把这些交易按时间打包成一页一页的“块”每一页就是一个区块Block。我在设计这个教学项目时给每个区块塞了六个字段index、timestamp、transactions、previous_hash、nonce以及最后算出来的当前区块哈希hash。为什么是这六个字段而不是更少因为每个字段回答的都是一个非常具体的问题。字段作用关键点index当前区块是第几个从0开始创世区块是0timestamp区块打包的时间用时间戳表示方便排序和校验transactions区块里实际存放的交易数据教学版用一个字典列表previous_hash前一个区块的哈希这是“链”能连起来的核心nonce挖矿时调整的随机数工作量证明的“解题答案”hash当前区块的数字指纹由上面所有字段计算得出你可以把这六个字段分成三组index和timestamp是区块的基本身份信息transactions是真正的业务数据previous_hash、nonce和hash则承担了“链式验证”的工作。前两个好理解后面三个得专门讲一讲尤其是哈希。1.2 哈希是整条链的“防篡改感应器”哈希hash本质上是一串固定长度的十六进制字符串它是通过一个叫SHA-256的算法算出来的。Python的hashlib库就内置了sha256一行代码就能调用。但理解哈希比会调用更重要因为整条链的安全性都建立在哈希的两个特性上。第一个特性是“雪崩效应”哪怕输入只改变一个字符算出来的整个哈希都会面目全非。第二个特性是“单向性”从哈希值几乎不可能反推出原始输入内容。这两个特性结合起来让哈希成了一个完美的“篡改感应器”。举个例子。假设某一笔交易的金额本来是100有人悄悄把它改成了101哪怕只改一个字符这个区块重新计算出的哈希就会变得完全不一样。因为下一个区块里记录的previous_hash还指向老的哈希两个哈希一对比马上就能发现这个区块被动过手脚整个链的验证也自然失败。我一开始也会问那如果有人把篡改后的区块以及它后面所有区块的previous_hash全都重新算一遍不就能蒙混过关了这就是工作量证明要解决的问题后面第3部分专门讲。现在你只需要先记住哈希保证了“数据改动能被立刻察觉”。2. Python实现从环境准备到区块类2.1 Python环境与依赖到底需要装些什么这个教学项目对Python版本要求很宽松Python 3.7以上基本都能跑。如果你还没配置环境直接去Python官网下载对应系统的安装包安装的时候记得勾选“Add Python to PATH”这样在终端里输入python --version就能看到版本号。常见的一类报错是python was not found多半是安装时没勾PATH或者电脑装了多个Python版本导致命令指向混乱把安装时的勾选补上就行。官方下载地址会因系统而异但通常搜索结果第一位就是官网。要是你用的是Linux系统可能自带了Python用python3 --version确认。我这里建议在项目目录里创建一个虚拟环境避免以后装其他库时互相冲突。执行下面两条命令就行python -m venv venv source venv/bin/activate # Windows下是 venv\Scripts\activate这个项目真正依赖的第三方库只有两个一个是Flask用来提供HTTP接口另一个是requests做多节点测试的时候会用到。requests可能在你测试共识时才需要但两者都推荐顺手装上后面不折腾。pip install flask requests用到的标准库有hashlib、json、time这些是Python自带的不需要额外安装。整个项目核心代码大约一百行左右非常适合逐行阅读。2.2 区块类的实现以及那些容易被忽略的细节区块我用一个简单的类来表示。每个区块在初始化的时候就计算出自己的哈希后续如果要挖矿会重算哈希。import hashlib import json import time class Block: def __init__(self, index, transactions, timestamp, previous_hash, nonce0): self.index index self.transactions transactions self.timestamp timestamp self.previous_hash previous_hash self.nonce nonce self.hash self.compute_hash() def compute_hash(self): block_string json.dumps({ index: self.index, transactions: self.transactions, timestamp: self.timestamp, previous_hash: self.previous_hash, nonce: self.nonce }, sort_keysTrue).encode() return hashlib.sha256(block_string).hexdigest()这里有个极其重要的细节就是json.dumps里的sort_keysTrue。我最初写这段代码时没有加它结果在本地跑一切正常一旦把链的数据从A节点传给B节点再算哈希两边怎么都对不上。原因是Python字典是保持插入顺序的同样的内容插入顺序不同序列化出来的字符串就不同哈希自然不同。加了这个参数就强制所有节点都按照相同的键顺序去序列化这样“同一份数据必然得到同一个哈希”。还有一处要注意json.dumps(...).encode()默认使用UTF-8编码字节串。这里必须转成字节串因为hashlib.sha256接收的是字节不是字符串。如果你输入的是字符串会直接抛TypeError这也是新手比较容易遇见的问题。至于timestamp为什么用time.time()返回的浮点数我只能说教学版图省事。浮点数在跨节点传输时存在精度问题你如果只是想本地跑通无所谓但一旦要做多节点实验建议换成整数时间戳否则在某些边界情况下哈希验证会莫名其妙失败。这一点在文章第5部分还会单独提。2.3 区块链类、交易池和创世区块区块有了链本身更好理解。区块链类维护三个核心状态一整条链的数据、一个待打包交易池以及当前挖矿难度。class Blockchain: def __init__(self): self.chain [] self.pending_transactions [] self.difficulty 4 self.create_genesis_block() def create_genesis_block(self): genesis Block(0, [], time.time(), 0) genesis.hash self.proof_of_work(genesis) self.chain.append(genesis) def add_transaction(self, sender, recipient, amount): self.pending_transactions.append({ sender: sender, recipient: recipient, amount: amount }) # proof_of_work 和 is_chain_valid 会在下一部分补充创世区块是区块链的起点它没有前一个区块可以指。所以它的previous_hash我用字符串0代替。这个块通常不走普通挖矿流程而是先创建再单独挖一次矿这样后面每个块才有一个合法的“祖先”。pending_transactions就是“待打包交易池”。在真实系统里用户发起交易后交易会先进入一个池子矿工批量打包。所以这里我刻意把“新增交易”和“挖矿”分成两个动作而不是一有交易立刻打包这个设计更贴近真实区块链的行为。3. 挖矿与工作量证明算力和随机数的平衡3.1 为什么“挖矿”只是在反复猜数字工作量证明Proof of WorkPoW是这个教学项目里看着最绕的部分其实逻辑极其简单。我们给新区块定一个目标它的哈希字符串必须以若干个0开头。0的个数就是难度。如果当前这个nonce算出来的哈希不满足就把nonce加1重新算一遍直到撞上一个满足条件的为止。def proof_of_work(self, block): 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注意这段代码里“改nonce、重算哈希”是唯一的办法。因为SHA-256是单向的你没法通过分析输入推测“什么值会产生以0000开头的哈希”只能一次一次试。nonce本身没有任何业务含义它就是一个为了让最终哈希满足条件的“抽奖号码”。挖矿消耗的算力本质上就是消耗在这无数次哈希计算里。有一个写得比较顺手的表达是while computed_hash[:self.difficulty] ! 0 * self.difficulty效果和startswith一样但startswith读起来更直白也不容易把切片边界写错。3.2 难度参数怎么定才合理哈希字符串的每一位都是十六进制字符每一位上有16种可能。难度为4要求前4位全部为0那么单次算出的哈希满足条件的概率是(1/16)^4 1/65536也就是说平均大约需要65536次哈希尝试才能挖出一个合法区块。如果你的机器每秒能算5万到10万次SHA-256那平均1秒左右能产出一个块。这个体验刚刚好能感受到“挖矿需要计算”又不用等太久。如果把难度调到5概率变成(1/16)^5约等于1/1048576也就是平均要试一百万次时间会拉长到10到20秒。调到6就是一千六百多万次可能要几分钟。所以教学演示项目把难度保持在4或5最舒服。太低了没有“挖矿感”太高了你会误以为程序卡死。难度平均尝试次数教学演示体验34096几乎秒出没有挖矿的感觉465536等待1秒左右效果合适51048576等待10秒以上适合展示616777216等待数分钟不建议演示用不过我要提醒一个我踩过的坑项目一旦开始运行并挖出了一些块就不要随便改动difficulty。因为验证链的时候会用当前的difficulty去检查每一个区块的哈希前缀旧块已经按老难度挖出来了你把它调小旧块可能不满足新难度要求整条链会验证失败。教学版里难度的正确做法是初始化时定好之后不再动。3.3 校验整条链的完整性挖出来的链必须能自我证明“每一块都合法、每一环都连对”。is_chain_valid就是干这个的我建议无论如何都要把它写出来因为它是整条链安全性的第一道检查。def is_chain_valid(self, chainNone): chain chain or self.chain for i in range(1, len(chain)): curr chain[i] prev chain[i-1] if curr.previous_hash ! prev.hash: return False if curr.hash ! curr.compute_hash(): return False if not curr.hash.startswith(0 * self.difficulty): return False return True这里一共有三道检查缺一不可。第一道检查当前区块记录的previous_hash是否等于前一个区块的hash。这一步防止有人把第5块的前驱偷偷指向第3块从而绕过某个区块的篡改。第二道检查当前区块自己存的hash和它重新计算出来的hash是否一致。如果交易数据被改过重算的哈希不会等于旧值立刻败露。第三道检查当前区块的哈希是否仍然满足难度前缀要求。没有这一条就有人能够重新生成一个不满足PoW难度的哈希从而跳过计算成本伪造合法块。注意遍历从索引1开始不要从0开始。创世区块没有前一个区块它的previous_hash是默认值0让它也参与循环只会白白报错没有任何意义。4. 用Flask把链接出去API与简单共识4.1 核心接口只有一个原则把链和动作分开内存里的链写得再好别人也看不到、用不了。所以用Flask提供三个接口查看整条链、提交新交易、触发挖矿。这三个接口承担的角色完全不同查看和提交是给普通用户用的挖矿是给矿工用的把动作拆开后续做多节点时思路才清晰。from flask import Flask, jsonify, request app Flask(__name__) blockchain Blockchain() app.route(/chain, methods[GET]) def get_chain(): chain_data [block.__dict__ for block in blockchain.chain] return jsonify({length: len(chain_data), chain: chain_data}) app.route(/transactions/new, methods[POST]) def new_transaction(): data request.get_json() required [sender, recipient, amount] if not all(k in data for k in required): return Missing values, 400 blockchain.add_transaction(data[sender], data[recipient], data[amount]) response {message: Transaction added to pending list} return jsonify(response), 201 app.route(/mine, methods[GET]) def mine(): block blockchain.mine_block() if block is None: return jsonify({error: No pending transactions}), 400 return jsonify({message: New block mined, block: block.__dict__})/chain返回整条链/transactions/new接收JSON格式的交易数据并校验必需字段是否存在/mine触发挖矿并返回新生成的区块信息。如果你用curl测试可以这样curl http://127.0.0.1:5000/chain curl -X POST -H Content-Type: application/json -d {sender:alice,recipient:bob,amount:50} http://127.0.0.1:5000/transactions/new curl http://127.0.0.1:5000/mine这里有个细节容易被忽略request.get_json()在没有正确Content-Type时会返回None如果不做校验all(k in data for k in required)会抛TypeError。所以我在代码里先判断not all(...)但更稳妥的做法是再加一层if data is None的判断。代码越健壮后面联调越省心。4.2 多节点之间到底怎么“达成共识”当你有两个节点同时在运行它们各自挖矿产出的链很可能不一样。这时候就得有共识规则才能让网络收敛到同一条链上。教学版里最简单的共识就是“最长链规则”如果别的节点抛过来一条更长的链而且这条链验证通过那我方就用这条长链替换自己的链。这个逻辑我通常写成def resolve_conflicts(self, other_chain): if len(other_chain) len(self.chain) and self.is_chain_valid(other_chain): self.chain other_chain return True return False逻辑很简单但它反映了分布式系统的一个核心思想节点之间不一定同步但只要按统一规则替换最终会收敛到同一条链。别小看这几行代码你手上有两个节点的时候自己跑一遍再改掉中间一个区块的数据再让两个节点互相拉取感受会非常直观。4.3 为什么教学版故意不做完整的P2P网络有朋友拿到这个项目后会问怎么没有节点之间广播、怎么没有真正的P2P通信其实我是故意砍掉的。真实的P2P通信涉及节点发现、消息广播、心跳检测、数据同步等一堆问题这些问题每一个都够单独写一篇文章。教学版的目的是让你先把“链”和“挖矿”这两个核心骨架看懂后续再往里面加网络层比一开始就铺开容易得多。如果你想快速试验多节点完全可以在本机多开几个端口每个端口跑一个进程再用requests去互相拉取对方的/chain验证resolve_conflicts的逻辑。别把范围拉太大先让核心跑通再扩展外围。5. 实操中常见的坑以及我的排查经验5.1 哈希对不上十有八九是序列化问题我自己第一次在两个节点之间同步链时两边收到的区块数据一模一样但算出来的哈希始终不同。排查了很久最后发现就是json.dumps没有加sort_keysTrue。那是非常典型的哈希不一致问题同样的内容序列化顺序不同计算哈希的输入串就不同自然对不上。所以当你怀疑“明明数据一样但哈希不同”的时候第一步就是检查序列化规则是否一致。这里也再补一句之前提过的建议把timestamp从浮点数改成整数或ISO字符串能避开浮点精度带来的二次麻烦。5.2 挖矿太慢、接口卡死先查难度和线程我测试时把难度调到6结果一个块挖了将近4分钟。在这4分钟里整个Flask服务基本处于假死状态所有请求都在排队因为Python的while循环把GIL占满了单线程的Flask根本无暇处理新请求。这不是算法写错了而是难度的实际影响被低估了。第3.2节算过难度6意味着平均要试一千六百多万次哈希普通开发机跑起来确实吃力。解法有两个一是测试时把难度调回4如果你确实想演示一个比较耗时的挖矿过程就把挖矿扔到后台线程接口立刻返回“挖矿进行中”再用另一个接口去查询当前状态。真实矿池的架构思路也类似不可能让一个挖矿请求阻塞整个API。5.3 数据一重启就全部消失这个项目从头到尾都把链存在内存里所以一个很明显的问题是进程一关链就没了。很多初学者第一次发现重启后回到创世块状态会以为哪里写错了其实只是没有持久化。如果你想让数据跨重启保留最简单的办法是每次新区块产生后把整条链序列化成JSON写进一个文件启动时再加载进来。稍微进阶一点每个区块单独追加写入文件加载时逐条验证字符串读取后能重新构造出完整的Block对象。这一步做完你的链就不再只是内存里的玩具而是一个具备基本恢复能力的迷你区块链。5.4 接口直接返回内部字段会很危险不知道你发现没有我在Flask接口里为了省事直接用了block.__dict__让Python把对象里所有属性一股脑丢给前端。看起来方便但这是一个很不好的习惯。一旦你在Block类里增加了一些内部字段比如“挖矿耗时”“临时标记”“调试缓存”它们也会被接口暴露出去。多余字段不仅让接口变脏还可能泄露内部实现细节。更稳妥的做法是给Block写一个to_dict()方法明确指定哪些字段可以对外返回。这也是我在做正式项目时一定会改掉的地方建议你从教学版开始就养成约束接口返回结构的习惯。现象可能原因排查思路两个节点计算的哈希不一致序列化键顺序不同检查json.dumps有没有sort_keysTrue/mine接口长时间没响应难度太高导致挖矿阻塞降低难度或把挖矿放到后台线程重启后链消失链只存在内存里增加持久化存储逻辑接口返回了一堆奇怪的字段直接返回对象内部属性定义to_dict()控制返回字段时间戳偶发导致验证失败浮点数精度问题改用整数或ISO时间戳6. 最后说一点我自己的体会把这套代码完整跑通之后我再看区块链相关的文章理解速度明显不一样了。以前听到“挖矿要消耗算力”“最长链共识”这些词只是字面知道现在脑海里会自动浮现出那个while循环在那里拼命算哈希的画面会想起我自己等难度6挖矿等了四分钟的那种焦躁。这种理解是看多少文章都换不来的。如果你接着往下深挖方向其实很清晰把简单交易换成更贴近比特币模型的UTXO结构给交易加上数字签名引入Merkle树来高效验证区块里的交易再把真正的P2P节点通信补上。每往前一步都是独立的课题也都值得单独写一篇实战笔记。但无论走多远最初手写这个最小区块链带来的骨架感会一直留在那里帮你把所有新概念挂到正确的位置上。 用Python手写极简区块链：哈希、工作量证明与Flask实战
我早期接触区块链的时候总觉得这玩意儿被神化得太厉害了。后来花了几天时间用Python把一条极简区块链从头撸了一遍才彻底想明白所谓区块链核心不过就是三件事——把数据放进结构体、给结构体算一个独一无二的指纹、再把一串结构体按指纹顺序串起来。这篇文章就是我那次从零开始“拆玩具”的完整记录包含区块结构、SHA-256哈希、工作量证明、Flask接口以及我一路上踩过的坑。它适合刚学完Python基础、想看看“区块链到底怎么用代码实现”的读者你不需要懂密码学或分布式系统跟着把代码敲一遍概念自然就通了。1. 先想清楚一个区块里到底装什么1.1 把账本切成“块”区块这个词听起来玄乎你完全可以把它想成一页账本。手里有一笔接一笔的交易记录为了让别人查起来方便、又能防止事后篡改就把这些交易按时间打包成一页一页的“块”每一页就是一个区块Block。我在设计这个教学项目时给每个区块塞了六个字段index、timestamp、transactions、previous_hash、nonce以及最后算出来的当前区块哈希hash。为什么是这六个字段而不是更少因为每个字段回答的都是一个非常具体的问题。字段作用关键点index当前区块是第几个从0开始创世区块是0timestamp区块打包的时间用时间戳表示方便排序和校验transactions区块里实际存放的交易数据教学版用一个字典列表previous_hash前一个区块的哈希这是“链”能连起来的核心nonce挖矿时调整的随机数工作量证明的“解题答案”hash当前区块的数字指纹由上面所有字段计算得出你可以把这六个字段分成三组index和timestamp是区块的基本身份信息transactions是真正的业务数据previous_hash、nonce和hash则承担了“链式验证”的工作。前两个好理解后面三个得专门讲一讲尤其是哈希。1.2 哈希是整条链的“防篡改感应器”哈希hash本质上是一串固定长度的十六进制字符串它是通过一个叫SHA-256的算法算出来的。Python的hashlib库就内置了sha256一行代码就能调用。但理解哈希比会调用更重要因为整条链的安全性都建立在哈希的两个特性上。第一个特性是“雪崩效应”哪怕输入只改变一个字符算出来的整个哈希都会面目全非。第二个特性是“单向性”从哈希值几乎不可能反推出原始输入内容。这两个特性结合起来让哈希成了一个完美的“篡改感应器”。举个例子。假设某一笔交易的金额本来是100有人悄悄把它改成了101哪怕只改一个字符这个区块重新计算出的哈希就会变得完全不一样。因为下一个区块里记录的previous_hash还指向老的哈希两个哈希一对比马上就能发现这个区块被动过手脚整个链的验证也自然失败。我一开始也会问那如果有人把篡改后的区块以及它后面所有区块的previous_hash全都重新算一遍不就能蒙混过关了这就是工作量证明要解决的问题后面第3部分专门讲。现在你只需要先记住哈希保证了“数据改动能被立刻察觉”。2. Python实现从环境准备到区块类2.1 Python环境与依赖到底需要装些什么这个教学项目对Python版本要求很宽松Python 3.7以上基本都能跑。如果你还没配置环境直接去Python官网下载对应系统的安装包安装的时候记得勾选“Add Python to PATH”这样在终端里输入python --version就能看到版本号。常见的一类报错是python was not found多半是安装时没勾PATH或者电脑装了多个Python版本导致命令指向混乱把安装时的勾选补上就行。官方下载地址会因系统而异但通常搜索结果第一位就是官网。要是你用的是Linux系统可能自带了Python用python3 --version确认。我这里建议在项目目录里创建一个虚拟环境避免以后装其他库时互相冲突。执行下面两条命令就行python -m venv venv source venv/bin/activate # Windows下是 venv\Scripts\activate这个项目真正依赖的第三方库只有两个一个是Flask用来提供HTTP接口另一个是requests做多节点测试的时候会用到。requests可能在你测试共识时才需要但两者都推荐顺手装上后面不折腾。pip install flask requests用到的标准库有hashlib、json、time这些是Python自带的不需要额外安装。整个项目核心代码大约一百行左右非常适合逐行阅读。2.2 区块类的实现以及那些容易被忽略的细节区块我用一个简单的类来表示。每个区块在初始化的时候就计算出自己的哈希后续如果要挖矿会重算哈希。import hashlib import json import time class Block: def __init__(self, index, transactions, timestamp, previous_hash, nonce0): self.index index self.transactions transactions self.timestamp timestamp self.previous_hash previous_hash self.nonce nonce self.hash self.compute_hash() def compute_hash(self): block_string json.dumps({ index: self.index, transactions: self.transactions, timestamp: self.timestamp, previous_hash: self.previous_hash, nonce: self.nonce }, sort_keysTrue).encode() return hashlib.sha256(block_string).hexdigest()这里有个极其重要的细节就是json.dumps里的sort_keysTrue。我最初写这段代码时没有加它结果在本地跑一切正常一旦把链的数据从A节点传给B节点再算哈希两边怎么都对不上。原因是Python字典是保持插入顺序的同样的内容插入顺序不同序列化出来的字符串就不同哈希自然不同。加了这个参数就强制所有节点都按照相同的键顺序去序列化这样“同一份数据必然得到同一个哈希”。还有一处要注意json.dumps(...).encode()默认使用UTF-8编码字节串。这里必须转成字节串因为hashlib.sha256接收的是字节不是字符串。如果你输入的是字符串会直接抛TypeError这也是新手比较容易遇见的问题。至于timestamp为什么用time.time()返回的浮点数我只能说教学版图省事。浮点数在跨节点传输时存在精度问题你如果只是想本地跑通无所谓但一旦要做多节点实验建议换成整数时间戳否则在某些边界情况下哈希验证会莫名其妙失败。这一点在文章第5部分还会单独提。2.3 区块链类、交易池和创世区块区块有了链本身更好理解。区块链类维护三个核心状态一整条链的数据、一个待打包交易池以及当前挖矿难度。class Blockchain: def __init__(self): self.chain [] self.pending_transactions [] self.difficulty 4 self.create_genesis_block() def create_genesis_block(self): genesis Block(0, [], time.time(), 0) genesis.hash self.proof_of_work(genesis) self.chain.append(genesis) def add_transaction(self, sender, recipient, amount): self.pending_transactions.append({ sender: sender, recipient: recipient, amount: amount }) # proof_of_work 和 is_chain_valid 会在下一部分补充创世区块是区块链的起点它没有前一个区块可以指。所以它的previous_hash我用字符串0代替。这个块通常不走普通挖矿流程而是先创建再单独挖一次矿这样后面每个块才有一个合法的“祖先”。pending_transactions就是“待打包交易池”。在真实系统里用户发起交易后交易会先进入一个池子矿工批量打包。所以这里我刻意把“新增交易”和“挖矿”分成两个动作而不是一有交易立刻打包这个设计更贴近真实区块链的行为。3. 挖矿与工作量证明算力和随机数的平衡3.1 为什么“挖矿”只是在反复猜数字工作量证明Proof of WorkPoW是这个教学项目里看着最绕的部分其实逻辑极其简单。我们给新区块定一个目标它的哈希字符串必须以若干个0开头。0的个数就是难度。如果当前这个nonce算出来的哈希不满足就把nonce加1重新算一遍直到撞上一个满足条件的为止。def proof_of_work(self, block): 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注意这段代码里“改nonce、重算哈希”是唯一的办法。因为SHA-256是单向的你没法通过分析输入推测“什么值会产生以0000开头的哈希”只能一次一次试。nonce本身没有任何业务含义它就是一个为了让最终哈希满足条件的“抽奖号码”。挖矿消耗的算力本质上就是消耗在这无数次哈希计算里。有一个写得比较顺手的表达是while computed_hash[:self.difficulty] ! 0 * self.difficulty效果和startswith一样但startswith读起来更直白也不容易把切片边界写错。3.2 难度参数怎么定才合理哈希字符串的每一位都是十六进制字符每一位上有16种可能。难度为4要求前4位全部为0那么单次算出的哈希满足条件的概率是(1/16)^4 1/65536也就是说平均大约需要65536次哈希尝试才能挖出一个合法区块。如果你的机器每秒能算5万到10万次SHA-256那平均1秒左右能产出一个块。这个体验刚刚好能感受到“挖矿需要计算”又不用等太久。如果把难度调到5概率变成(1/16)^5约等于1/1048576也就是平均要试一百万次时间会拉长到10到20秒。调到6就是一千六百多万次可能要几分钟。所以教学演示项目把难度保持在4或5最舒服。太低了没有“挖矿感”太高了你会误以为程序卡死。难度平均尝试次数教学演示体验34096几乎秒出没有挖矿的感觉465536等待1秒左右效果合适51048576等待10秒以上适合展示616777216等待数分钟不建议演示用不过我要提醒一个我踩过的坑项目一旦开始运行并挖出了一些块就不要随便改动difficulty。因为验证链的时候会用当前的difficulty去检查每一个区块的哈希前缀旧块已经按老难度挖出来了你把它调小旧块可能不满足新难度要求整条链会验证失败。教学版里难度的正确做法是初始化时定好之后不再动。3.3 校验整条链的完整性挖出来的链必须能自我证明“每一块都合法、每一环都连对”。is_chain_valid就是干这个的我建议无论如何都要把它写出来因为它是整条链安全性的第一道检查。def is_chain_valid(self, chainNone): chain chain or self.chain for i in range(1, len(chain)): curr chain[i] prev chain[i-1] if curr.previous_hash ! prev.hash: return False if curr.hash ! curr.compute_hash(): return False if not curr.hash.startswith(0 * self.difficulty): return False return True这里一共有三道检查缺一不可。第一道检查当前区块记录的previous_hash是否等于前一个区块的hash。这一步防止有人把第5块的前驱偷偷指向第3块从而绕过某个区块的篡改。第二道检查当前区块自己存的hash和它重新计算出来的hash是否一致。如果交易数据被改过重算的哈希不会等于旧值立刻败露。第三道检查当前区块的哈希是否仍然满足难度前缀要求。没有这一条就有人能够重新生成一个不满足PoW难度的哈希从而跳过计算成本伪造合法块。注意遍历从索引1开始不要从0开始。创世区块没有前一个区块它的previous_hash是默认值0让它也参与循环只会白白报错没有任何意义。4. 用Flask把链接出去API与简单共识4.1 核心接口只有一个原则把链和动作分开内存里的链写得再好别人也看不到、用不了。所以用Flask提供三个接口查看整条链、提交新交易、触发挖矿。这三个接口承担的角色完全不同查看和提交是给普通用户用的挖矿是给矿工用的把动作拆开后续做多节点时思路才清晰。from flask import Flask, jsonify, request app Flask(__name__) blockchain Blockchain() app.route(/chain, methods[GET]) def get_chain(): chain_data [block.__dict__ for block in blockchain.chain] return jsonify({length: len(chain_data), chain: chain_data}) app.route(/transactions/new, methods[POST]) def new_transaction(): data request.get_json() required [sender, recipient, amount] if not all(k in data for k in required): return Missing values, 400 blockchain.add_transaction(data[sender], data[recipient], data[amount]) response {message: Transaction added to pending list} return jsonify(response), 201 app.route(/mine, methods[GET]) def mine(): block blockchain.mine_block() if block is None: return jsonify({error: No pending transactions}), 400 return jsonify({message: New block mined, block: block.__dict__})/chain返回整条链/transactions/new接收JSON格式的交易数据并校验必需字段是否存在/mine触发挖矿并返回新生成的区块信息。如果你用curl测试可以这样curl http://127.0.0.1:5000/chain curl -X POST -H Content-Type: application/json -d {sender:alice,recipient:bob,amount:50} http://127.0.0.1:5000/transactions/new curl http://127.0.0.1:5000/mine这里有个细节容易被忽略request.get_json()在没有正确Content-Type时会返回None如果不做校验all(k in data for k in required)会抛TypeError。所以我在代码里先判断not all(...)但更稳妥的做法是再加一层if data is None的判断。代码越健壮后面联调越省心。4.2 多节点之间到底怎么“达成共识”当你有两个节点同时在运行它们各自挖矿产出的链很可能不一样。这时候就得有共识规则才能让网络收敛到同一条链上。教学版里最简单的共识就是“最长链规则”如果别的节点抛过来一条更长的链而且这条链验证通过那我方就用这条长链替换自己的链。这个逻辑我通常写成def resolve_conflicts(self, other_chain): if len(other_chain) len(self.chain) and self.is_chain_valid(other_chain): self.chain other_chain return True return False逻辑很简单但它反映了分布式系统的一个核心思想节点之间不一定同步但只要按统一规则替换最终会收敛到同一条链。别小看这几行代码你手上有两个节点的时候自己跑一遍再改掉中间一个区块的数据再让两个节点互相拉取感受会非常直观。4.3 为什么教学版故意不做完整的P2P网络有朋友拿到这个项目后会问怎么没有节点之间广播、怎么没有真正的P2P通信其实我是故意砍掉的。真实的P2P通信涉及节点发现、消息广播、心跳检测、数据同步等一堆问题这些问题每一个都够单独写一篇文章。教学版的目的是让你先把“链”和“挖矿”这两个核心骨架看懂后续再往里面加网络层比一开始就铺开容易得多。如果你想快速试验多节点完全可以在本机多开几个端口每个端口跑一个进程再用requests去互相拉取对方的/chain验证resolve_conflicts的逻辑。别把范围拉太大先让核心跑通再扩展外围。5. 实操中常见的坑以及我的排查经验5.1 哈希对不上十有八九是序列化问题我自己第一次在两个节点之间同步链时两边收到的区块数据一模一样但算出来的哈希始终不同。排查了很久最后发现就是json.dumps没有加sort_keysTrue。那是非常典型的哈希不一致问题同样的内容序列化顺序不同计算哈希的输入串就不同自然对不上。所以当你怀疑“明明数据一样但哈希不同”的时候第一步就是检查序列化规则是否一致。这里也再补一句之前提过的建议把timestamp从浮点数改成整数或ISO字符串能避开浮点精度带来的二次麻烦。5.2 挖矿太慢、接口卡死先查难度和线程我测试时把难度调到6结果一个块挖了将近4分钟。在这4分钟里整个Flask服务基本处于假死状态所有请求都在排队因为Python的while循环把GIL占满了单线程的Flask根本无暇处理新请求。这不是算法写错了而是难度的实际影响被低估了。第3.2节算过难度6意味着平均要试一千六百多万次哈希普通开发机跑起来确实吃力。解法有两个一是测试时把难度调回4如果你确实想演示一个比较耗时的挖矿过程就把挖矿扔到后台线程接口立刻返回“挖矿进行中”再用另一个接口去查询当前状态。真实矿池的架构思路也类似不可能让一个挖矿请求阻塞整个API。5.3 数据一重启就全部消失这个项目从头到尾都把链存在内存里所以一个很明显的问题是进程一关链就没了。很多初学者第一次发现重启后回到创世块状态会以为哪里写错了其实只是没有持久化。如果你想让数据跨重启保留最简单的办法是每次新区块产生后把整条链序列化成JSON写进一个文件启动时再加载进来。稍微进阶一点每个区块单独追加写入文件加载时逐条验证字符串读取后能重新构造出完整的Block对象。这一步做完你的链就不再只是内存里的玩具而是一个具备基本恢复能力的迷你区块链。5.4 接口直接返回内部字段会很危险不知道你发现没有我在Flask接口里为了省事直接用了block.__dict__让Python把对象里所有属性一股脑丢给前端。看起来方便但这是一个很不好的习惯。一旦你在Block类里增加了一些内部字段比如“挖矿耗时”“临时标记”“调试缓存”它们也会被接口暴露出去。多余字段不仅让接口变脏还可能泄露内部实现细节。更稳妥的做法是给Block写一个to_dict()方法明确指定哪些字段可以对外返回。这也是我在做正式项目时一定会改掉的地方建议你从教学版开始就养成约束接口返回结构的习惯。现象可能原因排查思路两个节点计算的哈希不一致序列化键顺序不同检查json.dumps有没有sort_keysTrue/mine接口长时间没响应难度太高导致挖矿阻塞降低难度或把挖矿放到后台线程重启后链消失链只存在内存里增加持久化存储逻辑接口返回了一堆奇怪的字段直接返回对象内部属性定义to_dict()控制返回字段时间戳偶发导致验证失败浮点数精度问题改用整数或ISO时间戳6. 最后说一点我自己的体会把这套代码完整跑通之后我再看区块链相关的文章理解速度明显不一样了。以前听到“挖矿要消耗算力”“最长链共识”这些词只是字面知道现在脑海里会自动浮现出那个while循环在那里拼命算哈希的画面会想起我自己等难度6挖矿等了四分钟的那种焦躁。这种理解是看多少文章都换不来的。如果你接着往下深挖方向其实很清晰把简单交易换成更贴近比特币模型的UTXO结构给交易加上数字签名引入Merkle树来高效验证区块里的交易再把真正的P2P节点通信补上。每往前一步都是独立的课题也都值得单独写一篇实战笔记。但无论走多远最初手写这个最小区块链带来的骨架感会一直留在那里帮你把所有新概念挂到正确的位置上。