python多节点同步测试
上一篇只做到单节点的功能测试，这里继续多节点。
import hashlib
import json
import time
from urllib.parse import urlparse
from uuid import uuid4
import requests
from flask import Flask, jsonify, request
from typing import Any, Dict, List, Optional # 数据结构
''' typing.Optional 可选类型。 Optional[X] 等价于 Union[X, None]。也就是要么是X，要么是None 请注意，这不是一个可选参数的概念，它是一个具有默认值的参数。具有默认值的可选参数不需要在其类型注释上使用 Optional 限定符（虽然如果默认值为 None，则推断）。 如果允许 None 的显式值，则强制参数可以仍然具有 Optional 类型。 typing.Any 表示无约束类型的特殊类型。 每种类型都与 Any 兼容。 Any 兼容各种类型。 '''
class DadaCoinBlockChain:
def __init__(self):
self.current_transactions = [] # 交易列表
self.chain = [] # 区块链
self.nodes = set() # 保存网络节点
self.new_block(proof=100, preHash="1") # 创世区块
##创建区块
def new_block(self, proof: int, # 指定参数为int preHash: Optional[str] # 指定上一块的哈希数据类型为str，默认为None ) -> Dict[str, Any]: # 指定返回数据类型
block = {
"index": len(self.chain) + 1, # 索引
"timestamp": time.time(), # 当前时间
"transactions": self.current_transactions,
"proof": proof,
"preHash": preHash or self.hash(self.chain[-1])
}
self.current_transactions = []
self.chain.append(block)
return block
@staticmethod
def hash(block: Dict[str, Any]) -> str:
block_str = json.dumps(block, sort_keys=True).encode("utf-8")
return hashlib.sha256(block_str).hexdigest()
def new_transaction(self, sender: str, receiver: str, amount: int) -> int: # 创建交易
transaction = {
"sender": sender,
"receiver": receiver,
"amount": amount,
}
self.current_transactions.append(transaction)
return self.last_block["index"] + 1 # 索引标记交易的数量
@property
def last_block(self) -> Dict[str, Any]:
return self.chain[-1]
def proof_of_work(self, last_proof: int) -> int: # 工作量证明
proof = 0
while self.valid_proof(last_proof, proof) is False:
proof += 1
return proof
@staticmethod
def valid_proof(last_proof: int, proof: int) -> bool: # 验证证明
guess = f'{last_proof*proof}'.encode("utf-8")
guess_hash = hashlib.sha256(guess).hexdigest()
return guess_hash[-4:] == "1234"
def register_node(self, addr: str) -> None: # 加入网络的其他节点，用于更新
now_url = urlparse(addr)
self.nodes.add(now_url.netloc) # 增加网络节点
pass
def valid_chain(self, chain: List[Dict[str, Any]]) -> bool: # 区块链校验
last_block = self.chain[0]
current_index = 1
while current_index < len(self.chain):
block = self.chain[current_index]
# 哈希校验
if block["preHash"] != self.hash(last_block):
return False
# 工作量校验
if not self.valid_proof(last_block["proof"], block["proof"]):
return False
last_block = block
current_index += 1
return True
def resolve_conflict(self) -> bool: # 同步区块链
neighbours = self.nodes
new_chain = None
max_length = len(self.chain)
for node in neighbours:
response = requests.post(f'http://{node}/chain')
if response.status_code == 200:
chain = response.json()["chain"]
length = response.json()["length"]
if length > max_length and self.valid_chain(chain):
max_length = length
new_chain = chain
if new_chain:
self.chain = new_chain
return True
else:
return False
datacoin = DadaCoinBlockChain() # 创建一个网络节点
node_id = str(uuid4()).replace("-", "") # 生成钱包地址
print("当前节点钱包地址：", node_id)
app = Flask(__name__) # 初始化
# 首页
@app.route("/", methods=["POST"])
def index_page():
return "欢迎来到大大币系统，你好小韭菜"
# 获取区块链
@app.route("/chain", methods=["POST"])
def index_chain():
response = {
"chain": datacoin.chain, # 区块链
"length": len(datacoin.chain),
}
return jsonify(response), 200
# 创建新的交易
@app.route("/new_transaction", methods=["POST"])
def new_transaction():
values = request.get_json()
print("values:", values)
required = ["sender", "receiver", "amount"]
if not all(key in values for key in required):
return "数据不完整", 400
index = datacoin.new_transaction(values["sender"], values["receiver"], values["amount"])
response = {
"message": f'交易加入到区块{index}',
}
return jsonify(response), 200
# 挖矿
@app.route("/mine", methods=["POST"])
def index_mine():
last_block = datacoin.last_block
last_proof = last_block["proof"]
proof = datacoin.proof_of_work(last_proof) # 挖矿计算
# 系统奖励，挖矿产生交易
datacoin.new_transaction(sender="0", receiver=node_id, amount=10)
block = datacoin.new_block(proof, None) # 增加一个区块
response = {
"message": "新的区块创建",
"index": block["index"],
"transactions": block["transactions"],
"proof": block["proof"],
"preHash": block["preHash"],
}
return jsonify(response), 200
# 新增节点
@app.route("/new_node", methods=["POST"])
def new_node():
values = request.get_json()
nodes = values.get("nodes")
if nodes is None:
return "你妹的啥数据", 400
for node in nodes:
datacoin.register_node(node)
response = {
"message": f'网络节点已被追加',
"nodes": list(datacoin.nodes),
}
return jsonify(response), 200
# 刷新节点
@app.route("/node_refresh", methods=["POST"])
def node_refresh():
replaced = datacoin.resolve_conflict() # 共识算法进行最长替换
if replaced:
response = {
"message": "区块链已被替换为最长", # 区块链
"new_chain": datacoin.chain,
}
pass
else:
response = {
"message": "当前区块链已是最长，无需替换", # 区块链
"new_chain": datacoin.chain,
}
return jsonify(response), 200
if __name__ == "__main__":
app.run("127.0.0.1", port=5000)
这里复制三份代码，端口分别是5000,5002,5003，代表三个节点，全部启动。
1,5000节点，挖矿6个区块，5002挖矿2个区块，5003挖矿3个区块
2,5002添加两个节点,
{
"nodes": ["http://127.0.0.1:5000","http://127.0.0.1:5003"]
}5003添加两个节点
{
"nodes": ["http://127.0.0.1:5000","http://127.0.0.1:5002"]
}
3,节点同步，
顺利完成前面的功能测试，这部分的开发就完成了
更多精彩内容
