东营市城乡建设局网站百度大数据官网入口

目录

一、什么是Turbine算法?

二、流程图

三、模拟实现

3.1 Leader实现

3.2 Validator实现

---

Welcome to Code Block's blog

本篇文章主要介绍了

[小试牛刀-Turbine数据分发]
❤博主广交技术好友，喜欢文章的可以关注一下❤

一、什么是Turbine算法?

        Turbine 算法是一种用于区块链网络（尤其是 Solana 区块链）中高效数据传输和共识的算法。它的核心目标是在高吞吐量和低延迟的网络环境中，优化数据的传播和验证过程。Turbine 是 Solana 区块链的关键组件之一，帮助其实现每秒处理数万笔交易的能力。

二、流程图

       2-1 流程图

         如图2-1的所示,当客户端发送transaction到RPC节点后,会被分发到Leader节点(领导者节点),然后领导者节点对数据块进行封装后,会分发到验证节点(Validator)。当数据块数据过大时,数据的分发就会占用大量的网络带宽,造成网络短时负载增加。在区块链网络中,这种网络负载的造成的影响是巨大的,所以使用Turbine算法解决网络中的负载问题。如下图(2-2)所示:

        2-2 流程图

         当领导者分发包装的区块数据量过大时,将块(Block)拆分为小的Shred并对其进行签名然后分发到验证节点,这样验证节点对接收到得小的Shred验证并进行拼接,最终组成完整的块,然后对块内的交易数据进行验证,并最终完成整个区块的验证。

三、模拟实现

3.1 Leader实现

        为了更好的理解Turbine算法的功能实现,这里使用Python代码对其流程进行实现。为更好的模拟领导者节点和验证者节点，这里使用google的grpc依赖创建leader(server)和validator(client)进行演示。

        首先准备用户发送到链上的数据信息,这里模拟可直接对其进行定义,其定义如下:

# 接收到的交易数据
transaction = {"message_header": {"version": 1,"account_count": 2,"signature_count": 1},"account_keys": ["Alice", "Bob"],"recent_blockhash": "5KQmYg7s8v9wX2y3z4a5b6c7d8e9f0g1h2i3j4k5l6m7n8o9p0q1r2s3t4u5v6w7","instructions": [],"program_id": "TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA"
}

当用户发送该数据到链上之前应对其进行签名(模拟私钥签名),使用pynacl依赖使用私钥对数据进行签名并添加到要发送transcation中,其代码如下:

def get_signkey()->None:"""生成并返回签名密钥。"""config = load_config("config.yml")  # 加载配置文件private_key = config["private_key"]  # 获取私钥seed = base58.b58decode(private_key)[:32]  # 解码私钥并取前32字节作为种子signing_key = SigningKey(seed)  # 使用种子生成签名密钥return signing_keydef sign_transaction(transaction:dict[str,Any]) -> dict[str,Any]:"""对交易数据进行签名并返回签名后的交易数据。"""transaction_signature = get_signkey().sign(json.dumps(transaction).encode()).signature  # 对交易数据进行签名transaction.update({"signature": base58.b58encode(transaction_signature).decode()})  # 将签名添加到交易数据中return transaction

 之后对数据进行分片,即将其打包为Shred列表,为发送数据做准备,代码如下：

def create_shreds(transaction_data:str) -> list[Shred]:"""将交易数据分片并返回分片对象列表。"""shreds = []for i in range(0, len(transaction_data), SHRED_LENGTH):payload = transaction_data[i:i+SHRED_LENGTH]  # 分片数据base58_payload = base58.b58encode(payload).decode()  # 对分片数据进行 base58 编码shred = Shred(i // SHRED_LENGTH, len(transaction_data) // SHRED_LENGTH + 1, base58_payload)  # 创建分片对象signKey = get_signkey()  # 获取签名密钥shred.sign_shred(signKey)  # 对分片进行签名shreds.append(shred)return shreds

这里对数据根据SHRED_LENGTH进行分割,同时将分割Json数据转换base58编码,并对Shred进行签名,其内容格式如下(index,total,payload,signature):

Shred(0, 4, 5nZVN1neN8mfwrCcs3VAa1m5TDiaYrKP19fQ495t2ZrQJuvwh3
JHdvna11pcYbDng7B4LzjyTMXXbBps61PCPQGqVj5HJmDPZMv1Ng8k2soWZEAE
9SHxLeaEhfMQLKVMs4Q4dCezp, 2g4436FiWE61hSbNbsn35x3Po2DaYJcC9uBKEY
7NE7ZiPzTpYb4ijPVWJhLna7ZdhLxR3BhUdczDZkwQjobYaAJJ)

最后将打包好的Shred列表通过grpc服务发送到validator节点,其代码如下:

class StreamService(sync_pb2_grpc.StreamServiceServicer):"""实现 gRPC 服务的类。"""def BiStream(self, request_iterator, context):"""双向流方法。"""while True:signed_transaction = sign_transaction(transaction.copy())  # 对交易数据进行签名transaction_data = json.dumps(signed_transaction)  # 将交易数据转换为 JSON 字符串shreds = create_shreds(transaction_data)  # 创建分片对象列表for shred in shreds:response_data = json.dumps(shred.__dict__)  # 将分片对象转换为 JSON 字符串yield sync_pb2.SyncResponse(success=True, data=response_data)  # 服务器持续返回数据breakdef serve():"""创建并启动 gRPC 服务器。"""server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))sync_pb2_grpc.add_StreamServiceServicer_to_server(StreamService(), server)server.add_insecure_port("[::]:50051")server.start()server.wait_for_termination()if __name__ == "__main__":serve()

3.2 Validator实现

        当Validator连接到服务器后,会接收到服务器发送的Shred数据,因为领导者节点在发送数据时对每个Shred进行了签名,所以这里可以直接对每个Shred数据进行验证签名(该操作保证了领导者节点发送的Shred数据完整性)。然后对Shred携带的Payload数据进行组合,得到数据(块数据),然后根据用户携带的签名,对用户数据进行验证(即transaction签名)。其代码如下:

def process_responses(stub):"""处理来自 gRPC 服务器的响应。参数:stub (sync_pb2_grpc.StreamServiceStub): gRPC 客户端存根。返回:bytes: 累积的有效负载数据。"""responses = stub.BiStream(empty_request_iterator())payload = b""for response in responses:res_data = json.loads(response.data, object_hook=json_to_shred)verify = res_data.verify_shred(get_verify_key())print("验证每个shred的签名:",verify)print(f"Received from Server: {response.success}:{res_data}:{verify}")payload += base58.b58decode(res_data.payload)if res_data.index + 1 == res_data.total:breakreturn payloaddef verify_payload_signature(payload):"""验证累积的有效负载数据的签名。参数:payload (bytes): 累积的有效负载数据。抛出:nacl.exceptions.BadSignatureError: 如果签名验证失败。"""dict_data = json.loads(payload.decode('utf-8'))signature = dict_data.pop('signature')print("接收到的组合完成的数据:", dict_data)payload_verify = json.dumps(dict_data).encode()signature_bytes = base58.b58decode(signature)print("需验证的整体payload:",payload_verify)print("需验证的签名:",signature_bytes)get_verify_key().verify(payload_verify, signature_bytes)print("signature verify success")

需要完整代码可以去我的GitHub获取.

 对区块链内容感兴趣可以查看我的专栏:小试牛刀-区块链

感谢您的关注和收藏!!!!!!

​