Rust一致性哈希库pingora-ketama的使用，高性能分布式系统负载均衡解决方案
cargo add pingora-ketama
pingora-ketama = “0.6.0”
use pingora_ketama::{Ketama, Server};
fn main() {
// 创建一致性哈希环
let mut ring = Ketama::new();
// 添加服务器节点，每个节点有相应的权重
ring.add(Server::new("server1", 10));
ring.add(Server::new("server2", 20));
ring.add(Server::new("server3", 30));
// 构建哈希环
ring.build();
// 根据键查找对应的服务器
let key = "user_session_12345";
let server = ring.get(key);
println!("Key '{}' 被路由到服务器: {}", key, server.name());
// 示例：模拟多个键的路由
let keys = ["item_1", "item_2", "item_3", "item_4", "item_5"];
for key in keys.iter() {
let server = ring.get(key);
println!("Key '{}' -> 服务器: {}", key, server.name());
}
}
use pingora_ketama::{Ketama, Server};
use std::collections::HashMap;
// 分布式缓存节点结构
struct CacheNode {
name: String,
data: HashMap<String, String>,
}
impl CacheNode {
fn new(name: &str) -> Self {
Self {
name: name.to_string(),
data: HashMap::new(),
}
}
fn store(&mut self, key: &str, value: &str) {
self.data.insert(key.to_string(), value.to_string());
}
fn retrieve(&self, key: &str) -> Option<&String> {
self.data.get(key)
}
}
fn main() {
// 初始化一致性哈希环
let mut ring = Ketama::new();
// 创建缓存节点
let mut nodes = vec![
CacheNode::new("cache-node-1"),
CacheNode::new("cache-node-2"),
CacheNode::new("cache-node-3"),
CacheNode::new("cache-node-4"),
];
// 添加服务器到哈希环，设置不同的权重
for (i, node) in nodes.iter().enumerate() {
ring.add(Server::new(&node.name, (i + 1) as u32 * 10));
}
ring.build();
// 模拟数据存储
let data_items = [
("user:1001", "用户数据A"),
("user:1002", "用户数据B"),
("product:2001", "产品数据A"),
("product:2002", "产品数据B"),
("session:3001", "会话数据A"),
];
println!("=== 数据分布 ===");
for (key, value) in data_items.iter() {
let server_name = ring.get(key).name();
if let Some(node) = nodes.iter_mut().find(|n| n.name == server_name) {
node.store(key, value);
println!("存储: {} -> {} (节点: {})", key, value, server_name);
}
}
println!("\n=== 数据检索 ===");
// 测试数据检索
let test_keys = ["user:1001", "product:2002", "session:3001"];
for key in test_keys.iter() {
let server_name = ring.get(key).name();
if let Some(node) = nodes.iter().find(|n| n.name == server_name) {
if let Some(value) = node.retrieve(key) {
println!("检索: {} -> {} (节点: {})", key, value, server_name);
} else {
println!("检索: {} -> 未找到 (节点: {})", key, server_name);
}
}
}
println!("\n=== 节点负载统计 ===");
for node in nodes.iter() {
println!("节点 {}: {} 个数据项", node.name, node.data.len());
}
// 模拟节点故障（移除一个节点）
println!("\n=== 模拟节点故障 ===");
ring.remove("cache-node-2");
ring.build();
// 测试故障后数据重新分布
let key = "user:1001";
let new_server = ring.get(key).name();
println!("节点故障后，键 '{}' 被重新路由到: {}", key, new_server);
}
