在服務(wù)器端使用內(nèi)存來存儲客戶端發(fā)送過來的數(shù)據(jù)

在這篇文章中，我們將在服務(wù)器端使用內(nèi)存來存儲客戶端發(fā)送過來的數(shù)據(jù)。在實(shí)現(xiàn)數(shù)據(jù)存儲之前，我們先在客戶端使用Clap庫來解析命令行參數(shù)，并封裝成命令發(fā)送給服務(wù)器。

Clap解析命令行參數(shù) 在Cargo.toml文件中加入clap依賴：

clap = {version = "3.1", features = ["derive"]}

在src目錄下新建args.rs文件，寫入以下代碼：

 1useclap::Parser;
 2
 3#[derive(Debug,Parser)]
 4#[clap(name="kv_client")]
 5pubenumClientArgs{
 6Get{
 7#[clap(long)]
 8key:String,
 9},
10Set{
11#[clap(long)]
12key:String,
13#[clap(long)]
14value:String,
15},
16Publish{
17#[clap(long)]
18topic:String,
19#[clap(long)]
20value:String,
21},
22Subscribe{
23#[clap(long)]
24topic:String,
25},
26Unsubscribe{
27#[clap(long)]
28topic:String,
29#[clap(long)]
30id:u32,
31}
32}

在src/lib.rs中加入以下代碼：

1modargs;
2pubuseargs::*;

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

 1#[tokio::main]
 2asyncfnmain()->Result<(),?Box>{
 3......
 4
 5letclient_args=ClientArgs::parse();
 6
 7//解析命令行參數(shù)，生成命令
 8letcmd=process_args(client_args).await?;
 9//命令編碼
10cmd.encode(&mutbuf).unwrap();
11//發(fā)送命令
12stream.send(buf.freeze()).await.unwrap();
13info!("Send command successed！");
14
15loop{
16tokio::select!{
17Some(Ok(buf))=stream.next()=>{
18letcmd_res=CmdResponse::decode(&buf[..]).unwrap();
19info!("Receivearesponse:{:?}",cmd_res);
20}
21}
22}
23}

 1//生成CmdRequest命令
 2asyncfnprocess_args(client_args:ClientArgs)->Result>{
 3matchclient_args{
 4//生成GET命令
 5ClientArgs::Get{key}=>{
 6Ok(CmdRequest::get(key))
 7},
 8//生成SET命令
 9ClientArgs::Set{key,value}=>{
10Ok(CmdRequest::set(key,value.into()))
11},
12//生成PUBLISH命令
13ClientArgs::Publish{topic,value}=>{
14Ok(CmdRequest::publish(topic,value.into()))
15},
16//生成SUBSCRIBE命令
17ClientArgs::Subscribe{topic}=>{
18Ok(CmdRequest::subscribe(topic))
19},
20//生成UNSUBSCRIBE命令
21ClientArgs::Unsubscribe{topic,id}=>{
22Ok(CmdRequest::unsubscribe(topic,id))
23}
24}
25}

打開一個(gè)終端，啟動kv_sever。打開另一個(gè)終端執(zhí)行以下命令來測試客戶端：

RUST_LOG=info cargo run --bin kv_client get --key mykeyRUST_LOG=info cargo run --bin kv_client set --key mykey --value myvalue

服務(wù)器和客戶端都正常處理了收到的請求和響應(yīng)。

內(nèi)存存儲

我們使用dashmap crate在內(nèi)存中存儲數(shù)據(jù)，dashmap是一個(gè)快速的并發(fā)map。

我們先創(chuàng)建src/storage目錄，再創(chuàng)建src/storage/mod.rs文件，然后在src/lib.rs文件中引入storage模塊。

在src/storage/mod.rs文件中定義一個(gè)storage trait，以便于以后不同存儲方式的擴(kuò)展，代碼如下：

1usestd::Error;
2
3usebytes::Bytes;
4
5pubtraitStorage{
6fnget(&self,key:&str)->Result,Box>;
7fnset(&self,key:&str,value:Bytes)->Result<(),?Box>;
8}

在src/storage目錄下創(chuàng)建mem_storage.rs文件：

 1#[derive(Clone,Debug,Default)]
 2pubstructMemStorage{
 3map:DashMap
 4}
 5
 6implMemStorage{
 7pubfnnew()->Self{
 8Self{
 9map:Default::default(),
10}
11}
12}
13
14implStorageforMemStorage{
15fnget(&self,key:&str)->Result,Box>{
16Ok(self.map.get(key).map(|v|v.value().clone()))
17}
18
19fnset(&self,key:&str,value:Bytes)->Result,Box>{
20self.map.insert(key.to_string(),value.clone());
21Ok(Some(value))
22}
23}

修改kv_server.rs代碼：

 1asyncfnmain()->Result<(),?Box>{
 2......
 3
 4//初始化內(nèi)存存儲
 5letstorage=Arc::new(MemStorage::new());
 6
 7loop{
 8......
 9
10letstor=storage.clone();
11
12tokio::spawn(asyncmove{
13//使用Frame的LengthDelimitedCodec進(jìn)行編解碼操作
14letmutstream=Framed::new(stream,LengthDelimitedCodec::new());
15whileletSome(Ok(mutbuf))=stream.next().await{
16//對客戶端發(fā)來的protobuf請求命令進(jìn)行拆包
17letcmd_req=CmdRequest::decode(&buf[..]).unwrap();
18info!("Receiveacommand:{:?}",cmd_req);
19
20//處理請求命令
21letcmd_res=process_cmd(cmd_req,&stor).await.unwrap();
22
23buf.clear();
24
25//對protobuf的請求響應(yīng)進(jìn)行封包，然后發(fā)送給客戶端。
26cmd_res.encode(&mutbuf).unwrap();
27stream.send(buf.freeze()).await.unwrap();
28}
29info!("Client{:?}disconnected",addr);
30});
31}
32}
33
34//處理請求命令，返回Response
35asyncfnprocess_cmd(req:CmdRequest,storage:&MemStorage)->Result>{
36matchreq{
37//處理GET命令
38CmdRequest{
39req_data:Get(Get{key})),
40}=>{
41letvalue=storage.get(&key)?;
42Ok(CmdResponse::new(200,"getsuccess".to_string(),value.unwrap_or_default()))
43},
44//處理SET命令
45CmdRequest{
46req_data:Set(Set{key,value})),
47}=>{
48letvalue=storage.set(&key,value)?;
49Ok(CmdResponse::new(200,"setsuccess".to_string(),value.unwrap_or_default()))
50},
51_=>Err("Invalidcommand".into())
52}
53}

測試

1，打開一個(gè)終端，運(yùn)行kv_server：

RUST_LOG=info cargo run --bin kv_server

2，打開一個(gè)終端，運(yùn)行kv_client，執(zhí)行set命令：

RUST_LOG=info cargo run --bin kv_client set --key mykey --value myvalue

3，打開一個(gè)終端，運(yùn)行kv_client，執(zhí)行g(shù)et命令：

RUST_LOG=info cargo run --bin kv_client get --key mykey

執(zhí)行結(jié)果：

INFO kv_client: Send command successed！

INFO kv_client: Receive a response: CmdResponse { status: 200, message: "get success", value: b"myvalue" }

審核編輯：劉清