Rust中使用Protobuf
对于Rust语言来说，有quick-protobuf、rust-protobuf、prost等第三方crates可编译Protobuf文件。
目前最流行的是prost，但它需要配合使用prost-build这个crate来进行编译转换。此外，如果protobuf要配合tonic gRPC使用，则可以替换prost-build为tonic-build来编译转换为适配tonic的结构。
以配合tonic使用为例。代码结构：
$ cargo new proto_usage
$ mkdir proto_usage/protos
$ touch proto_usage/protos/voting.proto
$ touch proto_usage/build.rs
$ tree proto_usage
proto_usage
├── Cargo.toml
├── build.rs
├── protos
│ └── voting.proto
└── src
└── main.rs
Cargo.toml的内容：
[dependencies]
prost = "0.11.0"
tokio = { version = "1.21.0", features = ["macros", "rt-multi-thread"] }
tonic = "0.8.1"
[build-dependencies]
tonic-build = "0.8.0"
voting.proto文件内容：
syntax = "proto3";
package voting;
service Voting {
rpc Vote(VotingRequest) returns (VotingResponse);
}
message VotingRequest {
string url = 1;
enum Vote {
UP = 0;
DOWN = 1;
}
Vote vote = 2;
}
message VotingResponse { string confirmation = 1; }
build.rs文件中编写在编译期间编译proto文件的代码：
use std::io::Result; fn main() -> Result<()> { tonic_build::compile_protos("protos/voting.proto")?; Ok(()) }
在编译期间，build.rs将会编译voting.proto文件并转换为voting.rs文件，得到voting.rs文件后，就可以在需要的地方导入该文件，导入该文件后就会拥有该文件中定义的数据结构等内容。至于voting.proto文件编译转换后得到的voting.rs文件中到底是什么内容，见后文分析。
例如，在main.rs中导入该文件。src/main.rs文件的内容(此处示例仅导入但并没有使用)：
pub mod voting { tonic::include_proto!("voting"); } fn main() {}
默认情况下，build.rs编译得到的中间文件保存在OUT_DIR环境变量指定的目录中，如果没有明确设置OUT_DIR环境变量，则默认为cargo的构建目录。例如<target_dir>\debug\build\proto_usage-4e6f1dc6a899518f\out\voting.rs。如果没有修改过OUT_DIR环境变量，则可以通过tonic::include_proto!("voting")宏直接导入build.rs编译后得到的voting.rs文件。
如果修改过OUT_DIR环境变量的值，或者tonic_build编译时明确指定了输出路径(稍后给代码)，则不能使用tonic::include_proto!宏来导入voting.rs。此时应该用Rust自身提供的宏include!来明确指定导入的路径。参考官方手册说明https://docs.rs/tonic/latest/tonic/macro.include_proto.html。
例如，假设build.rs的输出路径为proto_usage/protos目录：
#![allow(unused)] fn main() { pub mod voting { // tonic::include_proto!("voting"); include!("../protos/voting.rs"); } }
tonic_build提供了方法来修改proto编译后.rs文件的保存路径(注意，修改路径之后需使用include!宏来导入.rs文件)：
use std::io::Result; fn main() -> Result<()> { //tonic_build::compile_protos("protos/voting.proto")?; tonic_build::configure() .build_server(true) // 是否编译生成用于服务端的代码 .build_client(true) // 是否编译生成用于客户端的代码 .out_dir("protos") // 输出的路径，此处指定为项目根目录下的protos目录 // 指定要编译的proto文件路径列表，第二个参数是提供protobuf的扩展路径， // 因为protobuf官方提供了一些扩展功能，自己也可能会写一些扩展功能， // 如存在，则指定扩展文件路径，如果没有，则指定为proto文件所在目录即可 .compile(&["protos/voting.proto"], &["protos"])?; Ok(()) }
Protobuf编译成什么样的Rust代码
对于前文的voting.proto文件：
syntax = "proto3";
package voting;
service Voting {
rpc Vote(VotingRequest) returns (VotingResponse);
}
message VotingRequest {
string url = 1;
enum Vote {
UP = 0;
DOWN = 1;
}
Vote vote = 2;
}
message VotingResponse { string confirmation = 1; }
将其编译后得到voting.rs文件，该文件内容并不复杂，可以直接打开看。
更直观的方式在导入它之后，
#![allow(unused)] fn main() { pub mod voting { include!("../protos/voting.rs"); } }
通过文档来了解它包含了什么：
$ cargo doc --open
可以看到，在voting模块下，拥有如下内容：
#![allow(unused)] fn main() { voting::{ // 模块部分以及模块中的结构 // 为客户端生成的代码 voting_client::{self, VotingClient}, // 为服务端生成的代码 // Voting是Trait voting_server::{self, VotingServer, Voting}, // 嵌套在VotingRequest中的Vote类型 voting_request::{self, Vote}, // Structs部分 VotingRequest, VotingResponse } }
voting_request、VotingRequest和VotingResponse结构：
#![allow(unused)] fn main() { // voting_request中包含了VotingRequest内嵌的enum类型 pub mod voting_request { #[derive(Clone, Copy, Debug, ...)] #[repr(i32)] pub enum Vote { Up = 0, Down = 1, } } // 请求类型 #[derive(Clone, PartialEq, ::prost::Message)] pub struct VotingRequest { pub url: String, pub vote: i32, } // 内嵌的enum Vote类型和VotingRequest vote字段之间的转换设置 impl VotingRequest { pub fn vote(&self) -> Vote pub fn set_vote(&mut self, value: Vote) } // 响应类型 #[derive(Clone, PartialEq, ::prost::Message)] pub struct VotingResponse { pub confirmation: String, } }
这些字段都是pub公开的，因此可以直接构建这些类型或直接进行字段赋值。
为客户端生成的voting_client::VotingClient类型，有几个方法需要注意：
#![allow(unused)] fn main() { pub fn new(inner: T) -> Self // 连接服务端并返回VotingClient pub async fn connect<D>(dst: D) -> Result<Self, Error> // 在proto文件中通过rpc定义的方法vote，客户端和服务端都有 // 在客户端，它会向服务端发送请求 pub async fn vote( &mut self, request: impl IntoRequest<VotingRequest> ) -> Result<Response<VotingResponse>, Status> }
另外注意，VotingClient实现了Clone、Sync以及Send，因此可以跨线程使用。
例如：
#[tokio::main] async fn main() -> Result<(), Box<dyn std::error::Error>> { let mut client = VotingClient::connect("http://127.0.0.1:8080").await?; loop { ...... let url = ...; let vote = ...; // 构建请求 let request = tonic::Request::new(VotingRequest {url, vote}); // 发送请求并等待响应 let response = client.vote(request).await?; println!("Got: '{}' from service", response.get_ref().confirmation); } Ok(()) }
为服务端生成的voting_server::VotingServer：
#![allow(unused)] fn main() { impl<T: Voting> VotingServer<T> { pub fn new(inner: T) -> Self pub fn from_arc(inner: Arc<T>) -> Self } }
其中T: Voting中的Trait Voteing要求实现vote()方法，该方法用于定义服务端接收到客户端请求时要如何处理的逻辑。
例如：
#[derive(Debug, Default)] pub struct VotingService {} #[tonic::async_trait] impl Voting for VotingService { async fn vote( &self, request: Request<VotingRequest>, ) -> Result<Response<VotingResponse>, Status> { let r: &VotingRequest = request.get_ref(); match r.vote { 0 => Ok(Response::new(voting::VotingResponse { confirmation: { format!("upvoted for {}", r.url) }, })), 1 => Ok(Response::new(voting::VotingResponse { confirmation: { format!("downvoted for {}", r.url) }, })), _ => Err(Status::new( tonic::Code::OutOfRange, "Invalid vote provided", )), } } } #[tokio::main] async fn main() { let voting_service = VotingService::default(); let server = VotingServer::new(voting_service); }
Rust中使用tonic和Protobuf
将上面的代码综合起来，实现一个通过gRPC通信的客户端和服务端。
$ cargo new tonic_grpc
$ mkdir tonic_grpc/protos
$ touch tonic_grpc/protos/voting.proto
$ touch tonic_grpc/build.rs
$ touch tonic_grpc/src/{client.rs, server.rs}
$ tree tonic_grpc
tonic_grpc/
├── Cargo.toml
├── build.rs
├── protos
│ └── voting.proto
└── src
├── client.rs
├── main.rs
└── server.rs
Cargo.toml内容：
[dependencies]
prost = "0.11.0"
tokio = { version = "1.21.0", features = ["macros", "rt-multi-thread"] }
tonic = "0.8.1"
[build-dependencies]
tonic-build = "0.8.0"
[[bin]]
name = "server"
path = "src/server.rs"
[[bin]]
name = "client"
path = "src/client.rs"
protos/voting.proto内容：
syntax = "proto3";
package voting;
service Voting {
rpc Vote(VotingRequest) returns (VotingResponse);
}
message VotingRequest {
string url = 1;
enum Vote {
UP = 0;
DOWN = 1;
}
Vote vote = 2;
}
message VotingResponse { string confirmation = 1; }
build.rs内容：
use std::io::Result; fn main() -> Result<()> { // tonic_build::compile_protos("protos/voting.proto")?; tonic_build::configure() .build_server(true) .build_client(true) .out_dir("protos") .compile(&["protos/voting.proto"], &["protos"])?; Ok(()) }
src/server.rs内容：
use tonic::{transport::Server, Request, Response, Status}; use voting::{ voting_server::{Voting, VotingServer}, VotingRequest, VotingResponse, }; pub mod voting { // tonic::include_proto!("voting"); include!("../protos/voting.rs"); } #[derive(Debug, Default)] pub struct VotingService {} #[tonic::async_trait] impl Voting for VotingService { async fn vote( &self, request: Request<VotingRequest>, ) -> Result<Response<VotingResponse>, Status> { let r: &VotingRequest = request.get_ref(); match r.vote { 0 => Ok(Response::new(voting::VotingResponse { confirmation: { format!("upvoted for {}", r.url) }, })), 1 => Ok(Response::new(voting::VotingResponse { confirmation: { format!("downvoted for {}", r.url) }, })), _ => Err(Status::new( tonic::Code::OutOfRange, "Invalid vote provided", )), } } } #[tokio::main] async fn main() -> Result<(), Box<dyn std::error::Error>> { let address = "[::1]:8080".parse().unwrap(); let voting_service = VotingService::default(); Server::builder() .add_service(VotingServer::new(voting_service)) .serve(address) .await?; Ok(()) }
src/client.rs内容：
use std::io::stdin; use voting::{voting_client::VotingClient, VotingRequest}; use crate::voting::voting_request; pub mod voting { // tonic::include_proto!("voting"); include!("../protos/voting.rs"); } #[tokio::main] async fn main() -> Result<(), Box<dyn std::error::Error>> { let mut client = VotingClient::connect("http://[::1]:8080").await?; let url = "http://helloworld.com/post1"; loop { let mut vote: String = String::new(); println!("Voting for <{}>, (d)own or (u)p: ", url); stdin().read_line(&mut vote).unwrap(); let vote_res = match vote.trim().to_lowercase().chars().next().unwrap() { 'u' => voting_request::Vote::Up, 'd' => voting_request::Vote::Down, _ => break, }; // here comes the service invocation let request = tonic::Request::new(VotingRequest { url: String::from(url), vote: vote_res.into(), }); let response = client.vote(request).await?; println!("Got: '{}' from service", response.get_ref().confirmation); } Ok(()) }
运行：
# 终端1，运行server端：
$ cargo run --bin server
# 终端2，运行client端：
$ cargo run --bin client
Voting for <http://helloworld.com/post1>, (d)own or (u)p:
u
Got: 'upvoted for http://helloworld.com/post1' from service
Voting for <http://helloworld.com/post1>, (d)own or (u)p:
d
Got: 'downvoted for http://helloworld.com/post1' from service
Voting for <http://helloworld.com/post1>, (d)own or (u)p:
d
多路复用的grpc
如果某客户端和某服务端之间有多个grpc通信需求，则可以复用连接。
例如，有两个proto文件：protos/voting.proto和protos/hello.proto。内容分别如下：
// voting.proto的内容
syntax = "proto3";
package voting;
service Voting {
rpc Vote(VotingRequest) returns (VotingResponse);
}
message VotingRequest {
string url = 1;
enum Vote {
UP = 0;
DOWN = 1;
}
Vote vote = 2;
}
message VotingResponse { string confirmation = 1; }
// hello.proto的内容
syntax = "proto3";
package hello;
service Greeter {
rpc SayHello(HelloReq) returns(HelloResp);
}
message HelloReq {
string content = 1;
}
message HelloResp {
string content = 1;
}
build.rs文件内容：
use std::io::Result; fn main() -> Result<()> { tonic_build::compile_protos("protos/voting.proto")?; tonic_build::compile_protos("protos/hello.proto")?; Ok(()) }
src/client.rs文件内容：
use greet::{greeter_client::GreeterClient, HelloReq}; use tonic::transport::{Channel, Endpoint}; use voting::{voting_client::VotingClient, VotingRequest}; use crate::voting::voting_request; pub mod voting { tonic::include_proto!("voting"); } pub mod greet { tonic::include_proto!("hello"); } type ThisErr = Box<dyn std::error::Error>; async fn voting(client: &mut VotingClient<Channel>) -> Result<(), ThisErr> { let url = "http://helloworld.com/post1"; let mut n = 0; loop { let vote_res = if n % 2 == 0 { voting_request::Vote::Up } else { voting_request::Vote::Down }; let request = tonic::Request::new(VotingRequest { url: String::from(url), vote: vote_res.into(), }); let response = client.vote(request).await?; println!("voting {}, Got: '{}'", n, response.get_ref().confirmation); n += 1; tokio::time::sleep(tokio::time::Duration::from_secs(1)).await; } } async fn greet(client: &mut GreeterClient<Channel>) -> Result<(), ThisErr> { let mut n = 0; loop { let hello_content = format!("hello {}", n); let req = tonic::Request::new(HelloReq { content: hello_content, }); let resp = client.say_hello(req).await?; println!("greet {}, Got: '{}'", n, resp.get_ref().content); n += 1; tokio::time::sleep(tokio::time::Duration::from_secs(1)).await; } } #[tokio::main] async fn main() -> Result<(), ThisErr> { // 构建一个transport::channel::Channel let channel = Endpoint::from_static("http://[::1]:8080").connect().await?; // 从同一个Channel构建两个客户端 let voting_client = VotingClient::new(channel.clone()); let greet_client = GreeterClient::new(channel); // 负责投票服务 let task_voting = tokio::spawn(async move { let mut c = voting_client.clone(); if let Err(e) = voting(&mut c).await { println!("voting error: {}", e); } }); // 负责say hello的服务 let task_greet = tokio::spawn(async move { let mut c = greet_client.clone(); if let Err(e) = greet(&mut c).await { println!("greet error: {}", e); } }); tokio::try_join!(task_greet, task_voting)?; Ok(()) }
src/server.rs文件内容：
use greet::{ greeter_server::{Greeter, GreeterServer}, HelloReq, HelloResp, }; use tonic::{transport::Server, Request, Response, Status}; use voting::{ voting_server::{Voting, VotingServer}, VotingRequest, VotingResponse, }; pub mod voting { tonic::include_proto!("voting"); } pub mod greet { tonic::include_proto!("hello"); } #[derive(Debug, Default)] pub struct VotingService {} #[tonic::async_trait] impl Voting for VotingService { async fn vote( &self, request: Request<VotingRequest>, ) -> Result<Response<VotingResponse>, Status> { let r: &VotingRequest = request.get_ref(); match r.vote { 0 => Ok(Response::new(voting::VotingResponse { confirmation: { format!("upvoted for {}", r.url) }, })), 1 => Ok(Response::new(voting::VotingResponse { confirmation: { format!("downvoted for {}", r.url) }, })), _ => Err(Status::new( tonic::Code::OutOfRange, "Invalid vote provided", )), } } } #[derive(Debug)] pub struct GreetService; #[tonic::async_trait] impl Greeter for GreetService { async fn say_hello(&self, request: Request<HelloReq>) -> Result<Response<HelloResp>, Status> { let hello_str = request.into_inner().content; println!("greet from client: {}", hello_str); Ok(Response::new(HelloResp { content: hello_str })) } } #[tokio::main] async fn main() -> Result<(), Box<dyn std::error::Error>> { let address = "[::1]:8080".parse().unwrap(); let voting_service = VotingService::default(); Server::builder() .add_service(VotingServer::new(voting_service)) .add_service(GreeterServer::new(GreetService)) .serve(address) .await?; Ok(()) }
更多tonic使用方式
tonic官方给的示例，非常详细：https://github.com/hyperium/tonic/tree/master/examples。例如流式(Stream)的grpc、负载均衡、带tls证书验证等。
如果要编写流式grpc，建议看一遍https://github.com/hyperium/tonic/blob/master/examples/routeguide-tutorial.md。