Rust分布式一致性库openraft的使用，高性能Raft共识算法实现与集群管理

主要特性

• 异步和事件驱动：基于Raft事件操作，不依赖周期性tick，通过消息批处理优化高吞吐量
• 可扩展的存储和网络：通过RaftLogStorage、RaftStateMachine和RaftNetwork trait可自定义存储和网络解决方案
• 统一的Raft API：提供单一的Raft类型来创建和交互Raft任务
• 集群组建：提供初始集群设置策略
• 内置追踪工具：集成tracing用于日志记录和分布式追踪

功能

• ✅ 领导者选举(通过策略或手动触发)
• ✅ 非投票者(学习者)角色
• ✅ 日志压缩(状态机快照)
• ✅ 快照复制
• ✅ 动态成员变更
• ✅ 线性一致性读
• ✅ 手动触发快照/选举
• ✅ 按策略或手动清理日志

性能

基准测试显示openraft具有出色的性能：

完整示例代码

以下是一个完整的openraft实现KV存储的示例：

use openraft::{
    Config, Raft, 
    storage::{RaftLogReader, RaftLogStorage, RaftStateMachine, Snapshot},
    network::RaftNetwork,
    AppData, AppDataResponse, NodeId,
    Entry, LogId, StateMachineChanges, SnapshotMeta,
};
use async_trait::async_trait;
use tokio::sync::RwLock;
use std::{collections::HashMap, sync::Arc};

// --- 1. 定义应用数据类型 ---
#[derive(Clone, Debug, Serialize, Deserialize)]
enum KVCommand {
    Set { key: String, value: String },
    Delete { key: String },
}

impl AppData for KVCommand {}

#[derive(Clone, Debug, Serialize, Deserialize)]
struct KVResponse {
    value: Option<String>,
}

impl AppDataResponse for KVResponse {}

// --- 2. 实现存储 ---
#[derive(Clone)]
struct KVStore {
    // Raft日志存储
    log: Arc<RwLock<Vec<u8>>>,
    // 状态机(实际KV数据)
    state_machine: Arc<RwLock<HashMap<String, String>>>,
    // 快照存储
    snapshot: Arc<RwLock<Vec<u8>>>,
}

impl KVStore {
    fn new() -> Self {
        Self {
            log: Arc::new(RwLock::new(Vec::new())),
            state_machine: Arc::new(RwLock::new(HashMap::new())),
            snapshot: Arc::new(RwLock::new(Vec::new())),
        }
    }
}

#[async_trait]
impl RaftLogStorage<NodeId> for KVStore {
    async fn get_log_entries(&self, range: std::ops::Range<u64>) -> Result<Vec<Entry<KVCommand>>, openraft::StorageError> {
        // 实现日志读取逻辑
        Ok(vec![])
    }

    async fn append_to_log(&self, entries: &[&Entry<KVCommand>]) -> Result<(), openraft::StorageError> {
        // 实现日志追加逻辑
        Ok(())
    }

    // 其他必要方法实现...
}

#[async_trait]
impl RaftStateMachine<NodeId, KVCommand, KVResponse> for KVStore {
    async fn apply_mutations(&self, mutations: Vec<StateMachineChanges<KVCommand>>) -> Result<Vec<KVResponse>, openraft::StorageError> {
        let mut res = Vec::new();
        let mut sm = self.state_machine.write().await;
        
        for change in mutations {
            for entry in change.entries {
                match entry.payload {
                    KVCommand::Set { key, value } => {
                        sm.insert(key, value);
                        res.push(KVResponse { value: None });
                    }
                    KVCommand::Delete { key } => {
                        let v = sm.remove(&key);
                        res.push(KVResponse { value: v });
                    }
                }
            }
        }
        Ok(res)
    }

    // 其他必要方法实现...
}

// --- 3. 实现网络 ---
struct Network {
    nodes: Arc<RwLock<HashMap<NodeId, Raft<NodeId, KVCommand, KVResponse>>>>,
}

#[async_trait]
impl RaftNetwork<NodeId> for Network {
    async fn send_append_entries(&self, target: NodeId, rpc: openraft::raft::AppendEntriesRequest<KVCommand>) -> Result<openraft::raft::AppendEntriesResponse, openraft::error::RPCError<NodeId>> {
        // 实现RPC调用逻辑
        Ok(openraft::raft::AppendEntriesResponse { success: true })
    }

    // 其他必要方法实现...
}

// --- 4. 主程序 ---
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // 初始化3节点集群配置
    let configs = vec![
        Config { id: 1, ..Default::default() },
        Config { id: 2, ..Default::default() },
        Config { id: 3, ..Default::default() },
    ];

    // 创建共享节点集合
    let nodes = Arc::new(RwLock::new(HashMap::new()));
    
    // 初始化所有节点
    for cfg in configs {
        let network = Network { nodes: nodes.clone() };
        let storage = KVStore::new();
        let raft = Raft::new(cfg, network, storage);
        
        let raft_clone = raft.clone();
        tokio::spawn(async move {
            raft_clone.run().await.expect("Raft节点运行失败");
        });
        
        nodes.write().await.insert(cfg.id, raft);
    }

    // 等待集群初始化
    tokio::time::sleep(std::time::Duration::from_secs(1)).await;

    // 获取leader节点
    let leader = nodes.read().await.get(&1).unwrap().clone();
    
    // 添加节点到集群
    leader.add_learner(2, true).await?;
    leader.add_learner(3, true).await?;

    // 提交数据
    leader.client_write(KVCommand::Set { 
        key: "foo".to_string(), 
        value: "bar".to_string() 
    }).await?;

    // 查询数据
    let read_result = leader.client_read().await?;
    println!("读取结果: {:?}", read_result);

    Ok(())
}

使用场景

Openraft目前被用于多个项目，包括：

• Databend的元服务集群
• CnosDB
• rrqlite
• matchengine-raft

许可证

Openraft采用MIT或Apache-2.0许可证，用户可自行选择。