1. 为什么要从零写一个区块链框架用再多不如自己造一遍先说个我自己踩过的坑。早几年我刚开始接触区块链的时候第一反应就是去找现成的库比如某个流行的Python区块链框架装上之后跑个Demo五分钟就能生成一条链看起来成就感满满。但真到了要理解它为什么这么设计、怎么改造成自己业务需要的形态时我整个人是懵的——所有的细节都被框架封装掉了我连区块里的hash是怎么算出来的都答不上来。后来我下定决心用纯Python从零写一个最简单的区块链。不依赖任何区块链框架只用了Python自带的hashlib、json、time这几个标准库再加上一个Flask做HTTP接口。整个过程写下来不到三百行代码但就是这三百行让我把区块链到底是什么这件事彻底想通了。这篇文章就是把这三百行代码从头到尾拆给你看。适合谁两种人最合适一种是刚接触区块链、想搞清楚底层原理的初学者另一种是像我一样用惯了框架、想补一补基础的技术人。你不需要懂密码学甚至不需要多深的Python基础只要会写基本的类和函数就能跟着这篇文章把链跑起来。我会从最核心的区块数据结构讲起然后是工作量证明、链的校验、HTTP接口再到多节点模拟。每段代码我都解释了为什么这么写而不是简单丢给你一段能跑的玩意儿。毕竟写区块链这件事真正值钱的是理解它背后的设计逻辑不是那几行代码本身。2. 把区块拆开看一个区块里到底存了什么区块链这个东西名字起得很直白——它就是一串由哈希值首尾相连的区块组成的链。所以要写它第一步不是写链而是先把区块这个基本单元定义清楚。我见过很多初学者一上来就急着写共识、写网络层结果连区块里应该放哪些字段都没想明白。这里我建议你慢一点。一个区块的本质是什么它就是一本账本中的一页记录着某一时间段内发生的所有交易同时带着一个防篡改指纹——哈希值。2.1 Block类的字段设计为什么是这五个字段我最终采用的Block类长这样import hashlib import json from time 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() # 当前区块的哈希值逐个说index区块在链上的编号。创世区块第一个区块是0之后依次递增。这个字段主要方便查询和定位不是安全的关键。transactions这里我做了简化直接用一个列表存交易数据。在真实场景里比如比特币区块里会经过默克尔树Merkle Tree把大量交易压缩成一个根哈希这样既节省空间又能快速验证某笔交易是否存在。但作为教学Demo直接存原始交易列表就够了逻辑更清晰。timestamp区块生成的时间。注意时间戳在这里不只是记录一下什么时候挖出来的它会在哈希计算中参与所以改时间戳会改变哈希值这也是防篡改的一部分。previous_hash这是整条链的关键。每个区块都保存着前一个区块的哈希值就像每个人手里都攥着上一个朋友的微信号——任何人想改中间某个区块后面所有区块的previous_hash就对不上了。nonce工作量证明要用的随机数后面专门讲。hash当前区块的哈希值也就是这个区块的身份证号。2.2 compute_hash里的小心机sort_keysTrue的坑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()把区块的各个字段序列化成字符串然后丢进SHA-256算一遍得到64位的十六进制字符串。这里有个我在实际开发中踩过的坑json.dumps如果不加sort_keysTrue字段顺序在不同Python版本、不同环境下可能不一致导致同一个区块算出不同的哈希值。你可能会想字段顺序是我自己写的每次都一样啊为什么还要排序因为你现在是自己构造dict但如果将来从外部接收JSON数据键的顺序由发送方决定万一对方把index放在timestamp后面哈希值就变了。加了sort_keysTrue之后不管输入顺序如何序列化结果永远按字母序排列保证哈希计算的确定性。还有一个小细节为什么要把数据encode()成字节因为hashlib.sha256()接收的是字节流而不是字符串Python的字符串是Unicode必须先编码。这一步忘了我折腾了好久报错信息还不太直观。提示在真实项目中建议给Block类加一个__repr__或to_dict方法方便调试时打印区块内容。不过初期写Demo直接用block.__dict__就行了Python会给实例自动生成这个属性字典。3. 工作量证明让记账权变成一件有成本的事有了区块类下一个核心问题就是谁来生成区块凭什么你生成的区块大家认可在比特币这类公链里答案是工作量证明Proof of WorkPoW。它的核心思想其实特别简单你要想出下一个区块就必须先解一道无捷径、只能硬算的数学题。这道题的难度决定了生成区块的成本从而让恶意攻击者即使想篡改也要付出巨大的算力代价。3.1 目标找到让哈希以N个零开头的noncePoW在代码层面怎么实现我给整条链设定一个难度值difficulty代表生成的区块哈希值必须以多少个0开头。比如difficulty 4就意味着哈希值的前四位必须是0000。class Blockchain: def __init__(self): self.chain [] self.pending_transactions [] self.difficulty 4 self.create_genesis_block()以0开头这件事在密码学上有什么意义你可以把它理解成一个抽奖活动SHA-256的输出是256位的十六进制字符串每一位有16种可能0到f。如果要求第一位是0中奖概率是1/16要求前两位是0概率变成1/256要求前4位是0概率就是1/65536。题目的巧妙之处在于除了暴力枚举nonce没有更好的解法。你不能提前算出哪个nonce能让哈希以四个零开头只能一个一个试。这就是工作量的来源——试错本身就是消耗CPU和电力的过程。实现如下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逻辑很简单从nonce0开始每次加1重新计算整个区块的哈希直到哈希值的前difficulty位全是0。找到那个nonce之后整个区块的哈希就固定下来了。3.2 difficulty设为4的原因算得快但能感受到成本初学者最爱问difficulty到底设几合适这里有一个体验上的权衡。太低的难度比如1基本上每个nonce都有一半概率中奖十六进制一位为0的概率是1/16实际是1/16挖矿跟没挖一样体现不出成本的概念太高的难度比如6或7在你只有单机CPU的情况下可能要算几百万次甚至上亿次等一个区块等到怀疑人生。我实测过几组数据供你参考普通笔记本单核Python循环difficulty4平均几次到几百次就能算出来流畅difficulty5平均要试几万次大概几秒钟difficulty6平均要试上百万次可能要一两分钟所以教学Demo我建议用4。你能清晰感受到它确实在算但不会等太久同时还有余量去演示调整难度。提示千万不要在生产环境里用Python写PoW纯Python的循环效率太低了。真实链用的是C/C实现、专门优化过的挖矿程序Python在这里只适合学习和原型验证。3.3 创世区块没有previous_hash的第一个区块讲完PoW还有一个特殊区块要处理——创世区块也就是整条链的第0号区块。它特殊在哪它没有前一个区块previous_hash无处可指。我给它约定的值是字符串0这算是一种行业惯例。它的timestamp直接用当前时间index固定为0transactions可以为空列表def create_genesis_block(self): genesis_block Block(0, [], time(), 0) genesis_block.hash genesis_block.compute_hash() self.chain.append(genesis_block)注意创世区块不需要做PoW因为它是链的起点是大家约定俗成接受的。它一旦生成哈希值就被固定下来之后所有区块都会向它看齐。4. 连接成链校验逻辑和最长链优先的规则区块有了PoW有了现在要把它们串成一条链。这里的核心不是把新区块append进去这个动作而是在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())接收两个参数一个是待添加的区块对象另一个是它宣称找到的有效哈希值proof。校验分三步校验previous_hash新区块声明的前序哈希必须等于当前链上最后一个区块的哈希。这一步直接杜绝了凭空插入区块的可能。校验哈希难度这个proof必须以difficulty个0开头。校验哈希一致性用区块当前内容重新计算哈希必须等于它宣称的proof。这一步确保nonce、交易数据、时间戳没有被篡改过。三步全过才允许append进链。这套校验逻辑真的写出来之后你会明白一件事区块链之所以不可篡改不是靠什么神秘力量就是靠每次添加时做一次全量校验而哈希一旦变化后续所有校验都会失败。4.1 交易是怎么被打包的pending_transactions的缓冲作用细心的你可能已经发现了我的Blockchain类里还有一个pending_transactions列表。它的作用是充当交易缓冲区。现实中的区块链不是每产生一笔交易就立刻挖一个区块那样太浪费了通常是把一段时间内的交易攒起来统一打包进下一个区块。我实现的流程是def new_transaction(self, sender, recipient, amount): self.pending_transactions.append({ sender: sender, recipient: recipient, amount: amount })然后挖矿方法把整个pending_transactions作为新区块的交易数据挖完之后清空缓冲区def mine(self): if not self.pending_transactions: return False last_block self.last_block new_block Block( indexlast_block.index 1, transactionsself.pending_transactions, timestamptime(), previous_hashlast_block.hash ) proof self.proof_of_work(new_block) self.add_block(new_block, proof) self.pending_transactions [] return new_block4.2 为什么真实区块链要最长链优先防止双花的简单解释还有一个规则必须讲清楚当网络里出现两条分叉链时所有节点应该选择哪一条答案是最长的那条。这个规则叫最长链优先Longest Chain Rule。用生活化的方式解释假设有两个人都想花同一笔钱双花攻击交易A被打进链1交易B被打进链2两条链同时存在。最终全网只认其中一条——工作量累计最多的那一条。因为每条链的生成都要消耗算力最长链意味着投入了最多的计算资源最可信。在单机Demo里这个规则暂时派不上用场但我还是建议你理解它因为后面做多节点模拟时这是共识机制的基础。我在代码里用一个valid_chain方法来做整链校验顺便也为后面多个节点各自持有一条链的场景做好了准备def valid_chain(self, chain): last_block chain[0] current_index 1 while current_index len(chain): block chain[current_index] if block.previous_hash ! last_block.hash: return False if not self.is_valid_proof(block, block.hash): return False last_block block current_index 1 return True这个方法从第1个区块开始逐个向后校验确保整条链上每个区块都接得住前一个且每个区块的哈希都符合难度要求。这就是链的完整性校验。5. 用Flask把链跑起来本地节点的HTTP接口实战写到这里区块链的核心逻辑已经完整了但有个问题这些代码跑在内存里外人看不到也没法交互。为了让这条链真正能用起来我给它套了一层HTTP接口这样就能用浏览器或命令行工具去访问它、提交交易、触发挖矿。这里我用的是Flask它是Python生态里最轻量的Web框架。为什么选它不是因为它是区块链的官方标配纯粹是因为写接口代码最少、最容易读懂。你只需要pip install flask5.1 四个核心接口查询、交易、挖矿、校验我先定义了四个最基础的接口from flask import Flask, request, jsonify 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(): tx_data request.get_json() required_fields [sender, recipient, amount] for field in required_fields: if not tx_data.get(field): return 缺少字段: {}.format(field), 400 blockchain.new_transaction( tx_data[sender], tx_data[recipient], tx_data[amount] ) return 交易已接收, 201 app.route(/mine, methods[GET]) def mine_block(): block blockchain.mine() if block: response { message: 新区块已挖出, index: block.index, transactions: block.transactions, nonce: block.nonce, previous_hash: block.previous_hash, hash: block.hash } return jsonify(response), 200 return 没有待打包的交易, 400 app.route(/validate, methods[GET]) def validate_chain(): is_valid blockchain.valid_chain(blockchain.chain) return jsonify({valid: is_valid}), 200 if __name__ __main__: app.run(host127.0.0.1, port5000)GET /chain查看整条链的所有区块。这是最直观的调试接口一开就能看到区块结构。POST /transactions/new提交一笔新交易入参是JSON格式的sender、recipient、amount三个字段。GET /mine触发一次挖矿。把所有pending交易打包成新区块做完PoW后接上链返回新区块信息。GET /validate做一次全链校验返回是否合法。用curl就能完整走一遍流程# 1. 发起一笔交易 curl -X POST http://127.0.0.1:5000/transactions/new \ -H Content-Type: application/json \ -d {sender: Alice, recipient: Bob, amount: 10} # 2. 挖矿 curl http://127.0.0.1:5000/mine # 3. 查看整条链 curl http://127.0.0.1:5000/chain跑完之后你会在终端里看到挖矿时计算nonce的过程也会看到链上多出一个带哈希值的新区块。我第一次跑通这个流程的时候挺有成就感的——虽然它和真正的比特币网络差了十万八千里但提交交易-打包-挖矿-上链这条核心链路是完全通的。5.2 POST的字段校验为什么必须做而不能偷懒刚才代码里有一段字段校验for field in required_fields: if not tx_data.get(field): return 缺少字段: {}.format(field), 400有读者可能觉得这是小题大做反正自己写的客户端字段肯定会传对。但我在实际开发里真吃过亏有一次我改了前端传参的名字把recipient写成了receiver后端没有任何校验直接就把None当成收款方存进了交易里。结果呢这笔垃圾交易被打包进了区块然后挖矿成功、上链了。后面所有基于这笔交易的数据分析全被污染了。所以任何暴露在HTTP层的数据入口都必须做字段完整性和类型的校验。这在区块链场景里尤其重要因为区块一旦上链理论上就很难回滚了。哪怕这只是个Demo也要从一开始养成交易入站即校验的习惯。提示生产级的实现还需要校验金额是否为正数、地址格式是否合法、发送方余额是否充足。教学Demo省略了余额检查但你心里要清楚这是个大坑。6. 多节点与共识从单机账本到分布式网络的演进单机版的链跑通之后我一度觉得挺满足了但紧接着一个灵魂拷问来了你这能叫区块链吗区块链最大的特点应该是分布式记账你一台机器自己记账自己验证跟用一个普通数据库有什么区别问得对。所以我做了第二步在同一台机器上开多个端口模拟多个节点然后写一套最简化的共识协议让各个节点之间能够同步链。这一步做完你才算真正摸到区块链的分布式膝盖。6.1 节点注册维护一个已知节点列表我在Blockchain类里加了一个字段nodes用来记录网络中其他节点的地址def register_node(self, node_url): parsed_url node_url.rstrip(/) if parsed_url not in self.nodes: self.nodes.add(parsed_url)注册接口app.route(/nodes/register, methods[POST]) def register_node(): values request.get_json() nodes values.get(nodes) if nodes is None: return 错误请提供节点列表, 400 for node in nodes: blockchain.register_node(node) return jsonify({message: 节点已注册, total_nodes: list(blockchain.nodes)})6.2 最简单的共识算法五个节点跑起来的实测真正的区块链网络共识是一个很复杂的课题涉及网络传播、分叉处理、奖惩机制等。但作为演示我实现一个最朴素的共识算法思路就一句话当发现别人的链比我自己的长而且整条链是合法的时候就用别人的链替换我的链。def resolve_conflicts(self): neighbours self.nodes new_chain None max_length len(self.chain) for node in neighbours: try: response requests.get(fhttp://{node}/chain, timeout2) 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 except requests.exceptions.RequestException: continue if new_chain: self.chain new_chain return True return False这里的valid_chain接收的是从JSON反序列化出来的字典列表所以原方法里需要做一个适配把dict还原成Block对象。我建议你在Block类里加一个静态方法staticmethod def from_dict(block_dict): return Block( indexblock_dict[index], transactionsblock_dict[transactions], timestampblock_dict[timestamp], previous_hashblock_dict[previous_hash], nonceblock_dict[nonce] )注意requests这个库需要额外安装pip install requests然后加一个共识触发接口app.route(/nodes/resolve, methods[GET]) def consensus(): replaced blockchain.resolve_conflicts() if replaced: response {message: 链已被替换, new_chain: blockchain.chain} else: response {message: 链已是最新, chain: blockchain.chain} return jsonify(response), 200实测时我开了三个终端终端Apython blockchain.py --port 5000终端Bpython blockchain.py --port 5001终端Cpython blockchain.py --port 5002然后在5000节点上注册另外两个节点在5001上挖几个块回到5000上调用/nodes/resolve你会看到5000的链被替换成了5001的长链。那一刻我第一次直观感受到全网统一账本是怎么实现的——不是靠某台中心服务器下发而是靠每个节点主动去比长度、认最长合法链。6.3 为什么这里不用复杂的PBFT等算法教学场景的取舍说明有些读者可能听说过PBFT、Raft、PoS这些共识算法会问你为什么不实现它们原因很简单这篇文章的目标是理解区块链的骨架而PoW加最长链优先是理解比特币这类公链最必要的组合。PBFT适合联盟链场景但它引入的视图切换、预准备/准备/提交三阶段通信会把Demo的复杂度拉高一个量级反而不利于初学者抓住主线。我建议的学习路径是先把PoW这条线吃透再去看PoS权益证明理解它如何解决PoW的能源消耗问题最后如果工作需要再深入PBFT等联盟链共识。一步一步来比一口气啃完所有算法要高效得多。7. 测试、踩坑与性能实测开发过程中最值得记录的细节最后这部分我把自己在开发过程中遇到的几个教科书不会告诉你的细节整理出来。这些坑虽然小但每一个都差点让我怀疑人生。7.1 坑一JSON序列化时字典键顺序导致的哈希不一致这个坑我在前面提到过但值得再强调一次因为它太隐蔽了。现象是这样的我写完compute_hash之后本地测试一切正常。然后我写了一个把链导出为JSON再导入回内存的测试结果每次导入后再算哈希和导出前不一样。排查了半天才发现问题出在json.dumps的键顺序上。Python的dict在3.7之后虽然保证插入顺序但不同来源的dict插入顺序不一样。从JSON文件读回来的dict和直接构造的dict键顺序可能不同哈希值自然不同。解决方案就是一切序列化都加sort_keysTrue。同理我把Block转成dict时也用json.dumps(..., sort_keysTrue)来算哈希而不是直接对dict对象算保证输入完全确定。7.2 坑二时间戳的精度问题time()返回的是浮点数精度到微秒。这在单机Demo里没问题但如果你在多节点场景下序列化再反序列化浮点数的精度可能会丢失。比如1582345678.123456被反序列化后变成1582345678.123456看起来一样但二进制表示可能有细微差异导致哈希值对不上。我的处理办法是记录区块时直接把时间戳四舍五入到整数timestampint(time())或者用time.time_ns()获取纳秒级整数。功能上这个Demo不需要精确时间所以整数时间戳完全够用还完美避开了浮点精度问题。7.3 坑三requests请求超时导致的共识失败resolve_conflicts里如果不设置timeout默认会一直等到对方响应为止。一旦某个节点挂掉了整个共识流程就会卡住表现是调用了/nodes/resolve接口后浏览器一直转圈。我给每个节点请求加了timeout2并且用try/except捕获异常后continue跳过不可达的节点。这是分布式系统里非常重要的一个工程习惯永远假设对端可能不可用并且为这种假设写好降级逻辑。7.4 性能实测难度4时一次挖矿的真实耗时我把自己实测的数据贴出来方便你判断自己机器配置是否正常。测试环境是普通笔记本Python 3.10Intel i5处理器difficulty4难度平均尝试次数平均耗时3约40次约0.01秒4约180次约0.05秒5约9000次约2.5秒6约400000次约110秒同一个算法难度的微小增加会带来算力需求的指数级增长。这就是比特币设计每2016个区块调整一次难度的根本原因——保证全网平均出块时间稳定在10分钟左右不管总算力怎么涨。提示如果你运行difficulty6以上的测试建议先把difficulty调回4不然真能等到怀疑人生。想做性能测试的话单独跑一个脚本不要连着Web服务一起跑否则接口会一直阻塞。7.5 一个实用扩展给区块加个简单的Merkle根前面我说过教学Demo直接存交易列表但如果你想离真实区块链更近一步可以试试给区块加一个简单的Merkle根。思路不复杂把交易列表两两组合分别做哈希再对上一层的哈希两两组合直到最终得到一个根哈希。如果交易数量不是2的幂就把最后一个哈希复制一份补齐。实现如下def merkle_root(transactions): if not transactions: return hashlib.sha256(b).hexdigest() layer [hashlib.sha256(json.dumps(tx, sort_keysTrue).encode()).hexdigest() for tx in transactions] while len(layer) 1: if len(layer) % 2 1: layer.append(layer[-1]) layer [hashlib.sha256((layer[i] layer[i1]).encode()).hexdigest() for i in range(0, len(layer), 2)] return layer[0]然后在Block的compute_hash里把transactions替换成merkle_root(transactions)就能极大缩小区块哈希计算的数据量——不用对整个交易列表做序列化只要根哈希稳定交易内容的任何变动都会导致根哈希变化进而影响区块哈希。这就是默克尔树一叶动而全树知的精髓。8. 写在最后这个项目还能往哪里走开发这个Demo的过程我最大的收获其实不是会写了区块链的代码而是彻底理解了为什么区块链需要这些机制——哈希是防篡改的指纹PoW是让记账有成本的准入机制最长链优先是全网收敛到统一账本的规则节点注册和共识是去中心化网络的基本协作方式。这些概念单独拿出来都能写一篇长文但只有把它们串在一起写一遍代码你才会真正有那种哦原来是这么回事的通透感。如果你跑通了这套代码想继续深入我建议按这个顺序扩展给交易加上数字签名用cryptography库生成密钥对让每笔交易都能验证确实是发送方本人发出的把链持久化到SQLite或文件里实现节点重启后数据不丢失引入UTXO模型替代当前的账户余额模型更接近比特币的真实设计尝试把PoW换成PoS对比两者的出块效率和能源消耗最后再分享一个小建议不要急着去看那些几千行的完整区块链源码先把自己的300行Demo跑熟、改坏、再修好。我自己就是这样前前后后把这个Demo推倒重写了三次每一次都比上一次理解更深。代码会骗人但原理不会你亲手写出来的链才是真正属于你的链。