文盘Rust -- tonic-Rust grpc初体验

gRPC 是开发中常用的开源高性能远程过程调用（RPC）框架，tonic 是基于 HTTP/2 的 gRPC 实现，专注于高性能、互操作性和灵活性。该库的创建是为了对 async/await 提供一流的支持，并充当用 Rust 编写的生产系统的核心构建块。今天我们聊聊通过使用tonic 调用grpc的的具体过程。
工程规划

rpc程序一般包含server端和client端，为了方便我们把两个程序打包到一个工程里面 新建tonic_sample工程

1. cargo new tonic_sample

复制代码

Cargo.toml 如下

1. [package]
2. name = "tonic_sample"
3. version = "0.1.0"
4. edition = "2021"
6. [[bin]] # Bin to run the gRPC server
7. name = "stream-server"
8. path = "src/stream_server.rs"
10. [[bin]] # Bin to run the gRPC client
11. name = "stream-client"
12. path = "src/stream_client.rs"
15. [dependencies]
16. tokio.workspace = true
17. tonic = "0.9"
18. tonic-reflection = "0.9.2"
19. prost = "0.11"
20. tokio-stream = "0.1"
21. async-stream = "0.2"
22. serde = { version = "1.0", features = ["derive"] }
23. serde_json = "1.0"
24. rand = "0.7"
25. h2 = { version = "0.3" }
26. anyhow = "1.0.75"
27. futures-util = "0.3.28"
29. [build-dependencies]
30. tonic-build = "0.9"

复制代码

tonic 的示例代码还是比较齐全的，本次我们参考 tonic 的 streaming example。
首先编写 proto 文件，用来描述报文。 proto/echo.proto

1. syntax = "proto3";
3. package stream;
5. // EchoRequest is the request for echo.
6. message EchoRequest { string message = 1; }
8. // EchoResponse is the response for echo.
9. message EchoResponse { string message = 1; }
11. // Echo is the echo service.
12. service Echo {
13.   // UnaryEcho is unary echo.
14.   rpc UnaryEcho(EchoRequest) returns (EchoResponse) {}
15.   // ServerStreamingEcho is server side streaming.
16.   rpc ServerStreamingEcho(EchoRequest) returns (stream EchoResponse) {}
17.   // ClientStreamingEcho is client side streaming.
18.   rpc ClientStreamingEcho(stream EchoRequest) returns (EchoResponse) {}
19.   // BidirectionalStreamingEcho is bidi streaming.
20.   rpc BidirectionalStreamingEcho(stream EchoRequest)
21.       returns (stream EchoResponse) {}
22. }

复制代码

文件并不复杂，只有两个 message 一个请求一个返回，之所以选择这个示例是因为该示例包含了rpc中的流式处理，包扩了server 流、client 流以及双向流的操作。 编辑build.rs 文件

1. use std::{env, path::PathBuf};
3. fn main() -> Result<(), Box<dyn std::error::Error>> {
4.     tonic_build::compile_protos("proto/echo.proto")?;
5.     Ok(())
6. }

复制代码

该文件用来通过 tonic-build 生成 grpc 的 rust 基础代码
完成上述工作后就可以构建 server 和 client 代码了
stream_server.rs

1. pub mod pb {
2.     tonic::include_proto!("stream");
3. }
5. use anyhow::Result;
6. use futures_util::FutureExt;
7. use pb::{EchoRequest, EchoResponse};
8. use std::{
9.     error::Error,
10.     io::ErrorKind,
11.     net::{SocketAddr, ToSocketAddrs},
12.     pin::Pin,
13.     thread,
14.     time::Duration,
15. };
16. use tokio::{
17.     net::TcpListener,
18.     sync::{
19.         mpsc,
20.         oneshot::{self, Receiver, Sender},
21.         Mutex,
22.     },
23.     task::{self, JoinHandle},
24. };
25. use tokio_stream::{
26.     wrappers::{ReceiverStream, TcpListenerStream},
27.     Stream, StreamExt,
28. };
29. use tonic::{transport::Server, Request, Response, Status, Streaming};
30. type EchoResult<T> = Result<Response<T>, Status>;
31. type ResponseStream = Pin<Box<dyn Stream<Item = Result<EchoResponse, Status>> + Send>>;
33. fn match_for_io_error(err_status: &Status) -> Option<&std::io::Error> {
34.     let mut err: &(dyn Error + 'static) = err_status;
36.     loop {
37.         if let Some(io_err) = err.downcast_ref::<std::io::Error>() {
38.             return Some(io_err);
39.         }
41.         // h2::Error do not expose std::io::Error with `source()`
42.         // https://github.com/hyperium/h2/pull/462
43.         if let Some(h2_err) = err.downcast_ref::<h2::Error>() {
44.             if let Some(io_err) = h2_err.get_io() {
45.                 return Some(io_err);
46.             }
47.         }
49.         err = match err.source() {
50.             Some(err) => err,
51.             None => return None,
52.         };
53.     }
54. }
56. #[derive(Debug)]
57. pub struct EchoServer {}
59. #[tonic::async_trait]
60. impl pb::echo_server::Echo for EchoServer {
61.     async fn unary_echo(&self, req: Request<EchoRequest>) -> EchoResult<EchoResponse> {
62.         let req_str = req.into_inner().message;
64.         let response = EchoResponse { message: req_str };
65.         Ok(Response::new(response))
66.     }
68.     type ServerStreamingEchoStream = ResponseStream;
70.     async fn server_streaming_echo(
71.         &self,
72.         req: Request<EchoRequest>,
73.     ) -> EchoResult<Self::ServerStreamingEchoStream> {
74.         println!("EchoServer::server_streaming_echo");
75.         println!("\tclient connected from: {:?}", req.remote_addr());
77.         // creating infinite stream with requested message
78.         let repeat = std::iter::repeat(EchoResponse {
79.             message: req.into_inner().message,
80.         });
81.         let mut stream = Box::pin(tokio_stream::iter(repeat).throttle(Duration::from_millis(200)));
83.         let (tx, rx) = mpsc::channel(128);
84.         tokio::spawn(async move {
85.             while let Some(item) = stream.next().await {
86.                 match tx.send(Result::<_, Status>::Ok(item)).await {
87.                     Ok(_) => {
88.                         // item (server response) was queued to be send to client
89.                     }
90.                     Err(_item) => {
91.                         // output_stream was build from rx and both are dropped
92.                         break;
93.                     }
94.                 }
95.             }
96.             println!("\tclient disconnected");
97.         });
99.         let output_stream = ReceiverStream::new(rx);
100.         Ok(Response::new(
101.             Box::pin(output_stream) as Self::ServerStreamingEchoStream
102.         ))
103.     }
105.     async fn client_streaming_echo(
106.         &self,
107.         _: Request<Streaming<EchoRequest>>,
108.     ) -> EchoResult<EchoResponse> {
109.         Err(Status::unimplemented("not implemented"))
110.     }
112.     type BidirectionalStreamingEchoStream = ResponseStream;
114.     async fn bidirectional_streaming_echo(
115.         &self,
116.         req: Request<Streaming<EchoRequest>>,
117.     ) -> EchoResult<Self::BidirectionalStreamingEchoStream> {
118.         println!("EchoServer::bidirectional_streaming_echo");
120.         let mut in_stream = req.into_inner();
121.         let (tx, rx) = mpsc::channel(128);
123.         tokio::spawn(async move {
124.             while let Some(result) = in_stream.next().await {
125.                 match result {
126.                     Ok(v) => tx
127.                         .send(Ok(EchoResponse { message: v.message }))
128.                         .await
129.                         .expect("working rx"),
130.                     Err(err) => {
131.                         if let Some(io_err) = match_for_io_error(&err) {
132.                             if io_err.kind() == ErrorKind::BrokenPipe {
133.                                 eprintln!("\tclient disconnected: broken pipe");
134.                                 break;
135.                             }
136.                         }
138.                         match tx.send(Err(err)).await {
139.                             Ok(_) => (),
140.                             Err(_err) => break, // response was droped
141.                         }
142.                     }
143.                 }
144.             }
145.             println!("\tstream ended");
146.         });
148.         // echo just write the same data that was received
149.         let out_stream = ReceiverStream::new(rx);
151.         Ok(Response::new(
152.             Box::pin(out_stream) as Self::BidirectionalStreamingEchoStream
153.         ))
154.     }
155. }
157. #[tokio::main]
158. async fn main() -> Result<(), Box<dyn std::error::Error>> {
159.     // 基础server
160.     let server = EchoServer {};
161.     Server::builder()
162.         .add_service(pb::echo_server::EchoServer::new(server))
163.         .serve("0.0.0.0:50051".to_socket_addrs().unwrap().next().unwrap())
164.         .await
165.         .unwrap();
166.     Ok(())
167. }

复制代码

server 端的代码还是比较清晰的，首先通过 tonic::include_proto! 宏引入grpc定义，参数是 proto 文件中定义的 package 。我们重点说说 server_streaming_echo function 。这个function 的处理流程明白了，其他的流式处理大同小异。首先 通过std::iter::repeat function 定义一个迭代器；然后构建 tokio_stream 在本示例中 每 200毫秒产生一个 repeat；最后构建一个 channel ，tx 用来发送从stream中获取的内容太，rx 封装到response 中返回。 最后 main 函数 拉起服务。
client 代码如下

1. pub mod pb {
2.     tonic::include_proto!("stream");
3. }
5. use std::time::Duration;
6. use tokio_stream::{Stream, StreamExt};
7. use tonic::transport::Channel;
9. use pb::{echo_client::EchoClient, EchoRequest};
11. fn echo_requests_iter() -> impl Stream<Item = EchoRequest> {
12.     tokio_stream::iter(1..usize::MAX).map(|i| EchoRequest {
13.         message: format!("msg {:02}", i),
14.     })
15. }
17. async fn unary_echo(client: &mut EchoClient<Channel>, num: usize) {
18.     for i in 0..num {
19.         let req = tonic::Request::new(EchoRequest {
20.             message: "msg".to_string() + &i.to_string(),
21.         });
22.         let resp = client.unary_echo(req).await.unwrap();
23.         println!("resp:{}", resp.into_inner().message);
24.     }
25. }
27. async fn streaming_echo(client: &mut EchoClient<Channel>, num: usize) {
28.     let stream = client
29.         .server_streaming_echo(EchoRequest {
30.             message: "foo".into(),
31.         })
32.         .await
33.         .unwrap()
34.         .into_inner();
36.     // stream is infinite - take just 5 elements and then disconnect
37.     let mut stream = stream.take(num);
38.     while let Some(item) = stream.next().await {
39.         println!("\treceived: {}", item.unwrap().message);
40.     }
41.     // stream is droped here and the disconnect info is send to server
42. }
44. async fn bidirectional_streaming_echo(client: &mut EchoClient<Channel>, num: usize) {
45.     let in_stream = echo_requests_iter().take(num);
47.     let response = client
48.         .bidirectional_streaming_echo(in_stream)
49.         .await
50.         .unwrap();
52.     let mut resp_stream = response.into_inner();
54.     while let Some(received) = resp_stream.next().await {
55.         let received = received.unwrap();
56.         println!("\treceived message: `{}`", received.message);
57.     }
58. }
60. async fn bidirectional_streaming_echo_throttle(client: &mut EchoClient<Channel>, dur: Duration) {
61.     let in_stream = echo_requests_iter().throttle(dur);
63.     let response = client
64.         .bidirectional_streaming_echo(in_stream)
65.         .await
66.         .unwrap();
68.     let mut resp_stream = response.into_inner();
70.     while let Some(received) = resp_stream.next().await {
71.         let received = received.unwrap();
72.         println!("\treceived message: `{}`", received.message);
73.     }
74. }
76. #[tokio::main]
77. async fn main() -> Result<(), Box<dyn std::error::Error>> {
78.     let mut client = EchoClient::connect("http://127.0.0.1:50051").await.unwrap();
79.     println!("Unary echo:");
80.     unary_echo(&mut client, 10).await;
81.     tokio::time::sleep(Duration::from_secs(1)).await;
83.     println!("Streaming echo:");
84.     streaming_echo(&mut client, 5).await;
85.     tokio::time::sleep(Duration::from_secs(1)).await; //do not mess server println functions
87.     // Echo stream that sends 17 requests then graceful end that connection
88.     println!("\r\nBidirectional stream echo:");
89.     bidirectional_streaming_echo(&mut client, 17).await;
91.     // Echo stream that sends up to `usize::MAX` requests. One request each 2s.
92.     // Exiting client with CTRL+C demonstrate how to distinguish broken pipe from
93.     // graceful client disconnection (above example) on the server side.
94.     println!("\r\nBidirectional stream echo (kill client with CTLR+C):");
95.     bidirectional_streaming_echo_throttle(&mut client, Duration::from_secs(2)).await;
97.     Ok(())
98. }

复制代码

测试一下,分别运行 server 和 client

1. cargo run --bin stream-server
2. cargo run --bin stream-client

复制代码

在开发中，我们通常不会再 client 和 server都开发好的情况下才开始测试。通常在开发server 端的时候采用 grpcurl 工具进行测试工作

1. grpcurl -import-path ./proto -proto echo.proto list
2. grpcurl -import-path ./proto -proto  echo.proto describe stream.Echo
3. grpcurl -plaintext -import-path ./proto -proto  echo.proto -d '{"message":"1234"}' 127.0.0.1:50051 stream.Echo/UnaryEcho

复制代码

此时，如果我们不指定 -import-path 参数，执行如下命令

1. grpcurl -plaintext 127.0.0.1:50051 list

复制代码

会出现如下报错信息

1. Failed to list services: server does not support the reflection API

复制代码

让服务端程序支持 reflection API

首先改造build.rs

1. use std::{env, path::PathBuf};
3. fn main() -> Result<(), Box<dyn std::error::Error>> {
4.     let out_dir = PathBuf::from(env::var("OUT_DIR").unwrap());
5.     tonic_build::configure()
6.         .file_descriptor_set_path(out_dir.join("stream_descriptor.bin"))
7.         .compile(&["proto/echo.proto"], &["proto"])
8.         .unwrap();
9.     Ok(())
10. }

复制代码

file_descriptor_set_path 生成一个文件，其中包含为协议缓冲模块编码的 prost_types::FileDescriptorSet 文件。这是实现 gRPC 服务器反射所必需的。
接下来改造一下 stream-server.rs，涉及两处更改。
新增 STREAM_DESCRIPTOR_SET 常量

1. pub mod pb {
2.     tonic::include_proto!("stream");
3.     pub const STREAM_DESCRIPTOR_SET: &[u8] =
4.         tonic::include_file_descriptor_set!("stream_descriptor");
5. }

复制代码

修改main函数

1. #[tokio::main]
2. async fn main() -> Result<(), Box<dyn std::error::Error>> {
3.     // 基础server
4.     // let server = EchoServer {};
5.     // Server::builder()
6.     //     .add_service(pb::echo_server::EchoServer::new(server))
7.     //     .serve("0.0.0.0:50051".to_socket_addrs().unwrap().next().unwrap())
8.     //     .await
9.     //     .unwrap();
11.     // tonic_reflection
12.     let service = tonic_reflection::server::Builder::configure()
13.         .register_encoded_file_descriptor_set(pb::STREAM_DESCRIPTOR_SET)
14.         .with_service_name("stream.Echo")
15.         .build()
16.         .unwrap();
18.     let addr = "0.0.0.0:50051".parse().unwrap();
20.     let server = EchoServer {};
22.     Server::builder()
23.         .add_service(service)
24.         .add_service(pb::echo_server::EchoServer::new(server))
25.         .serve(addr)
26.         .await?;
27.     Ok(())
28. }

复制代码

register_encoded_file_descriptor_set 将包含编码的 prost_types::FileDescriptorSet 的 byte slice 注册到 gRPC Reflection 服务生成器注册。
再次测试

1. grpcurl -plaintext 127.0.0.1:50051 list
2. grpcurl -plaintext 127.0.0.1:50051 describe stream.Echo

复制代码

返回正确结果。
以上完整代码地址

作者：京东科技 贾世闻
来源：京东云开发者社区 转载请注明来源

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！