☰
Python从零实现区块链Demo:哈希、PoW与Flask接口全解析
2026/10/13 18:58:08 网站建设 项目流程

想彻底搞懂区块链,最直接的办法不是读十篇文章,而是自己动手写一个。我这次用Python从零搭了一个最小可运行的区块链Demo,没有依赖任何现成的区块链框架,只在最外层套了一个Flask用来做HTTP接口,去掉注释也就两百行左右。哪怕你之前只写过几百行Python脚本,照着这段思路敲一遍,也能把一条完整的链跑起来。这个Demo能干什么呢?它能创建带哈希链式结构的区块序列,能通过工作量证明真正“挖”出新区块,能接收转账记录并打包进块里,还能通过HTTP接口查看整条链的状态。说白了,这是一个教学级别的骨架,但骨架恰恰是理解区块链最好的入口。

写这个项目之前,我给自己划了一条界线:目标是还原区块链的核心机制,而不是做一个生产级的分布式系统。所以下面所有代码都尽量单文件可运行,数据存在内存里,节点交互用最简单的HTTP请求完成。这样做的好处是,每一行代码都能对应到一个核心概念,你读完就能明白区块链的链式结构、哈希校验、工作量证明和基本记账逻辑是怎么回事。接下来我按设计思路、数据实现、PoW、交易与接口、排错这条路径展开。

1. 为什么用Python写区块链:设计思路与选型逻辑

1.1 先搞清楚Demo区块链到底由哪几部分组成

很多人一听到区块链就觉得玄乎,其实剥离掉那些商业叙事,最核心的骨架就四样东西:区块、链、共识规则、对外交互接口。区块是装数据的容器,链是给这些容器按顺序排好序的绳子,共识规则解决“谁有权加一个新块、怎么信任它”的问题,对外接口则让外部程序能往里写数据、读数据。

我习惯把这个结构类比成一本带页码和防伪印章的账本。每一页纸上记录若干笔账,同时在页眉写上“上一页印章的摘要”,最后盖上这一页自己的印章。如果有人偷偷改了第50页的内容,第51页的页眉摘要就跟第50页的实际内容对不上了,必须连第51页、第52页……全部重做一遍印章。区块链的本质就是这个思路,只不过“印章”换成了SHA-256哈希,“账本”从纸质换成了分布在各节点上的数据文件。

明白了这个类比,再看我后面的代码就会清楚很多:Block类负责造“页”,Blockchain类负责把“页”串成“本”并维护编号,工作量证明负责让每一页的“盖章”都有一点成本。至于Flask,只是方便你去翻阅这本账本。

1.2 技术选型:Python加Flask为什么刚刚好

选Python不是因为它比别的语言更适合区块链,而是因为它的表达成本很低。SHA-256有现成的hashlib实现,数据结构用dict加列表就能描述,序列化成JSON也就一行代码。对于教学型项目,你希望把80%的精力花在理解逻辑上,而不是花在内存管理和类型转换上。Python天然符合这个需求。

选Flask而不是Django,理由也很简单:这个Demo只需要四五个HTTP端点,Flask的轻量模型完全够用,而且启动一个服务只需要十几行代码。Django自带ORM、Admin、中间件这些重型装备,对一个小Demo来说是负担而不是帮助。

还有一点容易被忽略:我刻意不引入任何现成的区块链框架或SDK。原因不是它们不好,而是它们会帮你把底层细节全部封装掉。学习阶段最忌讳“全自动黑箱”,自己用基础库把区块、哈希、挖矿凑出来,虽然代码显得简陋,但每个环节都透明,出了问题也能一眼定位。

2. 区块数据结构与链式存储的实现

2.1 区块需要哪些字段,少一个都别扭

先看第一段核心代码,这里定义Block类。这个类承担两件事:保存一个区块应有的全部信息,以及能够随时重新计算自己的哈希。

import hashlib import json import time class Block: def __init__(self, index, transactions, timestamp, previous_hash, nonce=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): block_string = json.dumps({ "index": self.index, "transactions": self.transactions, "timestamp": self.timestamp, "previous_hash": self.previous_hash, "nonce": self.nonce }, sort_keys=True).encode() return hashlib.sha256(block_string).hexdigest()

index是区块在链上的位置编号,从0开始递增。transactions是这一块里打包的交易列表,这个Demo里用list of dict表示。timestamp是出块时间,我直接用time.time()生成的浮点数。previous_hash是前一个区块的SHA-256哈希,这是链式结构的核心引用。nonce是工作量证明阶段需要搜索的随机数,先默认置0。

hash字段是当前区块的指纹,由compute_hash()生成。计算时我把区块的五个关键字段序列化成JSON字符串,然后做SHA-256。这里有个很重要的细节:代码里加了sort_keys=True。Python的dict在序列化时不一定能保证字段顺序绝对稳定,尤其是在不同版本或不同环境下。只要字段顺序变了,同样内容的JSON字符串就会不同,算出来的哈希自然也不同,链的校验就会莫名其妙失败。加上sort_keys,等于强制按字典序排列字段,保证同一种内容永远得到同一种哈希结果。

2.2 哈希计算:为什么它是防篡改的关键

哈希在这里干了三件事。第一,它是内容的指纹。区块里任何一个字段,比如某笔转账金额从5改成了50,哪怕只改一个字符,重新计算哈希也会得到一串完全不同的64位十六进制字符串。第二,它是链的链接点。每个区块的previous_hash指向它前一个区块的哈希,等于把整条链焊死成一个整体。第三,它是工作量证明的搜索目标。后面你会看到,挖矿实际上就是不断调整nonce,直到算出的哈希满足某种特殊条件。

用生活化的方式来理解,哈希就是“盖章”。每盖一个新章,都必须把上一页的章内容掺和进来,所以只要有人动了前面任何一页,后面所有章的校验关系全部断裂。这就是为什么区块链不需要一个中心化的“管理员”来保证数据没被改——因为数据是否被改,看一眼前后哈希关系就能发现。

2.3 创世区块与链的初始化

有了区块类,下一步就是定义整条链。链本身不复杂,就是维护一个区块列表,外加一个等待打包的交易池。

class Blockchain: def __init__(self): self.chain = [] self.pending_transactions = [] self.difficulty = 4 self.miner_reward = 10 self.create_genesis_block() def create_genesis_block(self): genesis_block = Block(0, [], time.time(), "0" * 64, 0) genesis_block.hash = genesis_block.compute_hash() self.chain.append(genesis_block) @property def last_block(self): return self.chain[-1]

创建链的时候,chain列表是空的。我调用create_genesis_block()生成一个创世块,index为0,没有交易,previous_hash是64个0。为什么是64个0?因为SHA-256输出的十六进制字符串固定是64位,创世块没有前驱,就人为规定一个占位符。实际业务中,很多链的创世块还会写入一些特殊配置信息,但那不是教学Demo需要关心的。

pending_transactions是内存池,所有新提交但还没打包进区块的交易都会先放在这里。difficulty是挖矿难度,miner_reward是挖出新区块后奖励给矿工的代币数量。这两个参数放在链对象里,相当于全网共识的一部分,节点之间需要保持一致,否则一个说难度4、一个说难度5,校验结果就会冲突。

3. 工作量证明:让造块这件事“有成本”

3.1 为什么不能直接出块,非要搞一个PoW

假设出块不需要任何成本,那恶意节点就能干一件很讨厌的事:不停生成新块,把自己那条伪造的链做得特别长,然后告诉全网“我的链才是最长最权威的”,强行覆盖掉真实数据。工作量证明就是为了给“出块”这个动作增加成本,让伪造一条合法长链的代价变得高到不划算。

PoW的核心思路可以概括成“验证容易,生成困难”。验证一个区块的哈希是否满足条件,只需要重新算一次哈希,耗时几乎为零。但找到一个符合条件的nonce,往往需要穷举成千上万次。这个不对称性非常重要,它让节点没有动力去攻击整个网络,因为攻击消耗的算力成本远超可能获得的收益。当然,真实区块链的攻击模型比Demo复杂得多,但这个基本逻辑是相通的。

3.2 难度值与nonce搜索过程

接下来实现proof_of_work方法。它做的事情很简单:从nonce=0开始,不断计算区块哈希,直到哈希值的前4位变成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

这串循环看起来不起眼,但它就是“挖矿”的核心。difficulty=4意味着目标哈希必须以“0000”开头。哈希值每一位是0到f之间的十六进制字符,所以单次尝试命中的概率是1/16的4次方,也就是平均要尝试65536次才能成功。我自己的笔记本上跑difficulty=4,每个区块大约需要一两秒,体感刚好能让你“看出挖矿过程是有成本的”,又不至于等得烦躁。

下面这张表是我用同一台普通笔记本跑不同难度级别的粗略感受,可以作为参考来调节你自己项目的演示节奏:

difficulty目标前缀平均尝试次数我的实测耗时(每块)
2“00”256毫秒级,几乎瞬间
3“000”4096一两百毫秒
4“0000”655361到3秒
5“00000”104857615秒以上,不适合演示

教学场景建议用3或4。难度太低会让人觉得挖矿没有真实感,太高又会让一次普通演示变成漫长等待。线上体验类项目甚至可以把难度做成可配置,启动时从参数里读进来。

3.3 链上所有区块的完整性校验

挖矿只是出块,要保证整条链的数据没有被篡改,还需要一个校验方法。它的逻辑很直白:从第二个区块开始,逐块往后检查。第一,当前区块自己保存的hash是否等于重新计算出的哈希,不等说明区块内容被动过。第二,当前区块的previous_hash是否等于前一个区块的hash,不等说明链的链接关系被破坏了。第三,当前区块的哈希是否满足难度要求,这是对PoW合法性的验证。

def validate_chain(self, chain): for i in range(1, len(chain)): current = chain[i] previous = chain[i - 1] if current.hash != current.compute_hash(): return False if current.previous_hash != previous.hash: return False if not current.hash.startswith("0" * self.difficulty): return False return True

实现时要注意,这个版本接受的是Block对象列表。但后面用Flask做接口时,从一个节点拿到的是JSON,反序列化后会是dict列表,所以要么先把dict转回Block对象再调校验,要么在validate_chain里同时兼容两种格式。我后文会在几个节点之间做数据格式转换,就是为了避免这里埋坑。

4. 交易收集与记账逻辑

4.1 交易的结构设计与内存池

区块链如果只存空区块,那就只是个花架子。真实的价值在于记账,也就是把交易存进去。一个最小交易可以只包含三个字段:付款方sender、收款方recipient、金额amount。

def add_transaction(self, sender, recipient, amount): self.pending_transactions.append({ "sender": sender, "recipient": recipient, "amount": amount }) return self.last_block.index + 1

这个方法把一笔新交易塞进pending_transactions,然后返回下一区块的编号,也就是这比交易预计会被打包进哪个区块。调用方拿到这个编号就能给用户一个明确反馈:“你的交易被安排在了第N个区块”。

现实中的区块链要比这复杂得多。真实系统要考虑交易签名、公钥地址、余额校验、防双花、手续费机制等等。这个Demo全部跳过,因为教学目标是让大家看清“交易如何被收集、打包、随块上链”这条主线。如果一上来就把椭圆曲线签名、UTXO模型塞进去,大多数初学者的注意力就被拽到密码学细节里去了。

4.2 挖矿奖励与打包流程

有了交易池,接着实现mine_block。这个方法的执行流程是:先给矿工地址添加一条系统奖励交易,然后取内存池里的全部交易,创建一个新Block,通过PoW算出合法哈希,最后把区块追加到链上。

def mine_block(self, miner_address): self.add_transaction("SYSTEM", miner_address, self.miner_reward) new_block = Block( index=self.last_block.index + 1, transactions=self.pending_transactions, timestamp=time.time(), previous_hash=self.last_block.hash ) new_block.hash = self.proof_of_work(new_block) self.chain.append(new_block) self.pending_transactions = [] return new_block

这里有个必须注意的细节:add_transaction(“SYSTEM”, miner_address, reward)会先往pending_transactions里追加一笔奖励交易,然后new_block的transactions直接引用整个pending_transactions列表。追加完奖励交易、创建完区块之后,再清空pending_transactions,这样既保证奖励交易被包含进去,又不会让旧交易恒久占据内存池。

顺手提一下,这里奖励交易由“SYSTEM”发出,是一个简化写法。在真实链路里,它是一种没有发送方签名的特殊交易,通常叫“代币生成交易”或“区块奖励交易”,业务规则上跟普通转账是分开处理的。Demo里用同一个结构没问题,但如果以后要往生产方向演进,这段逻辑必须拆出来。

4.3 把交易和挖矿流程串成一个完整动作

单独看每个方法都不难,关键是理顺它们之间的调用关系。一个完整的数据上链流程是这样走的:

  1. 外部调用add_transaction,把一笔或多笔转账放进内存池。
  2. 调用mine_block,系统先产生一笔矿工奖励交易,再把全部待打包交易封装进新区块。
  3. PoW循环搜索nonce,直到区块哈希满足难度条件。
  4. 新区块append到chain,pending_transactions清空,流程结束。

写一个测试脚本把整条流程跑一遍:

blockchain = Blockchain() blockchain.add_transaction("alice", "bob", 5) blockchain.add_transaction("bob", "carol", 2) block = blockchain.mine_block("miner_node_001") print("区块哈希:", block.hash) print("交易数量:", len(block.transactions)) print("链长度:", len(blockchain.chain))

正常输出时,区块哈希是一串64位的十六进制字符,交易数量是3笔(两笔转账加一笔系统奖励),链长度从1变成2。每次出块后链都会增长,而哈希前缀的若干个0,正是工作量证明留下的“劳动痕迹”。

5. 用Flask暴露链与节点接口

5.1 五个核心HTTP端点,一个HTTP接口一座桥

代码写到这里还只是命令行级别的模块。为了让别人能通过HTTP往链上提交交易、查看链状态、参与挖矿,我加了Flask这一层。它相当于给链开了一扇门,外部程序用curl、Postman、浏览器就能操作这条链。

from flask import Flask, jsonify, request import requests app = Flask(__name__) node_identifier = "demo_miner_001" blockchain = Blockchain() @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(): data = request.get_json(force=True) required = ["sender", "recipient", "amount"] if data is None or not all(k in data for k in required): return "Missing values", 400 index = blockchain.add_transaction(data["sender"], data["recipient"], data["amount"]) return jsonify({"message": "交易将被添加到区块 " + str(index)}), 201 @app.route("/mine", methods=["GET"]) def mine(): if not blockchain.pending_transactions: return jsonify({"message": "没有待打包交易,请先添加交易再挖矿"}), 400 block = blockchain.mine_block(node_identifier) response = { "message": "新区块已生成", "index": block.index, "transactions": block.transactions, "previous_hash": block.previous_hash, "hash": block.hash, "nonce": block.nonce } return jsonify(response), 200 if __name__ == "__main__": app.run(host="127.0.0.1", port=5000)

/chain返回整条链,用于查看账本全貌。/transactions/new接收JSON格式的转账请求,参数校验通过后写入内存池。/mine触发一次挖矿,先把内存池里的交易打包成块,再返回新区块的关键信息。

这里我故意把/mine设置成GET请求,原因是浏览器直接访问就能触发,演示时比较省事。但从API设计规范来看,GET带副作用是不合适的,生产系统应该把它改成POST甚至是独立的采矿任务循环。教学项目里牺牲一下规范,换来的是演示便利。

5.2 从命令行到浏览器:实测整条链路

跑起来很简单:

pip install flask requests python blockchain.py

服务默认监听5000端口。先用curl添加两笔交易:

curl -X POST -H "Content-Type: application/json" \ -d '{"sender":"alice","recipient":"bob","amount":5}' \ http://127.0.0.1:5000/transactions/new

返回结果会提示“交易将被添加到区块 1”,这里返回的区块编号依据的是当前链上最后一个区块的index加1。然后触发挖矿:

curl http://127.0.0.1:5000/mine

返回的JSON里会包含新区块的index、交易列表、nonce、区块哈希等内容。再查看整条链:

curl http://127.0.0.1:5000/chain

你可以看到链上只有一个创世块加一个新区块,新区块里有三笔交易:alice付给bob 5、bob付给carol 2、系统奖励给demo_miner_001共10。走到这一步,一个完整的“提交交易-打包出块-查询账本”闭环就通了。

5.3 多节点注册与最长链共识机制

单节点能跑通,只完成了一半。区块链的一个核心卖点是去中心化,所以我让Demo支持多节点注册和共识切换。这里的共识规则用最经典也最容易理解的一条:当发现别的节点上的链更长、而且校验合法时,就把它替换成自己的链。

先加一个节点注册接口:

peers = set() @app.route("/nodes/register", methods=["POST"]) def register_node(): data = request.get_json() if data is None or "nodes" not in data: return "Error: nodes field missing", 400 for node in data["nodes"]: peers.add(node) return jsonify({"message": "节点注册成功", "peers": list(peers)}), 201

然后实现共识解析接口:

@app.route("/nodes/resolve", methods=["GET"]) def resolve_conflicts(): replaced = False best_chain = blockchain.chain max_length = len(best_chain) for node in peers: try: resp = requests.get(f"http://{node}/chain", timeout=2) except Exception: continue if resp.status_code != 200: continue candidate_length = resp.json()["length"] candidate_chain = [block_from_dict(b) for b in resp.json()["chain"]] if candidate_length > max_length and blockchain.validate_chain(candidate_chain): max_length = candidate_length best_chain = candidate_chain replaced = True if replaced: blockchain.chain = best_chain return jsonify({"message": "链已更新为最长有效链", "chain": [b.__dict__ for b in blockchain.chain]}), 200 return jsonify({"message": "当前链已是最新", "chain": [b.__dict__ for b in blockchain.chain]}), 200

在网络传输中,区块经过JSON序列化后会变成dict,而内存中的区块是Block对象。我准备了一个转换函数,用来在两者之间切换,保证validate_chain拿到的是一组正确类型的对象:

def block_from_dict(data): block = Block( index=data["index"], transactions=data["transactions"], timestamp=data["timestamp"], previous_hash=data["previous_hash"], nonce=data["nonce"] ) block.hash = data["hash"] return block

这个函数的定位就像一个“单位换算器”。外部的JSON没有直接操作Block属性的能力,你也不能把一个dict直接挂到chain列表里,否则后续调用last_block.compute_hash时就会报AttributeError。很多初学者在写多节点同步时出错,核心原因就是忽略了dict和对象之间的身份转换。

6. 常见问题与排查技巧实录

6.1 哈希永远对不上:先查序列化顺序和字段类型

我在测试中遇到的第一个典型问题,是validate_chain一直返回False。排查下来发现,要么是compute_hash里没加sort_keys=True,导致同一区块在不同时间算出的哈希不稳定;要么是交易数据里混进了自定义类型,json.dumps直接抛TypeError。

解决思路很简单:哈希计算的输入字符串必须严格确定。所有字段固定、顺序固定、类型可序列化,三者缺一不可。如果交易里将来要放时间对象、Decimal金额之类的高级类型,请先把它们统一转换成字符串或数字,再参与哈希计算。

另外还有一个隐蔽坑:如果Block对象创建之后,你又修改了它的某个字段,那么它内部保存的hash和重新计算出的hash就会不一致,validate_chain自然返回False。教学场景下,创建区块后尽量不要去改它的索引、时间戳、交易内容等核心字段。如果你确实需要修改,记得同步重算哈希。

6.2 flask的JSON返回异常:格式化与兼容性问题

Flask的jsonify对float、int、str这几种基础类型处理得很好,但一旦数据里出现bytes、set、自定义对象,直接序列化就会报错。比如我把区块时间戳用time.time()保存成浮点数,这是安全的;但如果有人手滑加了set类型,jsonify会当场崩溃。

更推荐的做法是在返回前做一次显式的数据清洗,把区块对象转成dict,再把dict里的非基础类型统一转成基础类型。我的/chain接口里写了[block.dictfor block in blockchain.chain],就是这种思路。以后如果扩展交易字段,优先保证所有字段都能被JSON原生支持。

6.3 两个节点数据不一致:手动触发同步的边界

节点A挖出了新区块,节点B因为还没有调用过resolve,它的链比A短。这个Demo里节点之间不会自动交换信息,需要手动调用/nodes/register注册对端,再通过/nodes/resolve触发同步。真实网络中是持续广播和同步的,但教学项目用“手动拉取”更能让大家看清每个动作的语义。

同时要明确一点:最长链规则能成立的先决条件,是网络上大部分算力由诚实节点掌握。如果一个恶意节点掌握超过全网半数的算力,理论上它可以构造出一条更长更“合法”的链来覆盖数据。虽然这被称为“恶意攻击”,但它并不算数据篡改,因为每个区块的哈希和PoW都是真实有效的。这个边界属于共识协议的研究范畴,Demo里不做展开。

6.4 这个教学骨架还能接着加什么

如果跑通了上面的Demo,还想继续深入,我建议按下面几个方向扩展,难度从低到高依次递进:

  • 给交易加上密钥签名,并验证签名是否匹配,让任意节点不能冒名转账。
  • 把区块内部的交易列表换成Merkle树根哈希,这样单块内大量交易的完整性校验就不需要遍历全部内容。
  • 用SQLite或普通文件持久化区块,让重启节点后链不会消失。
  • 用WebSocket替换HTTP轮询,让节点之间可以实现一定程度的自动广播。
  • 在出块难度里引入动态调整逻辑,让全网平均出块时间保持在一个稳定区间。

这些扩展方向每一条都能独立成篇,也都能让你对区块链某个细分机制的理解加深一层。我的建议是不要贪多,一个一个来,改完一次就对比一下“现在的链和之前的链在结构上发生了什么变化”。

最后说一点自己做这个项目时最深的体会。我一开始写代码时只想着快跑通,结果debug时间全花在序列化和哈希不稳定这些小问题上。后来耐下性子把数据格式、字段规则一件件理清楚,整条链就很少再出幺蛾子了。如果你也想自己搭一个,我建议别急着抄完整代码,先把Block和Blockchain两个类自己写一遍,跑通之后再上Flask接口,这样每层都会留下更深的肌肉记忆。等这套骨架玩熟了,你再看市面上的各种区块链技术资料,会发现很多原来听不懂的概念,其实是你亲手写过的东西。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询