我最初接触区块链原理的时候,满脑子都是“分布式账本”“共识机制”“默克尔树”这些听起来很唬人的词,总觉得这玩意儿离普通开发者很远。直到有人用一句话点破:区块链本质上就是一个加了约束条件的链表,每个节点里的数据用哈希锁在一起,篡改任何一个历史区块,后面所有的哈希都对不上。那一刻我豁然开朗。后来我用 Python 花了小半天时间,从零开始写出了一条能在本地跑起来的链,并且用 HTTP 接口把“发交易、挖矿、查链”串成了一套完整的交互流程。这篇文章就把这套实现完整拆开,从区块的数据结构到工作量证明再到链的校验,一步步还原出来。你可以跟着代码敲一遍,适合已经会 Python 基础语法、想搞懂区块链底层逻辑的读者,哪怕之前没接触过任何区块链知识,照着跑也能跑通。
1. 动手前先想清楚:我们到底要做一个什么样的区块链
1.1 区块链本质上就是一个链表,别被概念吓住
链表大家肯定学过数据结构,每个节点存着数据和指向下一个节点的指针。区块链也差不多,只是指针变成了一种特殊的“哈希引用”:每个区块里不仅存自己的数据,还存着上一个区块的哈希值。这个设计带来一个至关重要的性质:如果任何人偷偷改了某个历史区块里的交易数据,它的哈希就变了,那么下一个区块里存储的上一个区块哈希就对不上,于是整条链从那个位置开始就“断裂”了。
我后来在给别人讲的时候喜欢打个比方:把区块链想象成一摞叠起来的积木,每块积木侧面都印着前一块积木顶面的花纹。你抽掉中间任何一块,或者偷偷换掉一块,后面所有积木侧面的花纹就全都对不上了。这就是“不可篡改”在数据结构层面的真实含义,不是什么玄学,就是一层一层哈希校验。
所以我们做这个项目的目标很明确:写一个带有区块哈希链式结构的数据模型,给它配上一个能生成新区块的“挖矿”流程,再通过 HTTP 接口让外部用户发起交易、查看整条链。这个过程中重点不是代码有多么复杂,而是要把链式结构的约束条件落到每一行实现里。
1.2 为什么选 Python:不是为了炫技,是效率最优解
用 Python 做这件事,说实话不是因为 Python 适合做高性能的区块链——真要上线跑生产系统,大概率会选 Go、Rust 这类编译型语言。但做技术拆解和原型验证,Python 的优势太明显了:
- 标准库里有
hashlib,SHA-256 哈希直接调用,不需要引入额外的加密库; - 数据结构用
dict和list自然表达,JSON 序列化有现成的json.dumps; - Flask 几行代码就能拉起 HTTP 服务,把区块链从“内存里的一堆对象”变成一个可以被外部访问的“系统”。
我觉得对想理解区块链原理的人来说,Python 的核心价值在于把复杂概念压缩到最少代码量。哈希是区块链的地基,但你在 Python 里写hashlib.sha256(...).hexdigest()就搞定了。如果换成 C++,光处理字符编码和字节数组的长度问题就够你喝一壶。所以这篇文章的定位不是“生产级实现”,而是“教学级还原”,让你用最短的路径踩一遍核心逻辑。
1.3 功能清单:先立一个小目标
在动手写代码前,我给自己列了一个最小化的功能清单。你读后面的代码时,也建议先在心里立下这个目标,不然容易被各种概念带偏:
- 区块包含索引、时间戳、交易记录、上一个区块哈希、随机数 nonce 和自己的哈希;
- 整条链从创世区块开始,后续每个区块都指向上一个区块;
- 有一个工作量证明机制,挖矿需要找到符合条件的哈希;
- 支持创建交易记录,交易会进入待处理池,挖矿时被打包进新区块;
- 通过 HTTP 接口暴露三个操作:查看全链、发起交易、触发挖矿;
- 能校验整条链的完整性,并且支持用最长合法链作为“共识结果”替换本地区块链。
2. 核心数据结构:区块类与链类的设计
2.1 Block 类的字段选择:为什么是这五个字段
区块是整个系统的数据容器。我开始写的时候想过很多字段组合,比如要不要加版本号、要不要加难度值,最后都砍掉了。教学项目里字段越少,逻辑越清楚。下面这个Block类是我最终定稿的版本:
import hashlib import json import time from uuid import uuid4 class Block: def __init__(self, index, timestamp, transactions, previous_hash, nonce=0): self.index = index self.timestamp = timestamp self.transactions = transactions self.previous_hash = previous_hash self.nonce = nonce每个字段都有存在的理由:
index是区块在链上的位置,从 0 开始递增。它的主要用途是方便定位和排序,也让校验逻辑能检测出“某个区块被删除”的情况;timestamp记录区块生成时间,我用的是time.time()返回的浮点数。时间戳的作用是为交易提供一个时间维度的锚点,这也是很多追溯类应用的基础;transactions是一个列表,里面装着当前区块打包的交易记录,每笔交易是一个包含sender、recipient、amount的字典;previous_hash是上一个区块的哈希,这是整条链能够“串起来”的关键字段;nonce是工作量证明里的随机数,后面挖矿部分会反复修改它来寻找满足条件的哈希。
hash字段我没有放在构造参数里,而是在计算出哈希后再动态赋给对象。原因是哈希必须由其他字段计算得出,把它放进__init__反而容易造成“哈希参与了自身计算”的逻辑混乱。
2.2 用 hashlib 计算区块哈希:序列化稳定是关键
有了字段,就要有计算哈希的方法。我在这里走了弯路,最初直接用str(self.__dict__)去生成哈希,结果因为字典里键值顺序不稳定,同样的区块内容在不同运行环境下算出来的哈希不一样,整条链的校验时好时坏。所以最终老老实实用 JSON 序列化,并且添加了sort_keys=True:
def compute_hash(self): block_string = json.dumps({ "index": self.index, "timestamp": self.timestamp, "transactions": self.transactions, "previous_hash": self.previous_hash, "nonce": self.nonce }, sort_keys=True).encode() return hashlib.sha256(block_string).hexdigest()这里有一个非常容易被忽略的细节:json.dumps默认的ensure_ascii=True会把中文转成\uXXXX形式的转义字符。这个默认行为在哈希计算里反而是个优点,因为不管在哪个平台运行,同一份数据都会被转成同一种字节序列。如果你强行设置ensure_ascii=False,反而可能因为某些环境的默认编码不同导致哈希不稳定。我实测下来,sort_keys=True的作用是保证字典里的键按字母序排列,ensure_ascii保持默认,对教学项目来说是最稳妥的组合。
2.3 从创世区块到链类:落地的骨架代码
有了一笔一笔哈希,就可以写Blockchain类了。这里的重点是创世区块的处理。创世区块是整条链的第一块,它没有前一个区块,所以previous_hash我用字符串"0"表示。这个约定在业界很常见,不算什么了不起的设计,但能避免一个“先有鸡还是先有蛋”的问题:第一个区块没法通过计算得到previous_hash。
class Blockchain: def __init__(self): self.chain = [] self.pending_transactions = [] self.difficulty = 4 self.create_genesis_block() def create_genesis_block(self): genesis_block = Block(0, time.time(), [], "0", 0) genesis_block.hash = genesis_block.compute_hash() self.chain.append(genesis_block) @property def last_block(self): return self.chain[-1]pending_transactions是待处理交易池。新发起的交易先不直接上链,而是攒在这个池子里,等“挖矿”的时候一次性打包进新的区块。这个设计跟真实系统里“交易先进入内存池,矿工打包出块”的思路是一致的,虽然我们的实现没有网络广播,但流程是完整的。
3. 工作量证明与挖矿:让新区块变得“来之不易”
3.1 什么是工作量证明:哈希值前面的零,就是那道算出来的题
真实区块链里有各种共识机制,工作量证明是其中最有名的一种,核心思路是:“你要证明你花了算力,那我出一道特别难解、但特别好验证的题给你。”在我们这个项目里,题目长这样:找到一个nonce,使得整个区块的 SHA-256 哈希值以difficulty个 0 开头。
difficulty = 4的意思是目标哈希需要以"0000"开头。这个条件看起来很轻巧,实际上每次尝试猜中的概率只有 1/16^4,也就是 1/65536。从概率意义上讲,平均需要算 6 万多次才能找到一个合法 nonce。在代码里,我让区块的 nonce 从 0 开始不断累加,每次对区块内容重新计算哈希,直到结果满足前缀条件为止:
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,但任何人拿到这个区块后,只需要算一次哈希,再检查前面是不是有足够多的 0,就能验证挖矿结果的正确性。
3.2 为什么哈希不能提前预测:雪崩效应的直观感受
你可能想问:能不能用数学直接反推出 nonce,而不是一个个试?答案是基本不行。SHA-256 是单向哈希函数,输入稍有变化,输出就会翻天覆地。这种性质叫雪崩效应。我刚开始写代码时对这个没什么体感,直到有一次在测试时给交易数据多加了一个空格,发现区块哈希完全变成了另一串字符,跟原哈希毫无关系。
这个性质对区块链的意义非常大。恶意节点如果想篡改一个历史区块,不仅要重新计算被改区块的哈希,还得重新计算它后面所有区块的 nonce,因为后面每个区块的previous_hash都变了。当链足够长的时候,这个计算量是指数级别上升的,攻击者在算力上几乎不可能追上来。虽然我们这个小项目只有几秒钟就能挖完一个块,但逻辑是真实系统缩小后的样本。
3.3 挖矿的完整流程:交易打包、奖励发放、区块上链
挖矿在代码里做的事情比想象中多。它不仅是一个“找 nonce”的纯计算过程,还要处理交易打包和奖励记录。我给Blockchain加了一个mine方法,挖矿时先给矿工记录一笔“出块奖励”,然后把待处理交易全部打进新块,最后调用proof_of_work挖出合格哈希:
def new_transaction(self, sender, recipient, amount): self.pending_transactions.append({ "sender": sender, "recipient": recipient, "amount": amount }) return self.last_block.index + 1 def mine(self, miner_address): self.new_transaction(sender="0", recipient=miner_address, amount=1) block = Block(len(self.chain), time.time(), self.pending_transactions, self.last_block.hash) block.hash = self.proof_of_work(block) self.chain.append(block) self.pending_transactions = [] return block为什么要在这个环节先插入一条奖励交易?因为矿工不能凭空给地址加余额,账本上的每一笔钱都必须来自某个地方。真实系统里奖励是通过“coinbase 交易”实现的,也就是出块者给自己记账,在代码层面跟普通交易一样,只是发送方通常用特殊字符串表示“系统”。我这里用"0"表示系统节点。
出块之后立刻清空pending_transactions,这样下一轮挖矿从空池开始。这个清空操作经常被忽略,但它很关键。如果不清空,同一个挖矿流程产生的区块会重复包含旧交易。我一开始忘了清空,导致区块之间交易记录大量重复,后来在跑接口测试时发现链的长度和交易数量对不上,排查了半天才找到根因。
3.4 加块时的校验:不能随便什么块都能往链上挂
mine方法挖出的区块直接 append 到链上,但这是一个受信任的内部流程。真正的区块链系统里,任意节点都可能向其他节点广播自己的区块,所以必须设置一道关卡:只有满足所有约束条件的区块才能入链。我封装了一个add_block方法,专门在外部节点同步区块时使用:
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) return True def is_valid_proof(self, block, block_hash): return (block_hash.startswith('0' * self.difficulty) and block_hash == block.compute_hash())这个方法的逻辑是:先检查新块的previous_hash是否跟当前链末端区块的哈希一致,再验证新块的哈希是否满足难度要求并且确实是当前内容的哈希。两个条件都满足,才允许上链。这个设计让链的扩展变得严谨,同时也为后面的“链校验”打下了基础。
4. 交易、链校验与节点共识:让区块链真正可信起来
4.1 交易记录的结构:最简单的三方记账
我们的交易记录就是个普通字典,三个键:sender发送方、recipient接收方、amount金额。没有余额检查,没有签名验证,这些复杂度全被砍掉了。但你别小看这个极简结构,它已经具备了“账本”的核心要素:谁给了谁多少钱。
在new_transaction方法中,我把新交易追加到待处理交易池,并返回这笔交易预计会被打包进第几个区块。这个返回值对接口层做提示很有用,用户发完交易后会收到“这笔交易将被添加到第 5 个区块”的响应,互动感强了不少。实际测试中我也发现,提前返回正确的区块索引需要依赖len(self.chain)而不是下意识加 1,否则容易边界出错。
4.2 链校验的 3 个关键点:哈希连续、索引递增、PoW 有效
比“加一块”更严格的是“校验整条链”。这个is_chain_valid方法我会在节点同步时反复调用。它从第 1 个区块开始遍历到链尾,对每个区块做三项检查:
def is_chain_valid(self, chain): for i in range(1, len(chain)): block = chain[i] if block.hash != block.compute_hash(): return False if block.previous_hash != chain[i - 1].hash: return False if not block.hash.startswith('0' * self.difficulty): return False return True这三项检查对应的信任模型分别是一个假设链条的合法性:区块自身没被篡改、区块之间的连接没断、区块能通过工作量证明。我把遍历起点设成 1,跳过创世区块,因为创世区块的previous_hash是人为约定的"0",不参与一致性校验。
坦白说,我做教学版的时候一开始只检查了前两项,没有校验工作量证明。当时觉得“自己挖矿自己校验,没必要这么较真”。直到我想模拟节点同步时,发现一个偷偷把难度调低生成的假区块竟然能通过校验。加了 PoW 校验之后,攻击者必须付出真实的计算代价才能制造合法区块,防御级别完全不同了。这也是最容易踩的坑之一。
4.3 节点注册与最长链共识:当两条链打架时听谁的
单节点的区块链没有什么“分布式”可言,但为了把共识机制这根弦搭起来,我加了节点注册和链替换逻辑。节点注册就是维护一个“邻居节点地址集合”,register_node把对方地址加进去。真正体现共识思想的是resolve_conflicts:在所有链条中,两条合法链如果长度不同,以最长链为准。
这里的思想非常朴素:“每个人都认可付出最多工作量”的那条链。虽然我们的本地项目里只有一个节点,但如果后续扩展成多节点,每个节点只需要互相交换链数据,并调用resolve_conflicts去替换本地的链,就能达成一个非常基础的一致性协议。
def register_node(self, address): parsed_url = urlparse(address) self.nodes.add(parsed_url.netloc) def resolve_conflicts(self, other_chains): longest_chain = self.chain for chain in other_chains: if len(chain) > len(longest_chain) and self.is_chain_valid(chain): longest_chain = chain if longest_chain != self.chain: self.chain = longest_chain return True return False这段代码在真正多节点环境里还需要处理对象反序列化的问题,但在教学项目的语境里,逻辑主线是完全清晰的:先验证,再比较长度,最后决定是否替换。
5. HTTP 接口设计:让区块链从内存对象变成可交互系统
5.1 Flask 路由设计:链查询、发交易、挖矿三个入口
代码写到这个程度,区块链还是一个只能从 Python 交互的“内存玩具”,必须给它套上一层 HTTP 外壳,才能算是“可以运行的系统”。我用 Flask 搭接口,整个服务只有三个路由:GET /chain查看全链,POST /transactions/new发起交易,GET /mine触发挖矿。每个路由都绑定一个全局的Blockchain实例和一个node_identifier作为矿工地址。
app = Flask(__name__) blockchain = Blockchain() node_identifier = str(uuid4()).replace('-', '') @app.route('/chain', methods=['GET']) def full_chain(): response = { 'chain': [block.__dict__ for block in blockchain.chain], 'length': len(blockchain.chain) } return jsonify(response), 200 @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), 201 @app.route('/mine', methods=['GET']) def mine(): block = blockchain.mine(node_identifier) response = { 'message': 'New Block Forged', 'index': block.index, 'transactions': block.transactions, 'nonce': block.nonce, 'previous_hash': block.previous_hash, 'hash': block.hash } return jsonify(response), 200我用的是block.__dict__这种方式把 Block 对象序列化成字典,因为Block类里没有写to_dict方法,__dict__是最快速的转换方式。如果字段多了,建议还是写一个显式的序列化方法更可控。另外注意,/mine我设置成 GET 请求,本意是方便在浏览器里直接触发。真实项目里挖矿是一个改变系统状态的操作,应该用 POST。
5.2 运行和测试:用 curl 走一遍完整流程
环境准备很简单。确保本机安装了 Python 3 和 Flask,如果没装 Flask,打开终端执行:
pip install flask然后把代码保存为blockchain_demo.py,运行:
python blockchain_demo.py服务默认跑在http://127.0.0.1:5000。我建议用另一个终端窗口来测试接口,全套流程可以这样走通:
# 1. 查看初始链状态 curl http://127.0.0.1:5000/chain # 2. 发起一笔交易 curl -X POST http://127.0.0.1:5000/transactions/new \ -H "Content-Type: application/json" \ -d '{"sender": "Alice", "recipient": "Bob", "amount": 10}' # 3. 挖矿,把交易打包进新区块 curl http://127.0.0.1:5000/mine # 4. 再查看链,会看到新生成的区块 curl http://127.0.0.1:5000/chain在测试过程中,我建议留意几个点:第一次查看链时只有创世区块;发起交易后链没有变化,因为交易还在待处理池里;调用挖矿后,新块里会包含那条交易和一个矿工奖励记录。如果事务记录没有出现在区块里,八成是你的mine方法里忘了把pending_transactions传给Block构造器。
5.3 完整代码整合:一个文件就能跑
为了方便大家复制保存,我把前面拆解的代码整合成一个完整文件。这个版本去掉了节点注册的演示路由,保留核心三接口和链校验能力,方便你直接运行。
import hashlib import json import time from uuid import uuid4 from flask import Flask, jsonify, request class Block: def __init__(self, index, timestamp, transactions, previous_hash, nonce=0): self.index = index self.timestamp = timestamp self.transactions = transactions self.previous_hash = previous_hash self.nonce = nonce self.hash = None def compute_hash(self): block_string = json.dumps({ "index": self.index, "timestamp": self.timestamp, "transactions": self.transactions, "previous_hash": self.previous_hash, "nonce": self.nonce }, sort_keys=True).encode() return hashlib.sha256(block_string).hexdigest() class Blockchain: def __init__(self): self.chain = [] self.pending_transactions = [] self.difficulty = 4 self.create_genesis_block() def create_genesis_block(self): genesis_block = Block(0, time.time(), [], "0", 0) genesis_block.hash = genesis_block.compute_hash() self.chain.append(genesis_block) @property def last_block(self): return self.chain[-1] def new_transaction(self, sender, recipient, amount): self.pending_transactions.append({ "sender": sender, "recipient": recipient, "amount": amount }) return self.last_block.index + 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 def mine(self, miner_address): self.new_transaction(sender="0", recipient=miner_address, amount=1) block = Block(len(self.chain), time.time(), self.pending_transactions, self.last_block.hash) block.hash = self.proof_of_work(block) self.chain.append(block) self.pending_transactions = [] return block def is_chain_valid(self, chain): for i in range(1, len(chain)): block = chain[i] if block.hash != block.compute_hash(): return False if block.previous_hash != chain[i - 1].hash: return False if not block.hash.startswith('0' * self.difficulty): return False return True app = Flask(__name__) blockchain = Blockchain() node_identifier = str(uuid4()).replace('-', '') @app.route('/chain', methods=['GET']) def full_chain(): response = { 'chain': [block.__dict__ for block in blockchain.chain], 'length': len(blockchain.chain) } return jsonify(response), 200 @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), 201 @app.route('/mine', methods=['GET']) def mine(): block = blockchain.mine(node_identifier) response = { 'message': 'New Block Forged', 'index': block.index, 'transactions': block.transactions, 'nonce': block.nonce, 'previous_hash': block.previous_hash, 'hash': block.hash } return jsonify(response), 200 if __name__ == '__main__': app.run(host='0.0.0.0', port=5000)代码量不长,但是一个完整可运行的最小区块链系统。你可以把它当成“玩具”,但它的哈希链、工作量证明、交易打包和链校验逻辑跟真实系统完全同构。
6. 常见问题与排查技巧实录:我踩过的那些坑
6.1 哈希对不上:JSON 序列化的两个隐藏陷阱
这是我在项目调试里遇到最多的一个问题,表现是:区块链里的区块已经用compute_hash算好了哈希并存入对象,但调用is_chain_valid时永远返回 False。究其原因,绝大多数逃不出两个原因。
第一个原因是字典的键顺序不稳定。Python 3.7 以后字典保持插入顺序,但这不代表每次构造出来的字典顺序都一样。我在代码里用了一个sort_keys=True来强制排序,就是为了避免同一个区块在构造和校验时因为键顺序不同而算出不同哈希。第二个原因是类型不统一。比如时间戳,我在new_transaction里存了一个time.time()浮点数,但如果在测试过程中为了演示手动改成了字符串,就会导致哈希失败。这个问题很难排查,因为报错不会告诉你是哪个字段类型变了。
排查这类问题我有个习惯:写一个单独的小脚本,手动构造一个区块对象,调用两次compute_hash,对比结果是否一致。如果不一致,基本就是序列化方式不稳定。多跑几轮,很快能定位到具体字段。
6.2 链校验一直失败:创世区块和难度前缀要留意
如果你运行完整代码后执行is_chain_valid,发现空链或只含创世区块的链都会返回 False,那大概率是校验逻辑写得不严谨。我在初版代码里犯过一个低级错误:难度前缀判断用了block.hash.startswith('0' * self.difficulty)没错,但忘了考虑创世区块的哈希可能不满足难度要求。
解决办法有两个:要么让创世区块也参与挖矿,保证它满足难度;要么在校验时直接从索引 1 开始遍历,跳过创世区块。我最终选择了后者,因为创世区块是在系统初始化时生成的,它只需要存在即可,不必刻意满足难度。这个选择也让逻辑更清晰:难度约束从第一个真正“被挖出来”的区块开始生效。
6.3 接口返回 400:POST 请求的 JSON 格式问题
使用 curl 测试时经常遇到 400 错误,响应内容是我在代码里写死的Missing values。绝大多数情况下,原因是request.get_json()返回了None,说明请求头里的Content-Type不是application/json,或者请求体的 JSON 格式不合法。
解决方法是确保 curl 命令中加了-H "Content-Type: application/json",并且-d后面的 JSON 字符串用单引号包裹,避免 shell 对特殊字符做转义。Windows 的 CMD 下用 curl 需要注意引号规则跟 Linux/macOS 不同,如果遇到解析问题,可以把 JSON 写进文件再用-d @data.json方式提交。
6.4 挖矿变慢:难度值别一味调高
测试时你可能想体验一下难度带来的“真实感”,把self.difficulty调到 5 或 6。这里要提醒你,难度是每增加 1,计算量增加 16 倍。难度 4 时平均要算 6 万多次,难度 6 时直接到 1600 多万次,普通笔记本电脑可能要跑几十秒甚至几分钟。
如果只是想验证 PoW 逻辑,我建议控制在 3 到 4。如果你确实要调高难度,可以在proof_of_work里加一点进度打印,至少让你知道程序还在跑,不然屏幕上会一片空白,容易误以为死机了。
6.5 给 Python 新手的三个小建议:环境先跑通再改逻辑
我看见不少人下载代码后直接卡在跑不起来这一步,问题大多出在 Python 环境,跟区块链本身没关系。第一,确认你装的是 Python 3,可以在终端执行python --version来验证;第二,装依赖库时用pip install flask,不要用python install flask,这个是命令名错误;第三,如果系统里同时存在 Python 2 和 Python 3,运行指令用python3而不是python,否则可能启动的是 Python 2 导致代码报语法错误。
还有一个很基础但值得反复强调的点:在项目目录里创建一个虚拟环境再装依赖,能避免很多 Python 包管理方面的糟心事。创建命令是python -m venv venv,Windows 系统激活是venv\Scripts\activate,macOS/Linux 是source venv/bin/activate。
7. 进阶方向与扩展思路
跑通这个最小系统之后,你可以往很多方向扩展。每做一个扩展,对区块链的理解都会上一个台阶。
最直接的方向是加数字签名。现在任何一个人都可以伪造一笔“Alice 转给 Bob”的交易,因为交易记录里只有字符串,没有验证发送方身份的机制。真实区块链用的是非对称加密:发送方用私钥签名,任何人都能用公钥验证交易确实来自发送方。用 Python 实现这个,可以看标准库ecdsa或者cryptography。
第二个方向是改共识机制。把工作量证明换成权益证明,体验一下“持有越多,话语权越大”的另一种信任模型。这个改造的重点在于如何定义权益、如何公平地选出出块者,逻辑编排比 PoW 复杂不少,因为没有“哈希前缀匹配”这样简洁的校验条件了。
第三个方向是搭一个真正的 P2P 网络。现在的实现只有一个 Flask 服务,谈不上去中心化。你可以试着用 WebSocket 或者简单的 HTTP 长轮询让多个节点互相广播新区块和新交易,然后再用resolve_conflicts实现最长的链共识。这个方向工程量大些,但做起多节点同步时特别有意思。
第四个方向可以做实际场景落地。比如你在热搜词里看到“区块链溯源系统”,其实就是在链上存“产品从生产到流通各个环节的溯源信息”。你可以给区块增加一个data_type字段,让交易记录扩展到商品流通记录,再加一个查询接口,用链上记录追踪一件商品的完整路径。这会让区块链的技术价值落到一个具体的业务层面。
我个人比较推荐先做数字签名,因为它能让账本真正“可信”。你试过之后会发现,代码量只增加了一两百行,但整个系统从“演示玩具”变成了“有账号体系的最小信用系统”,那种感受非常奇妙。
最后再分享一个这套代码后续扩展时可以继续使用的技巧:给Block类加一个to_dict方法,把哈希字段统一序列化,这样在区块数据结构变复杂后,接口层和校验层都不需要改动太多。我在自己做溯源扩展时就是靠着这个接口隔离,才避免了多处重复修 bug 的窘境。
如果你把这篇文章里的代码敲完、敲通,再试着自己改改字段、加加功能,我相信你对“链式结构”“哈希不可逆”“工作量证明”这些概念的理解,会超过读十篇文章。这也是我用 Python 重写一遍区块链后的最大体会:它没那么难,但也没那么简单;你亲手敲一遍,就全都懂了。