当前位置：首页 > news >正文

Multi-Raft 分片架构的实现细节：数据放置策略与 Leader 均衡的工程权衡

news 2026/7/9 5:55:44

Multi-Raft 分片架构的实现细节：数据放置策略与 Leader 均衡的工程权衡

一、单 Raft 组的瓶颈：5 万 QPS 写入已经达到集群上限

一个基于 Raft 的分布式 KV 存储系统。单 Raft 组的写入上限约 5 万 QPS（受限于 Leader 节点的磁盘 fsync 带宽）。当业务增长到需要 30 万 QPS 时。单 Raft 组无论如何优化都无法满足需求。必须进行分片。

Multi-Raft 架构的基本思想是将数据切分为多个 Raft Group（Shard）。每个 Shard 是一个独立的 Raft 组。独立处理写入。总写入能力等于各 Shard 写入能力的叠加。但 Multi-Raft 引入了两个新的设计挑战。数据如何在不同 Shard 之间分配。以及如何确保各 Shard 的 Leader 在物理节点间均匀分布。

这两个问题的答案决定了 Multi-Raft 集群的实际可用性。一个放置不当的分片分布可能导致某些节点成为热点。某些 Shard 的 Leader 集中在同一台机器上。导致该机器负载过高。

二、Multi-Raft 的架构设计与分片策略

Multi-Raft 的每个 Shard 是一个独立的逻辑单元。有自己的日志序列和状态机。但多个 Shard 共享同一组物理节点。

graph TD subgraph 数据分片层 A["Key Space (Hash Ring)"] --> B["Shard 0: hash ∈ [0, 2^16)"] A --> C["Shard 1: hash ∈ [2^16, 2×2^16)"] A --> D["Shard 2: hash ∈ [2×2^16, 3×2^16)"] A --> E["Shard N: hash ∈ [N×2^16, (N+1)×2^16)"] end subgraph Multi-Raft 层 B --> F["Raft Group 0"] C --> G["Raft Group 1"] D --> H["Raft Group 2"] E --> I["Raft Group N"] end subgraph 物理节点层 J["Node 1"] K["Node 2"] L["Node 3"] M["Node 4"] N["Node 5"] end F --> J F --> K F --> L G --> K G --> L G --> M H --> L H --> M H --> N subgraph Leader 分布 O["Shard 0 Leader: Node 1"] P["Shard 1 Leader: Node 3"] Q["Shard 2 Leader: Node 5"] end J -.-> O K -.-> P L -.-> Q

数据放置策略决定了 Key 到 Shard 的映射关系。常见的策略有哈希分片（Hash-based Sharding）和范围分片（Range-based Sharding）。哈希分片将 Key 的哈希值对 Shard 总数取模。分布均匀但范围查询需要跨 Shard。范围分片按 Key 的字典序分区。支持高效的范围扫描但容易产生热点。

Leader 均衡是 Multi-Raft 运维的核心关注点。每个 Shard 的 Leader 负责处理该 Shard 的所有写入。如果多个 Shard 的 Leader 集中在同一台机器上。该机器成为集群瓶颈。理想的 Leader 分布是每个节点上的 Leader 数量大致相等。差异不超过 1。

Raft 协议本身不负责 Leader 的分布。Leader 由选举产生。在无干预的情况下。Leader 可能聚集在某些节点上。需要一个外部的调度器来监控 Leader 分布并迁移不均衡的 Leader。

三、Multi-Raft 的 Leader 均衡调度器实现

use std::collections::{HashMap, HashSet}; use std::sync::Arc; use tokio::sync::RwLock; /// Shard 元数据 #[derive(Clone, Debug)] struct ShardInfo { shard_id: u64, /// Shard 的 Leader 所在节点 leader_node_id: u64, /// Shard 的所有成员节点 members: HashSet<u64>, /// Shard 的写入 QPS（最近采样周期） write_qps: f64, /// 最近一次 Leader 迁移的时间 last_transfer: std::time::Instant, } /// Leader 均衡调度器 /// /// 设计原则： /// 1. 优先迁移轻负载 Shard 的 Leader（减少迁移影响） /// 2. 两次迁移之间必须有冷却间隔（防止抖动） /// 3. 基于 QPS 而非 Leader 数量做均衡判断 struct LeaderBalancer { /// Shard 信息映射 shards: Arc<RwLock<HashMap<u64, ShardInfo>>>, /// 节点信息映射 nodes: Arc<RwLock<HashMap<u64, NodeLoad>>>, /// 允许的最大负载偏差比（最热节点/最冷节点） max_imbalance_ratio: f64, /// 两次 Leader 迁移的最小间隔 min_transfer_interval: std::time::Duration, /// 每次检查周期最多迁移的 Leader 数量 max_transfers_per_cycle: usize, } /// 节点的负载信息 #[derive(Clone, Debug, Default)] struct NodeLoad { /// 当前 Leader 的 Shard 数 leader_count: usize, /// 当前节点的写入 QPS 总量 total_write_qps: f64, } /// 建议的迁移操作 #[derive(Debug)] struct TransferSuggestion { shard_id: u64, from_node: u64, to_node: u64, /// 迁移的预估收益（QPS 偏差减少量） estimated_benefit: f64, } impl LeaderBalancer { fn new( max_imbalance_ratio: f64, min_transfer_interval_ms: u64, max_transfers_per_cycle: usize, ) -> Self { Self { shards: Arc::new(RwLock::new(HashMap::new())), nodes: Arc::new(RwLock::new(HashMap::new())), max_imbalance_ratio, min_transfer_interval: std::time::Duration::from_millis(min_transfer_interval_ms), max_transfers_per_cycle, } } /// 主调度循环 async fn schedule(&self) -> Vec<TransferSuggestion> { let shards = self.shards.read().await; let mut nodes = self.nodes.write().await; // 重置节点负载计数 for load in nodes.values_mut() { load.leader_count = 0; load.total_write_qps = 0.0; } // 统计各节点的当前负载 for shard in shards.values() { if let Some(load) = nodes.get_mut(&shard.leader_node_id) { load.leader_count += 1; load.total_write_qps += shard.write_qps; } } // 检查是否需要均衡 if !self.needs_rebalance(&nodes) { return Vec::new(); } // 生成迁移建议 let mut suggestions = Vec::new(); // 找出负载最高和最低的节点 let (hot_node, cold_node) = self.find_extreme_nodes(&nodes); if let (Some(hot_id), Some(cold_id)) = (hot_node, cold_node) { // 从热节点找出可迁移的 Shard Leader let candidates: Vec<&ShardInfo> = shards.values() .filter(|s| s.leader_node_id == hot_id) .filter(|s| s.members.contains(&cold_id)) // 目标节点必须是 Shard 成员 .filter(|s| { // 冷却时间检查：最近迁移过的不再迁移 s.last_transfer.elapsed() >= self.min_transfer_interval }) .collect(); // 按 QPS 升序排列（优先迁移轻负载 Shard） let mut sorted: Vec<&&ShardInfo> = candidates.iter().collect(); sorted.sort_by(|a, b| a.write_qps.partial_cmp(&b.write_qps).unwrap()); for shard in sorted.iter().take(self.max_transfers_per_cycle) { suggestions.push(TransferSuggestion { shard_id: shard.shard_id, from_node: hot_id, to_node: cold_id, estimated_benefit: shard.write_qps, }); } } suggestions } /// 判断是否需要负载均衡 fn needs_rebalance(&self, nodes: &HashMap<u64, NodeLoad>) -> bool { if nodes.len() < 2 { return false; } let max_load = nodes.values() .map(|n| n.total_write_qps) .fold(0.0_f64, f64::max); let min_load = nodes.values() .map(|n| n.total_write_qps) .fold(f64::MAX, f64::min); if min_load < 1.0 { // 最小负载接近零，不做均衡（避免无意义的迁移） return false; } let ratio = max_load / min_load; ratio > self.max_imbalance_ratio } /// 找到负载最高和最低的节点 fn find_extreme_nodes(&self, nodes: &HashMap<u64, NodeLoad>) -> (Option<u64>, Option<u64>) { let mut hot_id = None; let mut cold_id = None; let mut max_load = 0.0f64; let mut min_load = f64::MAX; for (node_id, load) in nodes { if load.total_write_qps > max_load { max_load = load.total_write_qps; hot_id = Some(*node_id); } if load.total_write_qps < min_load { min_load = load.total_write_qps; cold_id = Some(*node_id); } } (hot_id, cold_id) } /// 执行一次 Leader 迁移 /// /// Raft Leader 迁移的实现方式： /// 向目标节点发送一个特殊的 Admin RPC /// 触发目标节点发起选举（带更高 term 的 RequestVote） async fn execute_transfer(&self, suggestion: &TransferSuggestion) -> Result<(), String> { // 在实际实现中，这里通过 Raft 库提供的 TransferLeadership API // 将 Shard 的 Leader 从 from_node 迁移到 to_node // // 迁移过程：目标节点发起选举（增加自己的 term） // 目标节点获得多数票后成为新 Leader // 旧 Leader 转为 Follower // // 注意：迁移期间该 Shard 有短暂的不可用窗口（选举耗时） // 通常为几十到几百毫秒 let _ = suggestion; // 实际实现中的 RPC 调用 Ok(()) } }

调度器的核心设计决策是基于 QPS 而非 Leader 数量来做均衡判断。如果 Shard 0 的 QPS 是 Shard 1 的 10 倍。单看 Leader 数量不能反映真实的负载不均。均衡的目标是写入 QPS 在各节点间均匀分布。而不是 Leader 数量均匀。

迁移时优先选择轻负载的 Shard。因为迁移期间有短暂的不可用窗口。迁移轻负载 Shard 对整体可用性的影响更小。冷却间隔防止迁移抖动。如果因为一次迁移导致负载格局变化。立即再次触发迁移。系统会在多次迁移之间振荡。