文盘Rust -- tonic-Rust grpc初体验 | 京东云技术团队

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

工程规划

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

1cargo new tonic_sample
2
3
4
5

Cargo.toml 如下

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

tonic 的示例代码还是比较齐全的，本次我们参考 tonic 的 streaming example。

首先编写 proto 文件，用来描述报文。 proto/echo.proto

1syntax = "proto3";
2
3package stream;
4
5// EchoRequest is the request for echo.
6message EchoRequest { string message = 1; }
7
8// EchoResponse is the response for echo.
9message EchoResponse { string message = 1; }
10
11// Echo is the echo service.
12service 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}
23
24
25
26

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

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

该文件用来通过 tonic-build 生成 grpc 的 rust 基础代码

完成上述工作后就可以构建 server 和 client 代码了

stream_server.rs

1pub mod pb {
2    tonic::include_proto!("stream");
3}
4
5use anyhow::Result;
6use futures_util::FutureExt;
7use pb::{EchoRequest, EchoResponse};
8use std::{
9    error::Error,
10    io::ErrorKind,
11    net::{SocketAddr, ToSocketAddrs},
12    pin::Pin,
13    thread,
14    time::Duration,
15};
16use tokio::{
17    net::TcpListener,
18    sync::{
19        mpsc,
20        oneshot::{self, Receiver, Sender},
21        Mutex,
22    },
23    task::{self, JoinHandle},
24};
25use tokio_stream::{
26    wrappers::{ReceiverStream, TcpListenerStream},
27    Stream, StreamExt,
28};
29use tonic::{transport::Server, Request, Response, Status, Streaming};
30type EchoResult<T> = Result<Response<T>, Status>;
31type ResponseStream = Pin<Box<dyn Stream<Item = Result<EchoResponse, Status>> + Send>>;
32
33fn match_for_io_error(err_status: &Status) -> Option<&std::io::Error> {
34    let mut err: &(dyn Error + 'static) = err_status;
35
36    loop {
37        if let Some(io_err) = err.downcast_ref::<std::io::Error>() {
38            return Some(io_err);
39        }
40
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        }
48
49        err = match err.source() {
50            Some(err) => err,
51            None => return None,
52        };
53    }
54}
55
56#[derive(Debug)]
57pub struct EchoServer {}
58
59#[tonic::async_trait]
60impl 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;
63
64        let response = EchoResponse { message: req_str };
65        Ok(Response::new(response))
66    }
67
68    type ServerStreamingEchoStream = ResponseStream;
69
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());
76
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)));
82
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        });
98
99        let output_stream = ReceiverStream::new(rx);
100        Ok(Response::new(
101            Box::pin(output_stream) as Self::ServerStreamingEchoStream
102        ))
103    }
104
105    async fn client_streaming_echo(
106        &self,
107        _: Request<Streaming<EchoRequest>>,
108    ) -> EchoResult<EchoResponse> {
109        Err(Status::unimplemented("not implemented"))
110    }
111
112    type BidirectionalStreamingEchoStream = ResponseStream;
113
114    async fn bidirectional_streaming_echo(
115        &self,
116        req: Request<Streaming<EchoRequest>>,
117    ) -> EchoResult<Self::BidirectionalStreamingEchoStream> {
118        println!("EchoServer::bidirectional_streaming_echo");
119
120        let mut in_stream = req.into_inner();
121        let (tx, rx) = mpsc::channel(128);
122
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                        }
137
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        });
147
148        // echo just write the same data that was received
149        let out_stream = ReceiverStream::new(rx);
150
151        Ok(Response::new(
152            Box::pin(out_stream) as Self::BidirectionalStreamingEchoStream
153        ))
154    }
155}
156
157#[tokio::main]
158async 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}
168
169
170
171
172

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 代码如下

1pub mod pb {
2    tonic::include_proto!("stream");
3}
4
5use std::time::Duration;
6use tokio_stream::{Stream, StreamExt};
7use tonic::transport::Channel;
8
9use pb::{echo_client::EchoClient, EchoRequest};
10
11fn 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}
16
17async 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}
26
27async 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();
35
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}
43
44async fn bidirectional_streaming_echo(client: &mut EchoClient<Channel>, num: usize) {
45    let in_stream = echo_requests_iter().take(num);
46
47    let response = client
48        .bidirectional_streaming_echo(in_stream)
49        .await
50        .unwrap();
51
52    let mut resp_stream = response.into_inner();
53
54    while let Some(received) = resp_stream.next().await {
55        let received = received.unwrap();
56        println!("\treceived message: `{}`", received.message);
57    }
58}
59
60async fn bidirectional_streaming_echo_throttle(client: &mut EchoClient<Channel>, dur: Duration) {
61    let in_stream = echo_requests_iter().throttle(dur);
62
63    let response = client
64        .bidirectional_streaming_echo(in_stream)
65        .await
66        .unwrap();
67
68    let mut resp_stream = response.into_inner();
69
70    while let Some(received) = resp_stream.next().await {
71        let received = received.unwrap();
72        println!("\treceived message: `{}`", received.message);
73    }
74}
75
76#[tokio::main]
77async 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;
82
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
86
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;
90
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;
96
97    Ok(())
98}
99
100
101
102
103

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

1cargo run --bin stream-server
2cargo run --bin stream-client
3
4
5
6

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

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

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

1grpcurl -plaintext 127.0.0.1:50051 list
2
3
4
5

会出现如下报错信息

1Failed to list services: server does not support the reflection API
2
3
4
5

让服务端程序支持 reflection API

首先改造build.rs

1use std::{env, path::PathBuf};
2
3fn 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}
11
12
13
14

file_descriptor_set_path 生成一个文件，其中包含为协议缓冲模块编码的 prost_types::FileDescriptorSet 文件。这是实现 gRPC 服务器反射所必需的。

接下来改造一下 stream-server.rs，涉及两处更改。

新增 STREAM_DESCRIPTOR_SET 常量

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

修改main函数

1#[tokio::main]
2async 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();
10
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();
17
18    let addr = "0.0.0.0:50051".parse().unwrap();
19
20    let server = EchoServer {};
21
22    Server::builder()
23        .add_service(service)
24        .add_service(pb::echo_server::EchoServer::new(server))
25        .serve(addr)
26        .await?;
27    Ok(())
28}
29
30
31
32

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

再次测试

1grpcurl -plaintext 127.0.0.1:50051 list
2grpcurl -plaintext 127.0.0.1:50051 describe stream.Echo
3
4
5
6

返回正确结果。

以上完整代码地址

作者：京东科技贾世闻

来源：京东云开发者社区转载请注明来源

文盘Rust -- tonic-Rust grpc初体验 | 京东云技术团队

工程规划

让服务端程序支持 reflection API

评论区

相关推荐