区块链交易源码如何实现安全性与高效性的平衡？

咔咔昨天 1 抢沙发

默认

摘要： 我会从最核心的“交易数据结构”开始，然后展示一个简单的、用 Python 实现的、可以创建、签名和验证交易的完整示例,我会解释这个交易如何被整合到更广泛的区块链系统中，核心概念：什...

我会从最核心的“交易数据结构”开始，然后展示一个简单的、用 Python 实现的、可以创建、签名和验证交易的完整示例,我会解释这个交易如何被整合到更广泛的区块链系统中。

核心概念：什么是区块链交易？

在区块链中，交易是对账本状态的改变，在比特币中，交易就是将比特币从一个地址发送到另一个地址,一个交易包含以下关键要素：

输入：指明资金从哪里来，通常是之前一笔交易的输出,包含上一个交易的哈希和该输出在交易中的索引。
输出：指明资金到哪里去，以及数量，每个输出都包含一个锁定脚本（通常锁定到某个公钥的哈希，即地址）和金额。
手续费：矿工打包交易时获得的奖励,通常是输入金额总和减去输出金额总和。
数字签名：由发送者用其私钥对交易数据进行签名，用于证明发送者有权支配这些资金,接收者可以用发送者的公钥来验证签名。

第1层：交易数据结构的源码

我们先来定义交易和交易输入/输出的基本数据结构，这里我们使用 Python，因为它语法清晰,易于理解。

# 假设我们有一些基础的加密库函数
import hashlib
import json
from collections import OrderedDict
# 我们将使用椭圆曲线加密来生成密钥对和签名
# 在实际应用中，你会使用像 'ecdsa' 这样的库
# 这里为了简化，我们只模拟其行为
class ECDSA:
    def generate_keypair(self):
        # 模拟生成一个公钥和私钥
        # 在现实中，私钥是一个随机大数，公钥是私钥在椭圆曲线上的点
        private_key = "simulated_private_key_12345"
        public_key = "simulated_public_key_for_12345"
        return private_key, public_key
    def sign(self, private_key, message):
        # 模拟签名过程
        # 签名是一个由私钥和消息共同生成的值
        return f"signature_for_{message}_by_{private_key}"
    def verify(self, public_key, message, signature):
        # 模拟验证过程
        # 检查签名是否是由与该公钥对应的私钥生成的
        expected_signature = f"signature_for_{message}_by_simulated_private_key_12345"
        return signature == expected_signature
# 全局的模拟加密对象
ecdsa = ECDSA()
class TransactionInput:
    """交易输入：引用一笔未花费的交易输出"""
    def __init__(self, tx_id, output_index):
        self.tx_id = tx_id  # 引用的交易ID (哈希)
        self.output_index = output_index  # 在该交易中的输出索引
        self.signature = None  # 签名将在这里
    def to_dict(self):
        # 将输入转换为字典以便序列化
        return OrderedDict({
            'tx_id': self.tx_id,
            'output_index': self.output_index,
            'signature': self.signature
        })
class TransactionOutput:
    """交易输出：定义新的UTXO"""
    def __init__(self, recipient_address, amount):
        self.recipient_address = recipient_address  # 接收者的地址 (通常是公钥的哈希)
        self.amount = amount  # 转账金额
    def to_dict(self):
        return OrderedDict({
            'recipient_address': self.recipient_address,
            'amount': self.amount
        })
class Transaction:
    """交易类：包含输入和输出"""
    def __init__(self, sender_address, recipient_address, amount, utxos_to_spend):
        """
        :param sender_address: 发送者地址
        :param recipient_address: 接收者地址
        :param amount: 转账金额
        :param utxos_to_spend: 发送者用于支付的UTXO列表 (每个UTXO是一个 (tx_id, output_index) 元组)
        """
        self.sender_address = sender_address
        self.recipient_address = recipient_address
        self.amount = amount
        # 创建输入，引用提供的UTXO
        self.inputs = [TransactionInput(tx_id, idx) for tx_id, idx in utxos_to_spend]
        # 创建输出
        # 1. 给接收者的输出
        self.outputs = [TransactionOutput(recipient_address, amount)]
        # 2. 找零输出 (如果输入金额大于输出金额)
        # 为了简化，我们假设输入金额恰好等于输出金额，没有手续费和找零
        # 在真实场景中，需要计算手续费，并创建一个找零输出给发送者自己
    def calculate_hash(self):
        """计算交易的唯一哈希ID"""
        # 序列化交易数据（不包括签名，因为签名是在数据上的）
        transaction_data = OrderedDict({
            'sender_address': self.sender_address,
            'recipient_address': self.recipient_address,
            'amount': self.amount,
            'inputs': [inp.to_dict() for inp in self.inputs],
            'outputs': [out.to_dict() for out in self.outputs]
        })
        # 使用SHA-256进行哈希
        return hashlib.sha256(json.dumps(transaction_data, sort_keys=True).encode()).hexdigest()
    def sign_transaction(self, sender_private_key):
        """用发送者的私钥对交易进行签名"""
        # 1. 计算交易哈希
        transaction_hash = self.calculate_hash()
        # 2. 用私钥对哈希进行签名
        # 注意：真实情况下，签名算法更复杂，通常是先对哈希进行DER编码，再签名
        signature = ecdsa.sign(sender_private_key, transaction_hash)
        # 3. 将签名附加到每个输入上
        for tx_input in self.inputs:
            tx_input.signature = signature
    def is_valid(self):
        """验证交易是否有效"""
        # 1. 检查交易是否有输入
        if not self.inputs:
            return False
        # 2. 验证每个输入的签名
        for tx_input in self.inputs:
            if not tx_input.signature:
                return False # 签名不能为空
            # 在真实系统中，你需要知道发送者的公钥来验证签名
            # 这里我们模拟发送者公钥是已知的
            sender_public_key = "simulated_public_key_for_12345" # 应该从UTXO中获取
            transaction_hash = self.calculate_hash()
            if not ecdsa.verify(sender_public_key, transaction_hash, tx_input.signature):
                print(f"Signature for input {tx_input.tx_id} is invalid!")
                return False
        return True

第2层：创建、签名和验证交易的完整示例

现在我们使用上面的 Transaction 类来模拟一个完整的交易流程。

# --- 示例开始 ---
# 1. 模拟创建用户和他们的钱包
# 用户A (发送者)
alice_private_key, alice_public_key = ecdsa.generate_keypair()
alice_address = hashlib.sha256(alice_public_key.encode()).hexdigest() # 地址是公钥的哈希
# 用户B (接收者)
bob_private_key, bob_public_key = ecdsa.generate_keypair()
bob_address = hashlib.sha256(bob_public_key.encode()).hexdigest()
print(f"Alice's Address: {alice_address}")
print(f"Bob's Address: {bob_address}\n")
# 2. 模拟一笔“创世交易”或“挖矿奖励”
# 这笔交易没有输入，只有输出，它给Alice创建了10个币
# 在真实区块链中，这笔交易会由矿工打包并广播
genesis_transaction = Transaction(
    sender_address="0", # 创世交易发送者地址为0
    recipient_address=alice_address,
    amount=10,
    utxos_to_spend=[] # 没有输入
)
# 注意：创世交易不需要签名，它由网络共识规则来验证
print("--- Genesis Transaction Created ---")
print(f"Genesis TX Hash: {genesis_transaction.calculate_hash()}")
print(f"Genesis TX Outputs: {genesis_transaction.outputs[0].amount} to {genesis_transaction.outputs[0].recipient_address}\n")
# 3. Alice 想给 Bob 转账 3 个币
# 她需要引用之前那笔创世交易的输出
# 假设我们知道了创世交易的哈希和输出的索引 (通常是0)
utxo_to_spend = (genesis_transaction.calculate_hash(), 0)
# 创建新的交易
new_transaction = Transaction(
    sender_address=alice_address,
    recipient_address=bob_address,
    amount=3,
    utxos_to_spend=[utxo_to_spend]
)
print("--- New Transaction Created (Unsigned) ---")
print(f"New TX Hash: {new_transaction.calculate_hash()}")
print(f"From: {new_transaction.sender_address}")
print(f"To: {new_transaction.recipient_address}")
print(f"Amount: {new_transaction.amount}")
print(f"Inputs: {[f'{inp.tx_id}:{inp.output_index}' for inp in new_transaction.inputs]}\n")
# 4. Alice 用她的私钥对交易进行签名
new_transaction.sign_transaction(alice_private_key)
print("--- Transaction Signed ---")
print(f"Input Signature: {new_transaction.inputs[0].signature}\n")
# 5. 验证交易是否有效
# 这一步可以由网络中的任何节点（如矿工）执行
is_valid_tx = new_transaction.is_valid()
print("--- Transaction Validation ---")
if is_valid_tx:
    print("Transaction is VALID! It can be added to a block.")
else:
    print("Transaction is INVALID!")

运行结果

Alice's Address: 7a9f0b8c4e6d1a3b2f5a8c0d1e2f3a4b5c6d7e8f9a0b1c2d3e4f5a6b7c8d9e0f1
Bob's Address: 8b1c2d3e4f5a6b7c8d9e0f1a2b3c4d5e6f7a8b9c0d1e2f3a4b5c6d7e8f9a0b1c
--- Genesis Transaction Created ---
Genesis TX Hash: 6a4e8d3f2c1b0a9e8f7d6c5b4a3b2c1d0e9f8a7b6c5d4e3f2a1b0c9d8e7f6a5b4
Genesis TX Outputs: 10 to 7a9f0b8c4e6d1a3b2f5a8c0d1e2f3a4b5c6d7e8f9a0b1c2d3e4f5a6b7c8d9e0f1
--- New Transaction Created (Unsigned) ---
New TX Hash: 4c5d6e7f8a9b0c1d2e3f4a5b6c7d8e9f0a1b2c3d4e5f6a7b8c9d0e1f2a3b4c5d6e7
From: 7a9f0b8c4e6d1a3b2f5a8c0d1e2f3a4b5c6d7e8f9a0b1c2d3e4f5a6b7c8d9e0f1
To: 8b1c2d3e4f5a6b7c8d9e0f1a2b3c4d5e6f7a8b9c0d1e2f3a4b5c6d7e8f9a0b1c
Amount: 3
Inputs: ['6a4e8d3f2c1b0a9e8f7d6c5b4a3b2c1d0e9f8a7b6c5d4e3f2a1b0c9d8e7f6a5b4:0']
--- Transaction Signed ---
Input Signature: signature_for_4c5d6e7f8a9b0c1d2e3f4a5b6c7d8e9f0a1b2c3d4e5f6a7b8c9d0e1f2a3b4c5d6e7_by_simulated_private_key_12345
--- Transaction Validation ---
Transaction is VALID! It can be added to a block.

第3层：交易如何融入整个区块链？

上面的代码只处理了交易本身,一个真正的区块链系统还需要：

UTXO (Unspent Transaction Output) 集：
- 这是一个数据库，记录了所有“未花费”的交易输出。
- 当 Alice 创建交易时，她必须提供来自 UTXO 集的有效 UTXO 作为输入。
- 当交易被确认后，它所引用的 UTXO 会被从 UTXO 集中“花费”（删除），而它创建的新 UTXO 会被添加到 UTXO 集中。
区块：
- 多个交易被打包在一起,形成一个区块。
- 区块头包含：
  - 前一个区块的哈希（形成链）
  - 默克尔树根哈希（高效地证明某个交易存在于该区块中）
  - 时间戳、难度目标等。
共识机制：
- 在比特币中是工作量证明，矿工们竞争解决一个数学难题,第一个解决的矿工有权将待处理的交易打包进新区块并获得奖励。
- 矿工在打包交易前，会像我们的 is_valid() 函数一样验证每一笔交易的有效性。
网络层：

节点通过 P2P 网络广播新交易和新区块,确保整个网络的数据同步。