使用Protobuf实现客户端与服务器之间的通信协议层

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

---

在上一篇文章中，我们用tokio实现了客户端和服务器的基本框架并设置了toml格式的配置文件。在这一篇文章中，我们参考Redis的命令：GET、SET、PUBLISH和SUBSCRIBE，使用Protobuf来实现客户端与服务器之间的通信协议层。为了处理Protobuf，我们加入了post库。同时加入了tracing库用于日志处理。Cargo.toml如下：

[dependencies]anyhow = "1"tokio = { version = "1.19", features = ["full"] }serde = { version = "1", features = ["derive"] }toml = "0.5"tracing = "0.1"tracing-subscriber = "0.3"bytes = "1"prost = "0.11"
[build-dependencies]prost-build = "0.11"

Protobuf在项目根目录下新建cmd.proto，加入如下代码：

 1syntax="proto3";
 2
 3packagecmd;
 4
 5//命令请求
 6messageCmdRequest{
 7oneofreq_data{
 8Getget=1;
 9Setset=2;
10Publishpublish=3;
11Subscribesubscribe=4;
12Unsubscribeunsubscribe=5;
13}
14}
15
16//服务器的响应
17messageCmdResponse{
18uint32status=1;
19stringmessage=2;
20bytesvalue=3;
21}
22
23//请求值命令
24messageGet{
25stringkey=1;
26}
27
28//存储值命令
29messageSet{
30stringkey=1;
31bytesvalue=2;
32uint32expire=3;
33}
34
35//向Topic发布值命令
36messagePublish{
37stringtopic=1;
38bytesvalue=2;
39}
40
41//订阅Topic命令
42messageSubscribe{
43stringtopic=1;
44}
45
46//取消订阅命令
47messageUnsubscribe{
48stringtopic=1;
49uint32id=2;
50}

在src目录下创建pb目录，在根目录下创建build.rs文件，加入如下代码：

1fnmain(){
2letmutconf=prost_build::new();
3conf.bytes(&["."]);
4conf.type_attribute(".","#[derive(PartialOrd)]");
5conf.out_dir("src/pb")
6.compile_protos(&["cmd.proto"],&["."])
7.unwrap();
8}

在src/pb目录下已经自动生成了cmd.rs文件。在src/pb目录下创建mod.rs文件，加入如下代码：

 1usebytes::Bytes;
 2
 3usecrate::{cmd_request::ReqData,CmdRequest,Get,Publish,Set,Subscribe,Unsubscribe};
 4
 5pubmodcmd;
 6
 7implCmdRequest{
 8//GET命令
 9pubfnget(key:implInto)->Self{
10Self{
11req_data:Some(ReqData::Get(Get{key:key.into()})),
12}
13}
14
15//SET命令
16pubfnset(key:implInto,value:Bytes,expire:u32)->Self{
17Self{
18req_data:Some(ReqData::Set(Set{
19key:key.into(),
20value,
21expire,
22})),
23}
24}
25
26//PUBLISH命令
27pubfnpublish(topic:implInto,value:Bytes)->Self{
28Self{
29req_data:Some(ReqData::Publish(Publish{
30topic:topic.into(),
31value,
32})),
33}
34}
35
36//订阅命令
37pubfnsubscribe(topic:implInto)->Self{
38Self{
39req_data:Some(ReqData::Subscribe(Subscribe{
40topic:topic.into(),
41})),
42}
43}
44
45//解除订阅命令
46pubfnunsubscribe(topic:implInto,id:u32)->Self{
47Self{
48req_data:Some(ReqData::Unsubscribe(Unsubscribe{
49topic:topic.into(),
50id,
51})),
52}
53}
54}
55
56implCmdResponse{
57pubfnnew(status:u32,message:String,value:Bytes)->Self{
58Self{
59status,
60message,
61value,
62}
63}
64}

在 src/lib.rs 中，引入pb模块：

1modpb;
2pubusepb::*;

客户端 & 服务器我们使用tokio-util库的Frame里的LengthDelimitedCodec(根据长度进行编解码)对protobuf协议进行封包解包。在Cargo.toml里加入tokio-util依赖：

[dependencies]......futures = "0.3"tokio-util = {version = "0.7", features = ["codec"]}......

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

 1#[tokio::main]
 2asyncfnmain()->Result<(),Box>{
 3tracing_subscriber::init();
 4
 5......
 6
 7loop{
 8......
 9
10tokio::spawn(asyncmove{
11//使用Frame的LengthDelimitedCodec进行编解码操作
12letmutstream=Framed::new(stream,LengthDelimitedCodec::new());
13whileletSome(Ok(mutbuf))=stream.next().await{
14//对客户端发来的protobuf请求命令进行拆包
15letcmd_req=CmdRequest::decode(&buf[..]).unwrap();
16info!("Receiveacommand:{:?}",cmd_req);
17
18buf.clear();
19
20//对protobuf的请求响应进行封包，然后发送给客户端。
21letcmd_res=CmdResponse::new(200,"success".to_string(),Bytes::default());
22cmd_res.encode(&mutbuf).unwrap();
23stream.send(buf.freeze()).await.unwrap();
24}
25info!("Client{:?}disconnected",addr);
26});
27}
28}

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

 1#[tokio::main]
 2asyncfnmain()->Result<(),Box>{
 3tracing_subscriber::init();
 4
 5......
 6
 7//使用Frame的LengthDelimitedCodec进行编解码操作
 8letmutstream=Framed::new(stream,LengthDelimitedCodec::new());
 9letmutbuf=BytesMut::new();
10
11//创建GET命令
12letcmd_get=CmdRequest::get("mykey");
13cmd_get.encode(&mutbuf).unwrap();
14
15//发送GET命令
16stream.send(buf.freeze()).await.unwrap();
17info!("Send info successed！");
18
19//接收服务器返回的响应
20whileletSome(Ok(buf))=stream.next().await{
21letcmd_res=CmdResponse::decode(&buf[..]).unwrap();
22info!("Receivearesponse:{:?}",cmd_res);
23}
24
25Ok(())
26}

我们打开二个终端，分别输入以下命令：

RUST_LOG=info cargo run --bin kv_serverRUST_LOG=info cargo run --bin kv_client

服务器执行结果：

INFO kv_server: Listening on 127.0.0.1:19999 ......INFO kv_server: Client: 127.0.0.1:50655 connectedINFO kv_server: Receive a command: CmdRequest { req_data: Some(Get(Get { key: "mykey" })) }

客户端执行结果：

INFO kv_client: Send info successed！INFO kv_client: Receive a response: CmdResponse { status: 200, message: "success", value: b"" }

服务器和客户端都正常处理了收到的请求和响应。 下一篇文章我们将在服务器端使用内存来存储客户端发送过来的数据。 完整代码：https://github.com/Justin02180218/kv_server_rust

---

审核编辑：汤梓红