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的节点。 - 节点通过网络查询和消息传递机制，从负责节点处获取存储的记录。
- 返回获取到的记录或者执行相应的处理逻辑。
- 根据 Kademlia 协议，节点首先根据
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: [] }
![【代码随想录】【算法训练营】【第56天】 [卡码98]所有可达路径](https://img-blog.csdnimg.cn/direct/a399aea782924ae4b00b0ce937c7afa4.png)
