如何用Rust构建一个KV Server系列

本系列是关于用Rust构建一个KV Server的系列文章，内容包括用tokio做底层异步网络通讯、使用toml文件做配置、protobuf做传输协议、内存/RockDB做数据存储、事件通知、优雅关机、并发连接限制及测量监控等。

让我们先使用tokio实现一个简单的Client & Server通讯模型，然后在此基础上逐步实现上面提及的各项内容。

创建一个新项目：

cargo new --lib kvserver_rust

在Cargo.toml文件中加入tokio依赖：

[dependencies]tokio = { version = "1.19", features = ["full"] }

Server

在src目录下创建bin文件夹，然后创建kv_server.rs文件：

use anyhow::Result;use tokio::{    io::{AsyncReadExt, AsyncWriteExt},    net::TcpListener,};
#[tokio::main]async fn main() -> Result<(), Box> {    let addr = "127.0.0.1:19999";    let listener = TcpListener::bind(addr).await?;    println!("Listening on {addr} ......");
    loop {        let (mut stream, addr) = listener.accept().await?;        println!("Client: {:?} connected", addr);
        tokio::spawn(async move {            let mut buf = vec![0u8; 1024];
            loop {                let n = stream.read(&mut buf).await.expect("从Socket读取数据失败！");
                if n == 0 {                    return;                }
                stream                    .write_all(&buf[0..n])                    .await                    .expect("向Socket写入数据失败！");            }        });    }}

在"127.0.0.1:19999"地址监听客户端的连接，收到客户端发来的信息后再返回给客户端。

Client

在src/bin目录下创建kv_client.rs文件：

use anyhow::Result;use tokio::{    io::{AsyncReadExt, AsyncWriteExt},    net::TcpStream,};
#[tokio::main]async fn main() -> Result<(), Box> {    let addr = "127.0.0.1:19999";    let mut stream = TcpStream::connect(addr).await?;
    let n = stream.write(b"Hello, world!").await?;    println!("Send info successed！n = {n}");
    let mut buf = vec![0u8; 1024];    let n = stream.read(&mut buf).await.expect("从Socket读取数据失败！");    println!("Receive info：{}, n = {n}", String::from_utf8(buf).unwrap());
    Ok(())}

连接server端"127.0.0.1:19999"这个地址，向Server端发送"Hello, world!"消息，然后再接收Server端返回的消息。

打开两个终端，分别执行：

cargo run --bin kv_servercargo run --bin kv_client

执行结果

kv_server：

Listening on 127.0.0.1:19999 ......Client: 127.0.0.1:51724 connected

kv_client：

Send info successed！n = 13Receive info：Hello, world!, n = 13

配置文件

使用 toml 做配置文件，serde 来处理配置的序列化和反序列化。在项目根目录下新建conf目录，并在下面新建server.conf文件：

[listen_address]addr = '127.0.0.1:19999'

和client.conf文件：

[connect_address]server_addr = '127.0.0.1:19999'

新建src/config.rs文件：

use std::{error::Error, fs};
use serde::{Deserialize, Serialize};
// Server端配置#[derive(Debug, Serialize, Deserialize)]pub struct ServerConfig {    pub listen_address: ListenAddress,}
// 监听地址#[derive(Debug, Serialize, Deserialize)]pub struct ListenAddress {    pub addr: String,}
// Client端配置#[derive(Debug, Serialize, Deserialize)]pub struct ClientConfig {    pub connect_address: ConnectAddress,}
// 连接地址#[derive(Debug, Serialize, Deserialize)]pub struct ConnectAddress {    pub server_addr: String,}
impl ServerConfig {    // 加载Server端配置文件    pub fn load(path: &str) -> Result> {        let config = fs::read_to_string(path)?;        let server_conf: Self = toml::from_str(&config)?;        Ok(server_conf)    }}
impl ClientConfig {    // 加载Client端配置文件    pub fn load(path: &str) -> Result> {        let config = fs::read_to_string(path)?;        let client_conf: Self = toml::from_str(&config)?;        Ok(client_conf)    }}

然后在lib.rs中加入：

mod config;pub use config::*;

修改src/bin/kv_server.rs代码：

#[tokio::main]async fn main() -> Result<(), Box> {    let server_conf = ServerConfig::load("conf/server.conf")?;    let listen_addr = server_conf.listen_address.addr;
    let listener = TcpListener::bind(&listen_addr).await?;    println!("Listening on {} ......", listen_addr);    ......}

修改src/bin/kv_client.rs代码：

#[tokio::main]async fn main() -> Result<(), Box> {    let client_conf = ClientConfig::load("conf/client.conf")?;    let connect_addr = client_conf.connect_address.server_addr;
    let mut stream = TcpStream::connect(&connect_addr).await?;    ......}

运行kv_sever和kv_client后，执行结果与上面一致。