首页 > 专题 > 区块链 > 正文

用Python实现一个区块链

2020-02-25 19:29:02
字体:
来源:转载
供稿:网友

  相信你和我一样对数字货币的崛起感到新奇,并且想知道其背后的技术——区块链是怎样实现的。

  但是理解区块链并非易事,至少对于我来说是如此。晦涩难懂的视频、漏洞百出的教程以及示例的匮乏令我倍受挫折。

  我喜欢在实践中学习,通过写代码来学习技术会掌握得更牢固。如果你也这样做,那么读完本文,你将获得一个可用的区块链以及对区块链的深刻理解。

  开始之前...

  首先你需要知道区块链是由被称为区块的记录构成的不可变的、有序的链式结构,这些记录可以是交易、文件或任何你想要的数据,最重要的是它们是通过 Hash 连接起来的。

  如果你不了解 Hash,这里有个例子

  其次,你需要安装 Python3.6+,Flask,Request

  pip installFlask==0.12.2requests==2.18.4

  同时你还需要一个 HTTP 客户端,比如 Postman,cURL 或任何其它客户端。

  最终的源代码在这里:

  第一步: 打造一个 Blockchain

  新建一个文件 blockchain.py,本文所有的代码都写在这一个文件中。首先创建一个 Blockchain 类,在构造函数中我们创建了两个列表,一个用于储存区块链,一个用于储存交易。

  classBlockchain(object):

  def__init__(self):

  self.chain=[]

  self.current_transactions=[]

  defnew_block(self):

  # Creates a new Block and adds it to the chain

  pass

  defnew_transaction(self):

  # Adds a new transaction to the list of transactions

  pass

  @staticmethod

  defhash(block):

  # Hashes a Block

  pass

  @property

  deflast_block(self):

  # Returns the last Block in the chain

  pass

  一个区块有五个基本属性:index,timestamp(in Unix time),transaction 列表,工作量证明(稍后解释)以及前一个区块的 Hash 值。

  block={

  'index':1,

  'timestamp':1506057125.900785,

  'transactions':[

  {

  'sender':"8527147fe1f5426f9dd545de4b27ee00",

  'recipient':"a77f5cdfa2934df3954a5c7c7da5df1f",

  'amount':5,

  }

  ],

  'proof':324984774000,

  'previous_hash':"2cf24dba5fb0a30e26e83b2ac5b9e29e1b161e5c1fa7425e73043362938b9824"

  }

  到这里,区块链的概念应该比较清楚了:每个新的区块都会包含上一个区块的 Hash 值。这一点非常关键,它是区块链不可变性的根本保障。如果攻击者破坏了前面的某个区块,那么后面所有区块的 Hash 都会变得不正确。不理解?慢慢消化~

  我们需要一个向区块添加交易的方法:

  classBlockchain(object):

  ...

  defnew_transaction(self,sender,recipient,amount):

  """

  Creates a new transaction to go into the next mined Block

  :param sender: Address of the Sender

  :param recipient: Address of the Recipient

  :param amount: Amount

  :return: The index of the Block that will hold this transaction

  """

  self.current_transactions.append({

  'sender':sender,

  'recipient':recipient,

  'amount':amount,

  })

  returnself.last_block['index']+1

  new_transaction() 方法向列表中添加一个交易记录,并返回该记录将被添加到的区块——下一个待挖掘的区块——的索引,稍后在用户提交交易时会有用。

  当 Blockchain 实例化后,我们需要创建一个初始的区块(创世块),并且给它预设一个工作量证明。

  除了添加创世块的代码,我们还需要补充 newblock(), newtransaction() 和 hash() 方法:

  importhashlib

  importjson

  fromtimeimporttime

  classBlockchain(object):

  def__init__(self):

  self.current_transactions=[]

  self.chain=[]

  # Create the genesis block

  self.new_block(previous_hash=1,proof=100)

  defnew_block(self,proof,previous_hash=None):

  block={

  'index':len(self.chain)+1,

  'timestamp':time(),

  'transactions':self.current_transactions,

  'proof':proof,

  'previous_hash':previous_hashorself.hash(self.chain[-1]),

  }

  # Reset the current list of transactions

  self.current_transactions=[]

  self.chain.append(block)

  returnblock

  defnew_transaction(self,sender,recipient,amount):

  self.current_transactions.append({

  'sender':sender,

  'recipient':recipient,

  'amount':amount,

  })

  returnself.last_block['index']+1

  @property

  deflast_block(self):

  returnself.chain[-1]

  @staticmethod

  defhash(block):

  block_string=json.dumps(block,sort_keys=True).encode()

  returnhashlib.sha256(block_string).hexdigest()

  上面的代码应该很直观,我们基本上有了区块链的雏形。但此时你肯定很想知道一个区块究竟是怎样被创建或挖掘出来的。

  新的区块来自工作量证明(PoW)算法。PoW 的目标是计算出一个符合特定条件的数字,这个数字对于所有人而言必须在计算上非常困难,但易于验证。这就是工作量证明的核心思想。

  举个例子:

  假设一个整数 x 乘以另一个整数 y 的积的 Hash 值必须以 0 结尾,即 hash(x * y) = ac23dc...0。设 x = 5,求 y?

  fromhashlibimportsha256

  x=5

  y=0# We don't know what y should be yet...

  whilesha256(f'{x*y}'.encode()).hexdigest()[-1]!="0":

  y+=1

  print(f'The solution is y = {y}')

  结果是 y = 21 // hash(5 * 21) = 1253e9373e...5e3600155e860

  在比特币中,工作量证明算法被称为 Hashcash,它和上面的问题很相似,只不过计算难度非常大。这就是矿工们为了争夺创建区块的权利而争相计算的问题。通常,计算难度与目标字符串需要满足的特定字符的数量成正比,矿工算出结果后,就会获得一定数量的比特币奖励(通过交易)。

  网络要验证结果,当然非常容易。

  让我们来实现一个 PoW 算法,和上面的例子非常相似,规则是:寻找一个数 p,使得它与前一个区块的 proof 拼接成的字符串的 Hash 值以 4 个零开头。

  importhashlib

  importjson

  fromtimeimporttime

  fromuuidimportuuid4

  classBlockchain(object):

  ...

  defproof_of_work(self,last_proof):

  proof=0

  whileself.valid_proof(last_proof,proof)isFalse:

  proof+=1

  returnproof

  @staticmethod

  defvalid_proof(last_proof,proof):

  guess=f'{last_proof}{proof}'.encode()

  guess_hash=hashlib.sha256(guess).hexdigest()

  returnguess_hash[:4]=="0000"

  衡量算法复杂度的办法是修改零的个数。4 个零足够用于演示了,你会发现哪怕多一个零都会大大增加计算出结果所需的时间。

  我们的 Blockchain 基本已经完成了,接下来我们将使用 HTTP requests 来与之交互。

  第二步:作为 API 的 Blockchain

  我们将使用 Flask 框架,它十分轻量并且很容易将网络请求映射到 Python 函数。

  我们将创建三个接口:

  /transactions/new创建一个交易并添加到区块

  /mine告诉服务器去挖掘新的区块

  /chain返回整个区块链

  我们的服务器将扮演区块链网络中的一个节点。我们先添加一些常规代码:

  importhashlib

  importjson

  fromtextwrapimportdedent

  fromtimeimporttime

  fromuuidimportuuid4

  fromflaskimportFlask,jsonify,request

  classBlockchain(object):

  ...

  # Instantiate our Node

  app=Flask(__name__)

  # Generate a globally unique address for this node

  node_identifier=str(uuid4()).replace('-','')

  # Instantiate the Blockchain

  blockchain=Blockchain()

  @app.route('/mine',methods=['GET'])

  defmine():

  return"We'll mine a new Block"

  @app.route('/transactions/new',methods=['POST'])

  defnew_transaction():

  return"We'll add a new transaction"

  @app.route('/chain',methods=['GET'])

  deffull_chain():

  response={

  'chain':blockchain.chain,

  'length':len(blockchain.chain),

  }

  returnjsonify(response),200

  if__name__=='__main__':

  app.run(host='127.0.0.1',port=5000)

  这是用户发起交易时发送到服务器的请求:

  {

  "sender":"my address",

  "recipient":"someone else's address",

  "amount":5

  }

  我们已经有了向区块添加交易的方法,因此剩下的部分就很简单了:

  @app.route('/transactions/new',methods=['POST'])

  defnew_transaction():

  values=request.get_json()

  # Check that the required fields are in the POST'ed data

  required=['sender','recipient','amount']

  ifnotall(kinvaluesforkinrequired):

  return'Missing values',400

  # Create a new Transaction

  index=blockchain.new_transaction(values['sender'],values['recipient'],values['amount'])

  response={'message':f'Transaction will be added to Block {index}'}

  returnjsonify(response),201

  挖掘端正是奇迹发生的地方,它只做三件事:计算 PoW;通过新增一个交易授予矿工一定数量的比特币;构造新的区块并将其添加到区块链中。

  @app.route('/mine',methods=['GET'])

  defmine():

  # We run the proof of work algorithm to get the next proof...

  last_block=blockchain.last_block

  last_proof=last_block['proof']

  proof=blockchain.proof_of_work(last_proof)

  # We must receive a reward for finding the proof.

  # The sender is "0" to signify that this node has mined a new coin.

  blockchain.new_transaction(

  sender="0",

  recipient=node_identifier,

  amount=1,

  )

  # Forge the new Block by adding it to the chain

  block=blockchain.new_block(proof)

  response={

  'message':"New Block Forged",

  'index':block['index'],

  'transactions':block['transactions'],

  'proof':block['proof'],

  'previous_hash':block['previous_hash'],

  }

  returnjsonify(response),200

  需注意交易的接收者是我们自己的服务器节点,目前我们做的大部分事情都只是围绕 Blockchain 类进行交互。到此,我们的区块链就算完成了。

  第三步:交互演示

  使用 Postman 演示,略。

  第四步:一致性

  这真的很棒,我们已经有了一个基本的区块链可以添加交易和挖矿。但是,整个区块链系统必须是分布式的。既然是分布式的,那么我们究竟拿什么保证所有节点运行在同一条链上呢?这就是一致性问题,我们要想在网络中添加新的节点,就必须实现保证一致性的算法。

  在实现一致性算法之前,我们需要找到一种方式让一个节点知道它相邻的节点。每个节点都需要保存一份包含网络中其它节点的记录。让我们新增几个接口:

  1./nodes/register接收以URL的形式表示的新节点的列表

  2./nodes/resolve用于执行一致性算法,用于解决任何冲突,确保节点拥有正确的链

  ...

  fromurllib.parseimporturlparse

  ...

  classBlockchain(object):

  def__init__(self):

  ...

  self.nodes=set()

  ...

  defregister_node(self,address):

  parsed_url=urlparse(address)

  self.nodes.add(parsed_url.netloc)

  注意到我们用 set 来储存节点,这是一种避免重复添加节点的简便方法。

  前面提到的冲突是指不同的节点拥有的链存在差异,要解决这个问题,我们规定最长的合规的链就是最有效的链,换句话说,只有最长且合规的链才是实际存在的链。

  让我们再添加两个方法,一个用于添加相邻节点,另一个用于解决冲突。

  ...

  importrequests

  classBlockchain(object)

  ...

  defvalid_chain(self,chain):

  last_block=chain[0]

  current_index=1

  whilecurrent_index(chain):

  block=chain[current_index]

  print(f'{last_block}')

  print(f'{block}')

  print("n-----------n")

  # Check that the hash of the block is correct

  ifblock['previous_hash']!=self.hash(last_block):

  returnFalse

  # Check that the Proof of Work is correct

  ifnotself.valid_proof(last_block['proof'],block['proof']):

  returnFalse

  last_block=block

  current_index+=1

  returnTrue

  defresolve_conflicts(self):

  neighbours=self.nodes

  new_chain=None

  # We're only looking for chains longer than ours

  max_length=len(self.chain)

  # Grab and verify the chains from all the nodes in our network

  fornodeinneighbours:

  response=requests.get(f'http://{node}/chain')

  ifresponse.status_code==200:

  length=response.json()['length']

  chain=response.json()['chain']

  # Check if the length is longer and the chain is valid

  iflength>max_lengthandself.valid_chain(chain):

  max_length=length

  new_chain=chain

  # Replace our chain if we discovered a new, valid chain longer than ours

  ifnew_chain:

  self.chain=new_chain

  returnTrue

  returnFalse

  现在你可以新开一台机器,或者在本机上开启不同的网络接口来模拟多节点的网络,或者邀请一些朋友一起来测试你的区块链。

  我希望本文能激励你创造更多新东西。我之所以对数字货币入迷,是因为我相信区块链会很快改变我们看待事物的方式,包括经济、政府、档案管理等。

发表评论 共有条评论
用户名: 密码:
验证码: 匿名发表