p2p、分布式,区块链笔记: 通过libp2p的Kademlia网络协议实现kv-store

news2024/11/19 15:21:27

Kademlia 网络协议

  • Kademlia 是一种分布式哈希表协议和算法,用于构建去中心化的对等网络,核心思想是通过分布式的网络结构来实现高效的数据查找和存储。在这个学习项目里,Kademlia 作为 libp2p 中的 NetworkBehaviour的组成。

  • 以下这些函数或方法是根据 Kademlia 网络协议设计的,它们实现了基本的网络操作,包括获取数据记录、获取数据提供者、存储数据记录和开始提供数据等功能(这里只展示了项目中用到的函数,常用函数可以看libp2p Kademlia DHT 规范,更多函数可见如下图中的源码部分)。

在这里插入图片描述

1. get_record

kademlia.get_record(key, Quorum::One);
  • 作用: 从 Kademlia 网络中获取与指定 key 相关的记录。
  • 参数:
    • key: 要获取记录的键。
    • Quorum::One: 获取记录时所需的一致性要求,这里是指只需要从一个节点获取记录即可。
  • 实现逻辑:
    • 根据 Kademlia 协议,节点首先根据 key 计算出其对应的 K-bucket 或者具体的节点 ID,然后向网络中查找负责该 key 的节点。
    • 节点通过网络查询和消息传递机制,从负责节点处获取存储的记录。
    • 返回获取到的记录或者执行相应的处理逻辑。

2. get_providers

kademlia.get_providers(key);
  • 作用: 获取能够提供与指定 key 相关数据的节点信息(即数据的提供者)。
  • 参数:
    • key: 要获取提供者信息的数据的键。
  • 实现逻辑:
    • 类似于 get_record,节点根据 key 计算出其对应的 K-bucket 或者节点 ID。
    • 节点向网络发送查询请求,询问哪些节点能够提供与 key 相关的数据。
    • 返回能够提供数据的节点列表或者执行相应的处理逻辑。

3. put_record

let record = Record {
    key,
    value,
    publisher: None,
    expires: None,
};
kademlia.put_record(record, Quorum::One).expect("Failed to store record locally.");
  • 作用: 将指定的记录存储到 Kademlia 网络中。
  • 参数:
    • record: 包含要存储的数据信息的记录对象,包括 key(键)、value(值)、publisher(发布者,可能为空)、expires(过期时间,可能为空)等字段。
    • Quorum::One: 存储记录时的一致性要求,这里是指只需要将记录存储在一个节点即可。
  • 实现逻辑:
    • 节点根据 key 计算出对应的 K-bucket 或节点 ID。
    • 节点将 record 发送给负责存储该 key 的节点,并根据指定的一致性要求存储副本。
    • 返回存储成功或失败的结果,或者执行相应的处理逻辑。

4. start_providing

kademlia.start_providing(key).expect("Failed to start providing key");
  • 作用: 在 Kademlia 网络中开始提供指定 key 的数据。
  • 参数:
    • key: 要开始提供的数据的键。
  • 实现逻辑:
    • 节点将 key 注册为它可以提供的数据标识。
    • 当其他节点查询或需要该 key 的数据时,该节点将响应并提供相应的数据。
    • 返回启动提供成功或失败的结果,或者执行相应的处理逻辑。

kv数据库主体代码及注释

use async_std::io;
use futures::{prelude::*, select};
use libp2p::kad::record::store::MemoryStore;
use libp2p::kad::{
    record::Key, AddProviderOk, Kademlia, KademliaEvent, PeerRecord, PutRecordOk, QueryResult,
    Quorum, Record,
};
use libp2p::{
    development_transport, identity,
    mdns::{Mdns, MdnsConfig, MdnsEvent},
    swarm::SwarmEvent,
    NetworkBehaviour, PeerId, Swarm,
};
use std::error::Error;

#[async_std::main]
async fn main() -> Result<(), Box<dyn Error>> {
    env_logger::init();

    //  创建本地密钥,本地peer id和传输控制组件
    let local_key = identity::Keypair::generate_ed25519();
    let local_peer_id = PeerId::from(local_key.public());
    let transport = development_transport(local_key).await?;

    // 事件行为控制
    // We create a custom network behaviour that combines Kademlia and mDNS.
    #[derive(NetworkBehaviour)]// https://docs.rs/libp2p/latest/libp2p/swarm/trait.NetworkBehaviour.html
    #[behaviour(out_event = "MyBehaviourEvent")]//这个 "MyBehaviourEvent" 定义在下边的代码中
    // NetworkBehaviour这个trait将对所描述的结构体中的每个成员依次进行操作,例如 NetworkBehavior::poll它将首先轮询第一个结构成员,直到返回poll::Pending,然后再转到后面的成员。
    // 关于 #[behaviour(out_event = "MyBehaviourEvent")]中的out_event :The final out event. If we find a `#[behaviour(out_event = "Foo")]` attribute on the struct, we set `Foo` as the out event. Otherwise we use `()`.
    struct MyBehaviour {
        kademlia: Kademlia<MemoryStore>,
        mdns: Mdns,
    }


    #[allow(clippy::large_enum_variant)] //  #[allow()为Lint语法属性检查控制,https://doc.rust-lang.org/reference/attributes/diagnostics.html#lint-check-attributes    //关于large_enum_variant 详见https://rust-lang.github.io/rust-clippy/master/index.html#/large_enum_variant
    enum MyBehaviourEvent {
        Kademlia(KademliaEvent),
        Mdns(MdnsEvent),
    }
    // 实现(impl)块,用于为类型KademliaEvent实现了From trait,使其能够被转换为类型MyBehaviourEvent。
    impl From<KademliaEvent> for MyBehaviourEvent {
        fn from(event: KademliaEvent) -> Self {
            MyBehaviourEvent::Kademlia(event)
        }
    }
    // 实现(impl)块,用于为类型  MdnsEvent   实现了From trait,使其能够被转换为类型MyBehaviourEvent。
    impl From<MdnsEvent> for MyBehaviourEvent {
        fn from(event: MdnsEvent) -> Self {
            MyBehaviourEvent::Mdns(event)
        }
    }

    // Create a swarm to manage peers and events.
    let mut swarm = {
        // Create a Kademlia behaviour.
        let store = MemoryStore::new(local_peer_id);
        let kademlia = Kademlia::new(local_peer_id, store);
        let mdns = Mdns::new(MdnsConfig::default()).await?;
        let behaviour = MyBehaviour { kademlia, mdns };
        Swarm::new(transport, behaviour, local_peer_id)
    };

    // 从命令行读取指令并赋值给可变变量"stdin"
    let mut stdin = io::BufReader::new(io::stdin()).lines().fuse();

    // Listen on all interfaces and whatever port the OS assigns.
    swarm.listen_on("/ip4/0.0.0.0/tcp/0".parse()?)?;

    // Kick it off.
    loop {
        select! {
            line = stdin.select_next_some() => handle_input_line(&mut swarm.behaviour_mut().kademlia, line.expect("Stdin not to close")),
            event = swarm.select_next_some() => match event { // swarm.select_next_some() 是一个方法,用于从一个事件流中获取下一个事件,后续送到match进行匹配
                SwarmEvent::NewListenAddr { address, .. } => {//当发生新的监听地址事件时
                    println!("Listening in {:?}", address);
                },
                SwarmEvent::Behaviour(MyBehaviourEvent::Mdns(MdnsEvent::Discovered(list))) => {// 发生mDNS服务发现事件时
                    for (peer_id, multiaddr) in list {
                        swarm.behaviour_mut().kademlia.add_address(&peer_id, multiaddr);
                    }
                }
                SwarmEvent::Behaviour(MyBehaviourEvent::Kademlia(KademliaEvent::OutboundQueryCompleted { result, ..})) => {// 当发出的 Kademlia 查询完成时
                    handle_query_result(&result);
                }
                _ => {} // 通配符模式,执行一个空的代码块
            }
        }
    }
}

// 下面是两个辅助函数,一个根据不同的查询结果类型执行不同的逻辑,另一个处理从命令行输入的命令
fn handle_query_result(result: &QueryResult) {
    match result {
     ...
    }
}

fn handle_input_line(kademlia: &mut Kademlia<MemoryStore>, line: String) {
    let mut args = line.split(' ');

    match args.next() {
    ...
    }
}

两个辅助函数

处理从命令行输入的命令

  • 这段 Rust 代码定义了一个函数 handle_input_line,用于处理从命令行读取的输入 line,并根据命令执行相应的操作。函数通过分割输入行来解析命令和参数,处理缺少参数的错误情况,并根据命令调用传入的 Kademlia 网络实例 (kademlia) 的相应方法。
fn handle_input_line(kademlia: &mut Kademlia<MemoryStore>, line: String) {
    // 将输入行按空格分割为多个参数
    let mut args = line.split(' ');

    // 匹配第一个参数(命令)
    match args.next() {
        Some("GET") => {
            // 如果命令是 "GET"
            let key = {
                // 尝试获取下一个参数作为键
                match args.next() {
                    Some(key) => Key::new(&key), // 从字符串创建 Key 对象
                    None => {
                        // 如果未提供键,则打印错误并从函数返回
                        eprintln!("缺少键");
                        return;
                    }
                }
            };
            // 调用 Kademlia 网络的 get_record 方法,传入指定的键和 Quorum::One
            kademlia.get_record(key, Quorum::One);
        }
        Some("GET_PROVIDERS") => {
            // 如果命令是 "GET_PROVIDERS"
            let key = {
                // 尝试获取下一个参数作为键
                match args.next() {
                    Some(key) => Key::new(&key), // 从字符串创建 Key 对象
                    None => {
                        // 如果未提供键,则打印错误并从函数返回
                        eprintln!("缺少键");
                        return;
                    }
                }
            };
            // 调用 Kademlia 网络的 get_providers 方法,传入指定的键
            kademlia.get_providers(key);
        }
        Some("PUT") => {
            // 如果命令是 "PUT"
            let key = {
                // 尝试获取下一个参数作为键
                match args.next() {
                    Some(key) => Key::new(&key), // 从字符串创建 Key 对象
                    None => {
                        // 如果未提供键,则打印错误并从函数返回
                        eprintln!("缺少键");
                        return;
                    }
                }
            };
            let value = {
                // 尝试获取下一个参数作为值
                match args.next() {
                    Some(value) => value.as_bytes().to_vec(), // 将值转换为字节向量
                    None => {
                        // 如果未提供值,则打印错误并从函数返回
                        eprintln!("缺少值");
                        return;
                    }
                }
            };
            // 创建一个包含指定键、值及可选字段的 Record 对象
            let record = Record {
                key,
                value,
                publisher: None,
                expires: None,
            };
            // 在 Kademlia 网络中以 Quorum::One 一致性存储记录
            kademlia
                .put_record(record, Quorum::One)
                .expect("本地存储记录失败。");
        }
        Some("PUT_PROVIDER") => {
            // 如果命令是 "PUT_PROVIDER"
            let key = {
                // 尝试获取下一个参数作为键
                match args.next() {
                    Some(key) => Key::new(&key), // 从字符串创建 Key 对象
                    None => {
                        // 如果未提供键,则打印错误并从函数返回
                        eprintln!("缺少键");
                        return;
                    }
                }
            };
            // 在 Kademlia 网络中开始提供指定的键
            kademlia
                .start_providing(key)
                .expect("启动提供键失败");
        }
        _ => {
            // 如果命令不匹配预期的任何命令
            eprintln!("期望命令为 GET、GET_PROVIDERS、PUT 或 PUT_PROVIDER");
        }
    }

根据不同的查询结果类型执行不同的逻辑

fn handle_query_result(result: &QueryResult) {
    match result {
        QueryResult::GetProviders(Ok(ok)) => {
            for peer in &ok.providers {
                println!(
                    "Peer {:?} provides key {:?}",
                    peer,
                    std::str::from_utf8(ok.key.as_ref()).unwrap()
                );
            }
        }
        QueryResult::GetProviders(Err(err)) => {
            eprintln!("Failed to get providers: {:?}", err);
        }
        QueryResult::GetRecord(Ok(ok)) => {
            for PeerRecord {
                record: Record { key, value, .. },
                ..
            } in &ok.records
            {
                println!(
                    "Got record {:?} {:?}",
                    std::str::from_utf8(key.as_ref()).unwrap(),
                    std::str::from_utf8(&value).unwrap(),
                );
            }
        }
        QueryResult::GetRecord(Err(err)) => {
            eprintln!("Failed to get record: {:?}", err);
        }
        QueryResult::PutRecord(Ok(PutRecordOk { key })) => {
            println!(
                "Successfully put record {:?}",
                std::str::from_utf8(key.as_ref()).unwrap()
            );
        }
        QueryResult::PutRecord(Err(err)) => {
            eprintln!("Failed to put record: {:?}", err);
        }
        QueryResult::StartProviding(Ok(AddProviderOk { key })) => {
            println!(
                "Successfully put provider record {:?}",
                std::str::from_utf8(key.as_ref()).unwrap()
            );
        }
        QueryResult::StartProviding(Err(err)) => {
            eprintln!("Failed to put provider record: {:?}", err);
        }
        _ => {}
    }
}

fn handle_input_line(kademlia: &mut Kademlia<MemoryStore>, line: String) {
    let mut args = line.split(' ');

    match args.next() {
        Some("GET") => {
            let key = {
                match args.next() {
                    Some(key) => Key::new(&key),
                    None => {
                        eprintln!("Expected key");
                        return;
                    }
                }
            };
            kademlia.get_record(key, Quorum::One);
        }
        Some("GET_PROVIDERS") => {
            let key = {
                match args.next() {
                    Some(key) => Key::new(&key),
                    None => {
                        eprintln!("Expected key");
                        return;
                    }
                }
            };
            kademlia.get_providers(key);
        }
        Some("PUT") => {
            let key = {
                match args.next() {
                    Some(key) => Key::new(&key),
                    None => {
                        eprintln!("Expected key");
                        return;
                    }
                }
            };
            let value = {
                match args.next() {
                    Some(value) => value.as_bytes().to_vec(),
                    None => {
                        eprintln!("Expected value");
                        return;
                    }
                }
            };
            let record = Record {
                key,
                value,
                publisher: None,
                expires: None,
            };
            kademlia
                .put_record(record, Quorum::One)
                .expect("Failed to store record locally.");
        }
        Some("PUT_PROVIDER") => {
            let key = {
                match args.next() {
                    Some(key) => Key::new(&key),
                    None => {
                        eprintln!("Expected key");
                        return;
                    }
                }
            };

            kademlia
                .start_providing(key)
                .expect("Failed to start providing key");
        }
        _ => {
            eprintln!("expected GET, GET_PROVIDERS, PUT or PUT_PROVIDER");
        }
    }
}

运行示例

PS C:\Users\kingchuxing\Documents\learning-libp2p-main\rust> cargo run --example 04-kv-store
Listening in "/ip4/172.23.118.182/tcp/65055"
Listening in "/ip4/192.168.0.104/tcp/65055"
Listening in "/ip4/127.0.0.1/tcp/65055"
GET 123
Failed to get record: NotFound { key: Key(b"123"), closest_peers: [] }
PUT 123
缺少值
PUT 123 123456789
Failed to put record: QuorumFailed { key: Key(b"123"), success: [], quorum: 1 }
GET 123     
Got record "123" "123456789"
PUT_PROVIDER 234 //输入提供者
Successfully put provider record "234"
GET_PROVIDERS 234 //获取提供者
Peer PeerId("12D3KooWB7CFnrmeH5gzRxA4CYR2YTg2K3NMvNHP5dWDPFwAHY38") provides key "234"
GET 234
Failed to get record: NotFound { key: Key(b"234"), closest_peers: [] }

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.coloradmin.cn/o/1890497.html

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈,一经查实,立即删除!

相关文章

controller不同的后端路径对应vue前端传递数据发送请求的方式,vue请求参数 param 与data 如何对应后端参数

目录 案例一&#xff1a; 为什么使用post发送请求&#xff0c;参数依旧会被拼接带url上呢&#xff1f;这应该就是param 与data传参的区别。即param传参数参数会被拼接到url后&#xff0c;data会以请求体传递 补充&#xff1a;后端controller 参数上如果没写任何注解&#xff0c…

Redis中hash类型的操作命令(命令的语法、返回值、时间复杂度、注意事项、操作演示)

文章目录 字符串和哈希类型相比hset 命令hget 命令hexistshdelhkeyshvalshgetallhmgethlenhsetnxhincrbyhincrbyfloat 字符串和哈希类型相比 假设有以下一种场景&#xff1a;现在要在 Redis 中存储一个用户的基本信息(id1、namezhangsan、age17)&#xff0c;下图表示使用字符串…

Vue3轻松创建交互式仪表盘

本文由ScriptEcho平台提供技术支持 项目地址&#xff1a;传送门 基于 Plotly.js 的 Vue 仪表盘组件 应用场景介绍 仪表盘是一种交互式可视化工具&#xff0c;用于监控和分析关键指标。它广泛应用于各种行业&#xff0c;例如金融、医疗保健和制造业。 代码基本功能介绍 本…

Linux源码阅读笔记12-RCU案例分析

在之前的文章中我们已经了解了RCU机制的原理和Linux的内核源码&#xff0c;这里我们要根据RCU机制写一个demo来展示他应该如何使用。 RCU机制的原理 RCU&#xff08;全称为Read-Copy-Update&#xff09;,它记录所有指向共享数据的指针的使用者&#xff0c;当要修改构想数据时&…

搭建论坛和mysql数据库安装和php安装

目录 概念 步骤 安装mysql8.0.30 安装php 安装Discuz 概念 搭建论坛的架构&#xff1a; lnmpDISCUZ l 表示linux操作系统 n 表示nginx前端页面的web服务 m 表示 mysql 数据库 用来保存用户和密码以及论坛的相关内容 p 表示php 动态请求转发的中间件 步骤 &#xff…

基于Cardinal的AWD攻防平台搭建与使用以及基于docker的题目环境部署

关于 CTF 靶场的搭建与完善勇师傅前面已经总结过了&#xff0c;参考&#xff1a; CTF靶场搭建及Web赛题制作与终端docker环境部署_ctfoj搭建-CSDN博客 基于H1ve一分钟搭好CTF靶场-CSDN博客 Nginx首页修改及使用Nginx实现端口转发_nginx 修改欢迎首页-CSDN博客 关于H1ve导…

《IT 领域准新生暑期预习指南:开启未来科技之旅》

IT专业入门&#xff0c;高考假期预习指南 高考的落幕&#xff0c;只是人生长途中的一个逗号&#xff0c;对于心怀 IT 梦想的少年们&#xff0c;新的征程已然在脚下铺展。这个七月&#xff0c;当分数尘埃落定&#xff0c;你们即将迈向新的知识殿堂&#xff0c;而这个假期&#…

235、二叉搜索树的最近公共祖先

给定一个二叉搜索树, 找到该树中两个指定节点的最近公共祖先。 百度百科中最近公共祖先的定义为&#xff1a;“对于有根树 T 的两个结点 p、q&#xff0c;最近公共祖先表示为一个结点 x&#xff0c;满足 x 是 p、q 的祖先且 x 的深度尽可能大&#xff08;一个节点也可以是它自…

代码随想录第42天|动态规划

198.打家劫舍 参考 dp[j] 表示偷盗的总金额, j 表示前 j 间房(包括j)的总偷盗金额初始化: dp[0] 一定要偷, dp[1] 则取房间0,1的最大值遍历顺序: 从小到大 class Solution { public:int rob(vector<int>& nums) {if (nums.size() < 2) {return nums[0];}vector&…

Docker安装PostgreSQL详细教程

本章教程,使用Docker安装PostgreSQL具体步骤。 一、拉取镜像 docker pull postgres二、启动容器 docker run -it --name postgres --restart always -e POSTGRES_PASSWORD=123456 -e

VideoPrism——探索视频分析领域模型的算法与应用

概述 论文地址:https://arxiv.org/pdf/2402.13217.pdf 视频是我们观察世界的生动窗口&#xff0c;记录了从日常瞬间到科学探索的各种体验。在这个数字时代&#xff0c;视频基础模型&#xff08;ViFM&#xff09;有可能分析如此海量的信息并提取新的见解。迄今为止&#xff0c;…

全国数学建模大赛(一)

全国数学建模大赛 &#x1f388;1.数学模型是什么&#xff1f;&#x1f52d;1.1原型与模型&#x1f52d;1.2模型的分类&#x1f52d;1.3数学模型的分类&#x1f52d;1.4数学模型的全过程&#x1f52d;1.5论文写作基本流程&#x1f52d;1.6数学建模的六个步骤&#x1f52d;1.7小…

【SpringBoot配置文件读取】无法读取yaml文件中文字符

1. yaml配置文件 注意要将该文件编码格式改为UTF-8 spring:application:name: 好好学习admin:name: 李斯age: 24books:- name: 数据结构desc: 数据书- name: 编译原理desc: 编译书2.配置实体类 Data设置get&#xff0c;set方法Component注册为BeanConfigurationProperties(p…

第6章:结构化开发方法

第6章&#xff1a;结构化开发方法 系统设计基本原理 1、抽象 抽象是一种设计技术&#xff0c;重点说明一个实体的本质方面&#xff0c;而忽略或者掩盖不是很重要或非本质的方面。 模块化 模块化是指将一个待开发的软件分解成若干个小的、简单的部分一模块&#xff0c;每个模…

reggie外卖优化

文章目录 一、redis缓存1.1 缓存验证码1.2 缓存菜品数据 二、spring-cache 一、redis缓存 1.1 缓存验证码 不用sesiion&#xff0c;而使用redis来存放验证码。 首先在用户请求验证码&#xff0c;将验证码保存在sesion中&#xff0c;当登录成功之后&#xff0c;将redis中的验证…

人工智能概论 | 基于A*算法的8数码问题求解

大学四年的全部课程和考试都已经结束啦&#xff01; 最近闲来无事&#xff0c;随便发发自己的实验报告&#xff0c;供后面的学弟学妹们参考~ 目录 实验1 基于A*算法的8数码问题求解 1.1 程序总体流程 1.2 关键代码展示 1.3 输出结果展示及分析 1.3.1 总步数展示 1.…

ENVI5.6使用笔记

目录 1. ENVI安装扩展2. ENVI绘制高光谱3D数据立体图3. 对本次工作存档&#xff0c;下次打开软件可直接续档4. 对图像进行分类 1. ENVI安装扩展 从ENVI App Store下载商店envi_app_store.zip&#xff0c;解压得到ENVI_App_Store.sav&#xff0c;将其复制到ENVI的扩展文件夹下&…

如何优化网站SEO排名?

选择那些容易排名的关键词。使用工具找到那些竞争少但有流量的词语。其次&#xff0c;内部链接非常重要。通过合理的内部链接&#xff0c;可以提升各个页面的权重。 增加FAQ部分能帮助你捕捉更多的长尾关键词流量。争取出现在精选摘要的位置&#xff0c;可以直接提升你的曝光率…

从全连接到卷积

一、全连接到卷积 1、卷积具有两个原则&#xff1a; 平移不变性&#xff1a;无论作用在哪个部分&#xff0c;它都要有相同的作用&#xff0c;而不会随着位置的改变而改变 局部性&#xff1a;卷积核作用处&#xff0c;作用域应该是核作用点的周围一小部分而不作用于更大的部分 …