在服务器端使用内存来存储客户端发送过来的数据

在这篇文章中，我们将在服务器端使用内存来存储客户端发送过来的数据。在实现数据存储之前，我们先在客户端使用Clap库来解析命令行参数，并封装成命令发送给服务器。

Clap解析命令行参数 在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//解析命令行参数，生成命令
 8letcmd=process_args(client_args).await?;
 9//命令编码
10cmd.encode(&mutbuf).unwrap();
11//发送命令
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}

打开一个终端，启动kv_sever。打开另一个终端执行以下命令来测试客户端：

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

服务器和客户端都正常处理了收到的请求和响应。

内存存储

我们使用dashmap crate在内存中存储数据，dashmap是一个快速的并发map。

我们先创建src/storage目录，再创建src/storage/mod.rs文件，然后在src/lib.rs文件中引入storage模块。

在src/storage/mod.rs文件中定义一个storage trait，以便于以后不同存储方式的扩展，代码如下：

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

在src/storage目录下创建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//初始化内存存储
 5letstorage=Arc::new(MemStorage::new());
 6
 7loop{
 8......
 9
10letstor=storage.clone();
11
12tokio::spawn(asyncmove{
13//使用Frame的LengthDelimitedCodec进行编解码操作
14letmutstream=Framed::new(stream,LengthDelimitedCodec::new());
15whileletSome(Ok(mutbuf))=stream.next().await{
16//对客户端发来的protobuf请求命令进行拆包
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的请求响应进行封包，然后发送给客户端。
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，打开一个终端，运行kv_server：

RUST_LOG=info cargo run --bin kv_server

2，打开一个终端，运行kv_client，执行set命令：

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

3，打开一个终端，运行kv_client，执行get命令：

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

执行结果：

INFO kv_client: Send command successed！

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

审核编辑：刘清