使用Protobuf實現(xiàn)客戶端與服務(wù)器之間的通信協(xié)議層

[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"

 1syntax="proto3";
 2
 3packagecmd;
 4
 5//命令請求
 6messageCmdRequest{
 7oneofreq_data{
 8Getget=1;
 9Setset=2;
10Publishpublish=3;
11Subscribesubscribe=4;
12Unsubscribeunsubscribe=5;
13}
14}
15
16//服務(wù)器的響應(yīng)
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發(fā)布值命令
36messagePublish{
37stringtopic=1;
38bytesvalue=2;
39}
40
41//訂閱Topic命令
42messageSubscribe{
43stringtopic=1;
44}
45
46//取消訂閱命令
47messageUnsubscribe{
48stringtopic=1;
49uint32id=2;
50}

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}

 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}

1modpb;
2pubusepb::*;

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

 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//對客戶端發(fā)來的protobuf請求命令進行拆包
15letcmd_req=CmdRequest::decode(&buf[..]).unwrap();
16info!("Receiveacommand:{:?}",cmd_req);
17
18buf.clear();
19
20//對protobuf的請求響應(yīng)進行封包，然后發(fā)送給客戶端。
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}

 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//創(chuàng)建GET命令
12letcmd_get=CmdRequest::get("mykey");
13cmd_get.encode(&mutbuf).unwrap();
14
15//發(fā)送GET命令
16stream.send(buf.freeze()).await.unwrap();
17info!("Send info successed！");
18
19//接收服務(wù)器返回的響應(yīng)
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"" }

審核編輯：湯梓紅