先亮观点我见过太多人把“自己动手写区块链”当成一个只配出现在简历里、或者毕业设计里的噱头。但如果你真的用Python把那条链从零搭出来你会突然看懂过去三年那些你似懂非懂的行业黑话——什么是不可篡改、什么是共识、为什么说区块链不适合高频交易。这篇文章我就把当初自己手写区块链的完整过程拆开给你看不绕弯子不堆概念直接上能跑通的代码再告诉你每一步为什么要这么写。在动手之前先把一个最容易被“区块链”三个字吓住的逻辑挑明区块链本质上就是一个特殊的链表。它特殊在什么地方一是每个节点的数据都由哈希“锁”着二是后一个节点的哈希值里包含了前一个节点的指纹三是任何节点想伪造数据都必须重新计算它之后所有节点的工作量证明。这三个特性全部加起来才构成了所谓“不可篡改”的假象。真相比听起来更朴素——它根本没打算防止你改它只是让你改不起。1.1 用记账本理解“链”的结构想象一本手工记账本每页纸记录一笔账下一页纸的右上角写着上一页纸右上角的密文编号。如果有人撕掉第3页重写了一页塞回去第4页上对不上的密文编号就会立刻出卖他。这个“密文编号”就是哈希页码之间的关联就是哈希指针。在真实区块链里这个“账本页”叫区块Block“密文编号”叫哈希Hash而“第N页的编号指向第N-1页”这件事在代码里就是每个区块结构体里有一个字段叫previous_hash。这个设计解决了什么问题它解决了顺序不可抵赖的问题。你去超市买瓶水再退回来账单上是“买水-退水”顺序清清楚楚但如果数据库被人改了你根本看不出顺序被动过。哈希链是专门用来让“顺序和内容”同时被锁死的结构。先有这个底层认知后面写代码才能看明白每一行在干什么。1.2 为什么不直接用一个数据库就完事很多人会问我用户表、交易表一台MySQL不香吗为什么非要用这种看起来又慢又笨的方案好问题。中心化数据库的前提是“所有人都信这一个库”。但现实场景里参与方彼此不信任谁也不想把自己的数据交给别人保管。区块链的做法是把账本复制给每一个参与节点大家各自存一份谁也别想独占修改权。想改数据可以你必须让参与共识的大部分节点同时承认你改后的版本这就是“共识机制”要干的事。这篇实战里我不会直接上P2P网络那会让文章失控。但我会把链、工作量证明、验证逻辑、REST接口一个个写出来。当你能独立跑通这篇文章的代码再去看任何一条公链的技术文档你都有一根非常清晰的底梁。2. 从区块开始哈希、时间戳和那个“神秘”的previous_hash2.1 用Python定义一个区块我先写一个最精简的区块类它足够让你看清一个区块内部到底长什么样import hashlib import json import time from typing import Any, Dict, List, Optional class Block: def __init__( self, index: int, transactions: List[Dict[str, Any]], timestamp: float, previous_hash: str, nonce: int 0, ): 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) - str: 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()这段代码里最值得留意的是compute_hash方法。它把一个区块里所有关键字段拼成一个JSON字符串再对这个字符串做SHA-256运算。为什么这里的字段顺序要用sort_keysTrue因为JSON字典默认的键顺序可能因为插入顺序不同而不同同一个语义的区块在别的语言或别的Python环境里算出来的哈希可能不一致。排序键之后只要字段内容一样在任何机器上算出来的指纹都完全一样。transactions我刻意设计成列表因为一个区块里可以塞一笔交易也可以塞一百笔。现在它只是一个装着字典的列表等讲完接口部分你会看到它被真正塞进自定义的数据。2.2 创世块所有链的起点整条链的第一块不能由previous_hash指向任何东西因为没有更早的块。通常的做法是给它一个全零的哈希占位符class Blockchain: def __init__(self): self.chain: List[Block] [] self.pending_transactions: List[Dict[str, Any]] [] self.difficulty 4 self.create_genesis_block() def create_genesis_block(self) - None: genesis_block Block( index0, transactions[], timestamptime.time(), previous_hash0 * 64, ) self.chain.append(genesis_block)创世块是唯一没有真实交易内容的块。因为大家用的代码一样创世块的哈希理论上全世界都一样。后来很多区块链网络的创世块都被硬编码进客户端代码里目的就是让所有节点从这个共同的起点开始对账。2.3 为什么区块里必须放时间戳时间戳在代码里看似可有可无——它不参与链的验证。但真实场景里时间戳的作用非常大它确定了区块的产生顺序。如果每个区块都记录自己的生成时间那么整条链就是一条按时间排序的账本。没有时间戳的话你无法回答“这笔交易到底发生在哪一天”这种基本业务问题。时间戳另一个作用是在后续同步时作为冲突评判维度之一。两个节点同时挖出了同一个高度的区块到底选哪一个时间戳靠前的区块通常会被优先接受。所以虽然验证时不一定直接用时间戳但设计区块结构时绝不能省它。3. 工作量证明给“记账”这件事加一道门槛3.1 什么叫“证明你干了活”如果任何人都能随手把区块加到链上那这个分布式账本早被垃圾数据塞爆了。中本聪在比特币里引入的工作量证明Proof of WorkPOW简单说就是你想在链上写下一笔账必须先解一道有一定难度的数学题。这道题长什么样回到我们的区块结构它要求的很简单——算出来的区块哈希前N位必须是0。N越大难度越高。我的difficulty初始值设成4也就是要求哈希以“0000”开头。这就意味着如果你想让一个区块被链接受光把数据填进去不够还得不断调整区块里的一个“随机变量”nonce让整个区块重新哈希直到哈希的前面四位刚好全部是0。3.2 用Python实现挖矿函数挖矿逻辑在代码里非常直白就是一个“暴力试错”的循环def proof_of_work(self, block: Block) - str: 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: Block, proof: str) - bool: 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, block_hash: str) - bool: return block_hash.startswith(0 * self.difficulty) and block_hash block.compute_hash()proof_of_work这个名字听起来神乎其神其实干的事就是nonce从0开始每次加1重新算哈希直到结果满足要求为止。所以你挖矿的本质就是高强度使用CPU跑SHA-256。比特币把难度调到很大让全网平均要计算几万亿亿次才能找到一个有效哈希这才有了专业矿机和矿场的出现。在这个简单实现里难度4很可能在一眨眼间就解出来了。但如果把难度调成6、7你会在终端里看到明显的CPU风扇起飞。这就是“工作量”的含义——付出的算力是可以被验证的。3.3 难度设置背后的权衡逻辑难度不是拍脑袋定的。它决定了整个网络出块的速度。如果平均1秒就出一个块节点间同步的时间可能就超过出块间隔会产生大量分叉如果1小时出一个块交易确认时间又太慢。比特币的目标是10分钟一个块所以每2016个块会调整一次难度让全网出块时间稳定在目标值附近。在我们的单机Demo里纯粹出于演示效果难度4刚刚好——既不让你等到心烦又能让你看到nonce真的在跳。后来我把难度调到6试了一次5秒钟才出一个块已经能感受到“我在花钱”的感觉了。等你自己跑的时候可以按这个思路去调难度体会一下参数与实际耗时之间的关系。4. 让这条链变得可用验证、持久化与REST接口4.1 链的完整性检查每一环都不能松一个只允许“追加”的区块链如果不加校验和普通数组没有任何区别——我可以随时改掉链上某个区块的数据让整条链信息错乱。所以关键的一步是写一个验证函数检查每一个区块的哈希是否连续def is_chain_valid(self, chain: Optional[List[Block]] None) - bool: chain chain if chain is not None else self.chain for i in range(1, len(chain)): current_block chain[i] previous_block chain[i - 1] if current_block.hash ! current_block.compute_hash(): print(f区块 {current_block.index} 的哈希被篡改) return False if current_block.previous_hash ! previous_block.hash: print(f区块 {current_block.index} 与前一区块的哈希链断裂) return False if not current_block.hash.startswith(0 * self.difficulty): print(f区块 {current_block.index} 不满足工作量证明要求) return False return True这个函数有三道检查当前区块内部的数据是否还能算出它自己记录的哈希。如果有人偷偷改了transactions里的字段这个判断立刻失败。当前区块的previous_hash是否等于前一个区块的hash。这是链的连续性检查。当前区块的哈希是否仍然满足困难度要求。防止有人绕开proof_of_work直接把一个假哈希填进去。三道检查合在一起才叫“验证一条链”。实际区块链节点在同步区块时每收到一个新区块都会跑一遍类似这样的验证验证不过就直接拒绝。这相当于给整个网络加了一道免疫系统。4.2 交易缓冲池与“挖出一个块”的完整流程现实项目不会每来一笔交易就出块而是把一堆待确认交易放进一个池子里等矿工攒够一批再打包成一个块。我也跟着这个思路来设计。def new_transaction(self, sender: str, receiver: str, amount: float) - int: self.pending_transactions.append( { sender: sender, receiver: receiver, amount: amount, } ) return self.last_block.index 1 def mine(self, miner_address: str) - Block: self.pending_transactions.append( { sender: 系统, receiver: miner_address, amount: 1.0, # 挖矿奖励 } ) new_block Block( indexself.last_block.index 1, transactionsself.pending_transactions, timestamptime.time(), previous_hashself.last_block.hash, ) proof self.proof_of_work(new_block) self.add_block(new_block, proof) self.pending_transactions [] return new_block这里有一个业界通行的做法也是很多人读源码时容易漏掉的一点挖矿时系统会先在待确认交易里塞一笔“铸币交易”——比如这里的sender是“系统”receiver是矿工地址金额是1.0。这叫“矿工奖励”是加密货币增发的唯一途径。如果没有这一步矿工凭什么心甘情愿消耗CPU帮你记账而mine函数的流程也就是一个矿工节点的完整动作把奖励交易加进池子把池子里所有待确认交易打包成新区块跑工作量证明找到合法的nonce把区块挂到链尾最后清空交易池。4.3 给链加一个持久化层别再让数据“一重启就没了”写完链逻辑之后我一开始很无所谓——反正演示嘛数据在内存里就行。然后我发现一个非常尴尬的问题每次重新运行脚本辛辛苦苦挖出来的链全没了。真实区块链节点是把区块数据落盘的要么存在LevelDB/RocksDB里要么直接存JSON文件。这个简单项目里我用一个本地JSON文件做持久化逻辑足够清晰也不引入额外依赖。import json import os CHAIN_FILE blockchain.json def save_chain(chain: List[Block]) - None: with open(CHAIN_FILE, w, encodingutf-8) as f: json.dump([block.__dict__ for block in chain], f, ensure_asciiFalse, indent2) def load_chain() - List[Block]: if not os.path.exists(CHAIN_FILE): return [] with open(CHAIN_FILE, r, encodingutf-8) as f: data json.load(f) chain [] for item in data: block Block( indexitem[index], transactionsitem[transactions], timestampitem[timestamp], previous_hashitem[previous_hash], nonceitem[nonce], ) block.hash item[hash] chain.append(block) return chain这块代码的核心是把区块对象的__dict__转成字典存下来读取时再反向构造。这里有个小坑直接json.dump(chain)会报错因为Block对象本身不是JSON可序列化类型。所以我用[block.__dict__ for block in chain]把每个区块转成字典这一步被我踩过几次之后现在写任何序列化逻辑都会先问一句“这个对象能不能一步到位转成基础类型”把持久化和链操作合在一起就是完整可复现的本地区块链模型。让我把区块链类补全先把所有核心逻辑整理成一个可直接运行的模块class Blockchain: def __init__(self): self.chain: List[Block] [] self.pending_transactions: List[Dict[str, Any]] [] self.difficulty 4 self.create_genesis_block() def create_genesis_block(self) - None: genesis_block Block( index0, transactions[], timestamptime.time(), previous_hash0 * 64, ) self.chain.append(genesis_block) property def last_block(self) - Block: return self.chain[-1] def proof_of_work(self, block: Block) - str: 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: Block, proof: str) - bool: 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, block_hash: str) - bool: return block_hash.startswith(0 * self.difficulty) and block_hash block.compute_hash() def new_transaction(self, sender: str, receiver: str, amount: float) - int: self.pending_transactions.append( { sender: sender, receiver: receiver, amount: amount, } ) return self.last_block.index 1 def mine(self, miner_address: str) - Block: self.pending_transactions.append( { sender: 系统, receiver: miner_address, amount: 1.0, } ) new_block Block( indexself.last_block.index 1, transactionsself.pending_transactions, timestamptime.time(), previous_hashself.last_block.hash, ) proof self.proof_of_work(new_block) self.add_block(new_block, proof) self.pending_transactions [] return new_block def is_chain_valid(self, chain: Optional[List[Block]] None) - bool: chain chain if chain is not None else self.chain for i in range(1, len(chain)): current_block chain[i] previous_block chain[i - 1] if current_block.hash ! current_block.compute_hash(): print(f区块 {current_block.index} 的哈希被篡改) return False if current_block.previous_hash ! previous_block.hash: print(f区块 {current_block.index} 与前一区块的哈希链断裂) return False if not current_block.hash.startswith(0 * self.difficulty): print(f区块 {current_block.index} 不满足工作量证明要求) return False return True这一整个类就是一个功能完整的“简单区块链”核心了。从结构上看它有链、有交易池、有挖矿、有验证。剩下的工作是把它暴露给外部使用者。4.4 用Flask给区块链加一个REST接口内存里的区块链再完美别人也看不到摸不着。为了让读者能直观操作它我加了一个轻量级的Flask服务。这里不需要复杂的权限系统只需要几个最基本的接口from flask import Flask, jsonify, request app Flask(__name__) blockchain Blockchain() app.route(/chain, methods[GET]) def get_chain(): return jsonify( { chain: [block.__dict__ for block in blockchain.chain], length: len(blockchain.chain), } ) app.route(/transactions/new, methods[POST]) def create_transaction(): data request.get_json() required (sender, receiver, amount) if not all(k in data for k in required): return 参数缺失, 400 index blockchain.new_transaction(data[sender], data[receiver], data[amount]) return jsonify({message: f交易已加入区块 {index} 的待确认池}), 201 app.route(/mine, methods[GET]) def mine_block(): miner request.args.get(miner, miner-001) block blockchain.mine(miner) return jsonify( { message: 新区块挖矿成功, index: block.index, transactions: block.transactions, previous_hash: block.previous_hash, nonce: block.nonce, hash: block.hash, } ) app.route(/validate, methods[GET]) def validate_chain(): valid blockchain.is_chain_valid() return jsonify({valid: valid}) if __name__ __main__: app.run(host0.0.0.0, port5000)运行方法很简单pip install flask python blockchain.py然后你可以在另一个终端用curl或者浏览器调接口。我建议你按下面这个顺序体验一遍# 查看当前链 curl http://127.0.0.1:5000/chain # 添加一笔交易 curl -X POST http://127.0.0.1:5000/transactions/new \ -H Content-Type: application/json \ -d {sender: alice, receiver: bob, amount: 10} # 挖矿把交易打包进新区块 curl http://127.0.0.1:5000/mine?mineralice # 再查一次链看到新区块 curl http://127.0.0.1:5000/chain # 验证链是否完整 curl http://127.0.0.1:5000/validate这一步做完你的区块链就不再是只能看不能碰的理论模型了它已经有了一个可以被外部程序调用的API层。这其实就是很多区块链项目里“节点”对外服务的基本形态一个HTTP接口外面接钱包、浏览器等各类工具。4.5 跑起来测试时我把链“改坏”了一次我写完整链路之后干了件“坏事”直接改了blockchain.chain[1].transactions[0][amount]把转账金额从10改成了1000。然后再调用/validate终端立刻打出了“区块1的哈希被篡改”整条链验证失败。这个测试虽然简单但非常有助于建立直觉在区块链里篡改任何一个历史区块的数据代价不是改一个字段而是从这个被篡改的区块开始后面所有区块的哈希全都要重新算一遍。如果后面还压着几百上千个区块每个区块都要重新完成工作量证明算力成本就会呈指数级上升。这也解释了为什么“区块链不可篡改”其实是不准确的表述。准确的说法是篡改的成本高到不划算。当某个账本的数据对某人来说足够值钱高到足以覆盖重算所有后续区块的算力成本时链照样可能被改写。所谓“安全”是账本价值与篡改成本之间的博弈而不是某种魔法。5. 现实世界里的坑并发、时间戳与边界情况5.1 并发挖矿与交易池竞态这个单机版区块链最容易被忽略的问题就是它没有考虑并发。如果直接用Flask的默认开发服务器多个请求同时到达时Python的全局解释器锁会让它们排队执行看起来“好像也没事”。但一旦你换成了多线程或者多进程的服务比如Gunicorn带着多个worker运行就会遇到一个非常尴尬的竞态条件两个请求同时调用/mine都从pending_transactions里读数据、都打包了一模一样的新区块、先后往self.chain后面追加——第二条链很可能因为previous_hash对不上直接被add_block拒绝。解决思路也很简单加一个线程锁import threading mining_lock threading.Lock() app.route(/mine, methods[GET]) def mine_block(): with mining_lock: miner request.args.get(miner, miner-001) block blockchain.mine(miner) # 构造响应的代码真实区块链项目里这种并发问题会被共识算法、节点间通信、分叉处理机制包裹住复杂很多。但底层的思维是一样的任何时候对共享状态进行“读-改-写”操作都要保证原子性。先看清这一点写再多代码都不会歪。5.2 时间戳不是万能的但绝不能省你可能已经注意到我在验证函数里并没有专门检查时间戳是否单调递增。这意味着恶意节点可以故意把区块时间戳改得很早或很晚而链验证依然通过。这在真实网络中是个非常经典的问题——时间戳攻击。解决办法是让节点比对本地时间拒绝那些时间戳偏差超过一定范围的区块。比如比特币客户端只接受与本地时间相差不超过2小时的区块头。但在单机Demo里我不打算实现这种比较策略因为引入系统时间判断会增加逻辑复杂度而且当你在不同电脑上运行这些代码时由于时钟差异验证可能会无辜失败。实际写代码时我的取舍是时间戳负责“业务语义”和“排序参考”而链的完整性主要由哈希链来保证。分清楚每个字段的责任边界比试图用一个字段解决所有问题重要得多。5.3 难度与出块时间一条体验上的“舒适区”我最初把difficulty设成2结果每次挖矿快到肉眼根本来不及观察。这不叫“体验区块链”这叫“过家家”。后来调到4才真正有了“算一下”的感觉。再调到6终端里开始能看到CPU风扇的声音拔起来了。不同难度下单次挖矿的大概耗时大致如下同一台机器仅供参考难度前导零数量平均尝试次数实测耗时22约256次毫秒级33约4096次毫秒级44约65536次约0.2秒55约104万次约2秒66约1680万次约15秒这张表同时也解释了为什么比特币全网算力那么恐怖——它的目标哈希前导零远不止6位而是接近70位所以需要全网矿工每秒进行天文数字级别的哈希运算。理解这张表你就理解了“POW是用电力换共识”这句行业老话的真正含义。5.4 一个小选项要不要把交易也做签名在写这个区块链的最终版本前我考虑过要不要给交易加上数字签名。加签名意味着每一笔交易都要验证sender是否真的有对应的私钥来发起这笔转账。这样做非常接近真实区块链了但文章主题是“简单”所以我做了取舍消息和交易结构保持简单把签名验证作为你自己动手的扩展题。如果你想继续深入推荐你从ecdsa或cryptography库入手为sender生成一对密钥把交易里的signature字段加上然后在new_transaction时验证签名。这是我从“能跑的区块链”到“可信的区块链”之间遇到的最重要的一道坎。6. 把这套代码拿去实战前的几点提醒6.1 这是教学玩具不是生产系统说实话这个链最多算一个概念验证原型它缺少真实区块链里最关键的几个模块P2P网络发现、节点间共识协议、交易内存池的优先级策略、账户余额的UTXO或账户模型、数据库索引、私钥管理。如果你拿它去接真实业务会被攻击者在五分钟之内击穿。但看一件事的价值不能只看“它产线能不能用”。它最大的价值是让你亲手拆开区块链的黑盒亲眼看到它是怎么一步步把“信任”变成“可验证的成本”。有了这个底子再去啃比特币白皮书或者看以太坊源码你会发现里面的概念你全都认识只是复杂度被真实场景放大了。6.2 我踩过的那些低级但真实的坑第一最容易踩的坑是忘记在哈希计算里包含nonce。我一开写代码时觉得nonce只是挖矿用的临时变量不参与计算结果发现不管nonce怎么变哈希都不变永远找不到合法值。后来才意识到nonce正是工作量证明里唯一可变的“试错参数”如果它不参与哈希挖矿循环就是死循环。第二持久化时忘记保存nonce。这个更隐蔽。因为重新加载区块时如果你只保存了哈希和字段但没有保存nonce而compute_hash又依赖nonce那么即使你什么都没改加载出来的区块哈希也和新算的对不上。所以我在save_chain里特别保留了nonce字段这算是序列化设计里的一个小教训。第三调高难度后没有耐心。测试难度6时我在终端里盯着看了十几秒还以为代码卡死了。程序并没有卡只是挖矿真的需要时间。如果你看到这里也想调难度试试建议把难度控制在5以内否则等待的时间足够你去泡杯咖啡。6.3 下一步可以往哪里走当你完整跑通这套逻辑并看懂它之后我建议你按下面这几条线去扩展每一条都会逼着你学习真实区块链的一个侧面加共识与P2P用socket或WebSocket让多个节点互相同步链处理冲突时采纳最长链。加交易签名为每笔交易增加签名字段并验证签名合法性。加UTXO模型放弃简单的“sender-receiver-amount”改成类似比特币的UTXO模型搞清楚什么是“未花费交易输出”。加默克尔树把区块里的交易列表压缩成一颗默克尔树根加深对区块头结构的理解。其中最推荐先做P2P同步因为当你真正手工同步两三条链并且看着它们因为最长链规则自动收敛时你对“去中心化系统如何达成共识”的理解会跟看再多文章都不一样。最后再分享一个我的实操体会如果你也正在学区块链别一头扎进共识算法的数学证明里。先把一条最简单的链用代码从零跑通。等你能亲手伪造一笔交易又亲手验证出它被篡改你才真正摸到了区块链的门把手。跨过这道门槛后面全是顺水推舟。