持久化数据到硬盘上并采用RocksDB作为存储引擎

在上一篇文章中，我们使用内存做数据的存储。在这一篇文章中，我们持久化数据到硬盘上，采用RocksDB作为存储引擎。同时也增加了一个service层，将命令解析和存储逻辑提取到这一层中。

RocksDB 先在Cargo.toml文件中加入rocksdb依赖：

然后，在src/storage目录下创建rocksdb_storage.rs文件。代码如下：

 1#[derive(Debug)]
 2pubstructRocksDbStorage(DB);
 3
 4implRocksDbStorage{
 5pubfnnew(path:implAsRef)->Self{
 6Self(DB::open_default(path).unwrap())
 7}
 8}
 9
10implStorageforRocksDbStorage{
11fnget(&self,key:&str)->Result,Box>{
12letv=self.0.get(key)?.unwrap();
13Ok(Some(v.into()))
14}
15
16fnset(&self,key:&str,value:bytes::Bytes)->Result,Box>{
17self.0.put(key,value.clone()).unwrap();
18Ok(Some(value))
19}
20}

Service层 创建src/service目录，然后创建mod.rs文件及cmd_service.rs文件。在mod.rs文件中加入如下代码：

1pubmodcmd_service;
2
3pubtraitCmdService{
4//解析命令，返回Response
5fnexecute(self,store:&implStorage)->CmdResponse;
6}

在cmd_service.rs文件中为命令实现CmdService trait，代码如下：

 1usecrate::{CmdResponse,CmdService,Get,Set};
 2
 3//为GET实现execute
 4implCmdServiceforGet{
 5fnexecute(self,store:&implcrate::Storage)->CmdResponse{
 6//从存储中获取数据，返回CmdResponse
 7matchstore.get(&self.key){
 8Ok(Some(value))=>value.into(),
 9Ok(None)=>"Notfound".into(),
10Err(e)=>e.into(),
11}
12}
13}
14
15//为SET实现execute
16implCmdServiceforSet{
17//存储数据
18fnexecute(self,store:&implcrate::Storage)->CmdResponse{
19matchstore.set(&self.key,self.value){
20Ok(Some(value))=>value.into(),
21Ok(None)=>"Setfail".into(),
22Err(e)=>e.into(),
23}
24}
25}

在src/pb/mod.rs中实现从Bytes、&str、Box转换为CmdResponse：

 1implFromforCmdResponse{
 2fnfrom(v:Bytes)->Self{
 3Self{
 4status:200u32,
 5message:"success".to_string(),
 6value:v,
 7}
 8}
 9}
10
11implFrom<&str>forCmdResponse{
12fnfrom(s:&str)->Self{
13Self{
14status:400u32,
15message:s.to_string(),
16..Default::default()
17}
18}
19}
20
21implFrom>forCmdResponse{
22fnfrom(e:Box)->Self{
23Self{
24status:500u32,
25message:e.to_string(),
26..Default::default()
27}
28}
29}

然后在src/service/mod.rs中加入service代码：

 1//设置默认存储为RocksDB
 2pubstructService{
 3store_svc:Arc>,
 4}
 5
 6//在多线程中进行clone
 7pubstructStoreService{
 8store:Store,
 9}
10
11implService{
12pubfnnew(store:Store)->Self{
13Self{
14store_svc:Arc::new(StoreService{store}),
15}
16}
17
18//执行命令
19pubasyncfnexecute(&self,cmd_req:CmdRequest)->CmdResponse{
20println!("===ExecuteCommandBefore===");
21letcmd_res=process_cmd(cmd_req,&self.store_svc.store).await;
22println!("===ExecuteCommandAfter===");
23cmd_res
24}
25}
26
27//实现Clonetrait
28implCloneforService{
29fnclone(&self)->Self{
30Self{
31store_svc:self.store_svc.clone(),
32}
33}
34}
35
36//处理请求命令，返回Response
37asyncfnprocess_cmd(cmd_req:CmdRequest,store:&implStorage)->CmdResponse{
38matchcmd_req.req_data{
39//处理GET命令
40Some(ReqData::Get(cmd_get))=>cmd_get.execute(store),
41//处理SET命令
42Some(ReqData::Set(cmd_set))=>cmd_set.execute(store),
43_=>"Invalidcommand".into(),
44}
45}

配置 我们修改配置，在conf/server.conf中加入RocksDB路径

......[rocksdb_path]path = '/tmp/kvserver'

在src/config.rs中加入如下代码：

 1//Server端配置
 2#[derive(Debug,Serialize,Deserialize)]
 3pubstructServerConfig{
 4......
 5pubrocksdb_path:RocksdbPath,
 6}
 7
 8......
 9
10//RocksDB存储目录
11#[derive(Debug,Serialize,Deserialize)]
12pubstructRocksdbPath{
13pubpath:String,
14}

修改kv_server 在kv_server.rs中使用service执行命令，删除process_cmd函数：

 1#[tokio::main]
 2asyncfnmain()->Result<(), Box>{
 3......
 4
 5//初始化Service及存储
 6letservice=Service::new(RocksDbStorage::new(rocksdb_path));
 7
 8loop{
 9......
10letsvc=service.clone();
11
12tokio::spawn(asyncmove{
13//使用Frame的LengthDelimitedCodec进行编解码操作
14letmutstream=Framed::new(stream,LengthDelimitedCodec::new());
15whileletSome(Ok(mutbuf))=stream.next().await{
16......
17
18//执行请求命令
19letcmd_res=svc.execute(cmd_req).await;
20
21......
22}
23info!("Client{:?}disconnected",addr);
24});
25}
26}

审核编辑：刘清