文盤Rust -- tonic-Rust grpc初體驗

来源:https://www.cnblogs.com/Jcloud/archive/2023/09/18/17710909.html
-Advertisement-
Play Games

gRPC 是開發中常用的開源高性能遠程過程調用（RPC）框架，tonic 是基於 HTTP/2 的 gRPC 實現，專註於高性能、互操作性和靈活性。該庫的創建是為了對 async/await 提供一流的支持，並充當用 Rust 編寫的生產系統的核心構建塊。今天我們聊聊通過使用tonic 調用grpc... ...


gRPC 是開發中常用的開源高性能遠程過程調用（RPC）框架，tonic 是基於 HTTP/2 的 gRPC 實現，專註於高性能、互操作性和靈活性。該庫的創建是為了對 async/await 提供一流的支持，並充當用 Rust 編寫的生產系統的核心構建塊。今天我們聊聊通過使用tonic 調用grpc的的具體過程。

工程規劃

rpc程式一般包含server端和client端，為了方便我們把兩個程式打包到一個工程裡面 新建tonic_sample工程

cargo new tonic_sample

Cargo.toml 如下

[package]
name = "tonic_sample"
version = "0.1.0"
edition = "2021"

[[bin]] # Bin to run the gRPC server
name = "stream-server"
path = "src/stream_server.rs"

[[bin]] # Bin to run the gRPC client
name = "stream-client"
path = "src/stream_client.rs"


[dependencies]
tokio.workspace = true
tonic = "0.9"
tonic-reflection = "0.9.2"
prost = "0.11"
tokio-stream = "0.1"
async-stream = "0.2"
serde = { version = "1.0", features = ["derive"] }
serde_json = "1.0"
rand = "0.7"
h2 = { version = "0.3" }
anyhow = "1.0.75"
futures-util = "0.3.28"

[build-dependencies]
tonic-build = "0.9"

tonic 的示例代碼還是比較齊全的，本次我們參考 tonic 的 streaming example。

首先編寫 proto 文件，用來描述報文。 proto/echo.proto

syntax = "proto3";

package stream;

// EchoRequest is the request for echo.
message EchoRequest { string message = 1; }

// EchoResponse is the response for echo.
message EchoResponse { string message = 1; }

// Echo is the echo service.
service Echo {
  // UnaryEcho is unary echo.
  rpc UnaryEcho(EchoRequest) returns (EchoResponse) {}
  // ServerStreamingEcho is server side streaming.
  rpc ServerStreamingEcho(EchoRequest) returns (stream EchoResponse) {}
  // ClientStreamingEcho is client side streaming.
  rpc ClientStreamingEcho(stream EchoRequest) returns (EchoResponse) {}
  // BidirectionalStreamingEcho is bidi streaming.
  rpc BidirectionalStreamingEcho(stream EchoRequest)
      returns (stream EchoResponse) {}
}

文件並不複雜，只有兩個 message 一個請求一個返回，之所以選擇這個示例是因為該示例包含了rpc中的流式處理，包擴了server 流、client 流以及雙向流的操作。 編輯build.rs 文件

use std::{env, path::PathBuf};

fn main() -> Result<(), Box<dyn std::error::Error>> {
    tonic_build::compile_protos("proto/echo.proto")?;
    Ok(())
}

該文件用來通過 tonic-build 生成 grpc 的 rust 基礎代碼

完成上述工作後就可以構建 server 和 client 代碼了

stream_server.rs

pub mod pb {
    tonic::include_proto!("stream");
}

use anyhow::Result;
use futures_util::FutureExt;
use pb::{EchoRequest, EchoResponse};
use std::{
    error::Error,
    io::ErrorKind,
    net::{SocketAddr, ToSocketAddrs},
    pin::Pin,
    thread,
    time::Duration,
};
use tokio::{
    net::TcpListener,
    sync::{
        mpsc,
        oneshot::{self, Receiver, Sender},
        Mutex,
    },
    task::{self, JoinHandle},
};
use tokio_stream::{
    wrappers::{ReceiverStream, TcpListenerStream},
    Stream, StreamExt,
};
use tonic::{transport::Server, Request, Response, Status, Streaming};
type EchoResult<T> = Result<Response<T>, Status>;
type ResponseStream = Pin<Box<dyn Stream<Item = Result<EchoResponse, Status>> + Send>>;

fn match_for_io_error(err_status: &Status) -> Option<&std::io::Error> {
    let mut err: &(dyn Error + 'static) = err_status;

    loop {
        if let Some(io_err) = err.downcast_ref::<std::io::Error>() {
            return Some(io_err);
        }

        // h2::Error do not expose std::io::Error with `source()`
        // https://github.com/hyperium/h2/pull/462
        if let Some(h2_err) = err.downcast_ref::<h2::Error>() {
            if let Some(io_err) = h2_err.get_io() {
                return Some(io_err);
            }
        }

        err = match err.source() {
            Some(err) => err,
            None => return None,
        };
    }
}

#[derive(Debug)]
pub struct EchoServer {}

#[tonic::async_trait]
impl pb::echo_server::Echo for EchoServer {
    async fn unary_echo(&self, req: Request<EchoRequest>) -> EchoResult<EchoResponse> {
        let req_str = req.into_inner().message;

        let response = EchoResponse { message: req_str };
        Ok(Response::new(response))
    }

    type ServerStreamingEchoStream = ResponseStream;

    async fn server_streaming_echo(
        &self,
        req: Request<EchoRequest>,
    ) -> EchoResult<Self::ServerStreamingEchoStream> {
        println!("EchoServer::server_streaming_echo");
        println!("\tclient connected from: {:?}", req.remote_addr());

        // creating infinite stream with requested message
        let repeat = std::iter::repeat(EchoResponse {
            message: req.into_inner().message,
        });
        let mut stream = Box::pin(tokio_stream::iter(repeat).throttle(Duration::from_millis(200)));

        let (tx, rx) = mpsc::channel(128);
        tokio::spawn(async move {
            while let Some(item) = stream.next().await {
                match tx.send(Result::<_, Status>::Ok(item)).await {
                    Ok(_) => {
                        // item (server response) was queued to be send to client
                    }
                    Err(_item) => {
                        // output_stream was build from rx and both are dropped
                        break;
                    }
                }
            }
            println!("\tclient disconnected");
        });

        let output_stream = ReceiverStream::new(rx);
        Ok(Response::new(
            Box::pin(output_stream) as Self::ServerStreamingEchoStream
        ))
    }

    async fn client_streaming_echo(
        &self,
        _: Request<Streaming<EchoRequest>>,
    ) -> EchoResult<EchoResponse> {
        Err(Status::unimplemented("not implemented"))
    }

    type BidirectionalStreamingEchoStream = ResponseStream;

    async fn bidirectional_streaming_echo(
        &self,
        req: Request<Streaming<EchoRequest>>,
    ) -> EchoResult<Self::BidirectionalStreamingEchoStream> {
        println!("EchoServer::bidirectional_streaming_echo");

        let mut in_stream = req.into_inner();
        let (tx, rx) = mpsc::channel(128);

        tokio::spawn(async move {
            while let Some(result) = in_stream.next().await {
                match result {
                    Ok(v) => tx
                        .send(Ok(EchoResponse { message: v.message }))
                        .await
                        .expect("working rx"),
                    Err(err) => {
                        if let Some(io_err) = match_for_io_error(&err) {
                            if io_err.kind() == ErrorKind::BrokenPipe {
                                eprintln!("\tclient disconnected: broken pipe");
                                break;
                            }
                        }

                        match tx.send(Err(err)).await {
                            Ok(_) => (),
                            Err(_err) => break, // response was droped
                        }
                    }
                }
            }
            println!("\tstream ended");
        });

        // echo just write the same data that was received
        let out_stream = ReceiverStream::new(rx);

        Ok(Response::new(
            Box::pin(out_stream) as Self::BidirectionalStreamingEchoStream
        ))
    }
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // 基礎server
    let server = EchoServer {};
    Server::builder()
        .add_service(pb::echo_server::EchoServer::new(server))
        .serve("0.0.0.0:50051".to_socket_addrs().unwrap().next().unwrap())
        .await
        .unwrap();
    Ok(())
}

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

pub mod pb {
    tonic::include_proto!("stream");
}

use std::time::Duration;
use tokio_stream::{Stream, StreamExt};
use tonic::transport::Channel;

use pb::{echo_client::EchoClient, EchoRequest};

fn echo_requests_iter() -> impl Stream<Item = EchoRequest> {
    tokio_stream::iter(1..usize::MAX).map(|i| EchoRequest {
        message: format!("msg {:02}", i),
    })
}

async fn unary_echo(client: &mut EchoClient<Channel>, num: usize) {
    for i in 0..num {
        let req = tonic::Request::new(EchoRequest {
            message: "msg".to_string() + &i.to_string(),
        });
        let resp = client.unary_echo(req).await.unwrap();
        println!("resp:{}", resp.into_inner().message);
    }
}

async fn streaming_echo(client: &mut EchoClient<Channel>, num: usize) {
    let stream = client
        .server_streaming_echo(EchoRequest {
            message: "foo".into(),
        })
        .await
        .unwrap()
        .into_inner();

    // stream is infinite - take just 5 elements and then disconnect
    let mut stream = stream.take(num);
    while let Some(item) = stream.next().await {
        println!("\treceived: {}", item.unwrap().message);
    }
    // stream is droped here and the disconnect info is send to server
}

async fn bidirectional_streaming_echo(client: &mut EchoClient<Channel>, num: usize) {
    let in_stream = echo_requests_iter().take(num);

    let response = client
        .bidirectional_streaming_echo(in_stream)
        .await
        .unwrap();

    let mut resp_stream = response.into_inner();

    while let Some(received) = resp_stream.next().await {
        let received = received.unwrap();
        println!("\treceived message: `{}`", received.message);
    }
}

async fn bidirectional_streaming_echo_throttle(client: &mut EchoClient<Channel>, dur: Duration) {
    let in_stream = echo_requests_iter().throttle(dur);

    let response = client
        .bidirectional_streaming_echo(in_stream)
        .await
        .unwrap();

    let mut resp_stream = response.into_inner();

    while let Some(received) = resp_stream.next().await {
        let received = received.unwrap();
        println!("\treceived message: `{}`", received.message);
    }
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let mut client = EchoClient::connect("http://127.0.0.1:50051").await.unwrap();
    println!("Unary echo:");
    unary_echo(&mut client, 10).await;
    tokio::time::sleep(Duration::from_secs(1)).await;

    println!("Streaming echo:");
    streaming_echo(&mut client, 5).await;
    tokio::time::sleep(Duration::from_secs(1)).await; //do not mess server println functions

    // Echo stream that sends 17 requests then graceful end that connection
    println!("\r\nBidirectional stream echo:");
    bidirectional_streaming_echo(&mut client, 17).await;

    // Echo stream that sends up to `usize::MAX` requests. One request each 2s.
    // Exiting client with CTRL+C demonstrate how to distinguish broken pipe from
    // graceful client disconnection (above example) on the server side.
    println!("\r\nBidirectional stream echo (kill client with CTLR+C):");
    bidirectional_streaming_echo_throttle(&mut client, Duration::from_secs(2)).await;

    Ok(())
}

測試一下,分別運行 server 和 client

cargo run --bin stream-server
cargo run --bin stream-client

在開發中，我們通常不會再 client 和 server都開發好的情況下才開始測試。通常在開發server 端的時候採用 grpcurl 工具進行測試工作

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

此時，如果我們不指定 -import-path 參數，執行如下命令

grpcurl -plaintext 127.0.0.1:50051 list

會出現如下報錯信息

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

讓服務端程式支持 reflection API

首先改造build.rs

use std::{env, path::PathBuf};

fn main() -> Result<(), Box<dyn std::error::Error>> {
    let out_dir = PathBuf::from(env::var("OUT_DIR").unwrap());
    tonic_build::configure()
        .file_descriptor_set_path(out_dir.join("stream_descriptor.bin"))
        .compile(&["proto/echo.proto"], &["proto"])
        .unwrap();
    Ok(())
}

file_descriptor_set_path 生成一個文件，其中包含為協議緩衝模塊編碼的 prost_types::FileDescriptorSet 文件。這是實現 gRPC 伺服器反射所必需的。

接下來改造一下 stream-server.rs，涉及兩處更改。

新增 STREAM_DESCRIPTOR_SET 常量

pub mod pb {
    tonic::include_proto!("stream");
    pub const STREAM_DESCRIPTOR_SET: &[u8] =
        tonic::include_file_descriptor_set!("stream_descriptor");
}

修改main函數

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // 基礎server
    // let server = EchoServer {};
    // Server::builder()
    //     .add_service(pb::echo_server::EchoServer::new(server))
    //     .serve("0.0.0.0:50051".to_socket_addrs().unwrap().next().unwrap())
    //     .await
    //     .unwrap();

    // tonic_reflection 
    let service = tonic_reflection::server::Builder::configure()
        .register_encoded_file_descriptor_set(pb::STREAM_DESCRIPTOR_SET)
        .with_service_name("stream.Echo")
        .build()
        .unwrap();

    let addr = "0.0.0.0:50051".parse().unwrap();

    let server = EchoServer {};

    Server::builder()
        .add_service(service)
        .add_service(pb::echo_server::EchoServer::new(server))
        .serve(addr)
        .await?;
    Ok(())
}

register_encoded_file_descriptor_set 將包含編碼的 prost_types::FileDescriptorSet 的 byte slice 註冊到 gRPC Reflection 服務生成器註冊。

再次測試

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

返回正確結果。

以上完整代碼地址

作者：京東科技 賈世聞

來源：京東雲開發者社區 轉載請註明來源


您的分享是我們最大的動力!

-Advertisement-
Play Games
更多相關文章
  • Python中裝飾器的順序
    什麼是裝飾器，它們如何被使用，以及我們如何利用它們來構建代碼。我們將看到裝飾器是如何成為一個強大的工具，可以用來為我們的應用程式添加功能，並且可以在Python編程語言中找到。 裝飾器順序 在Python中，裝飾器是一個特殊的函數，可以修改另一個函數的行為。裝飾器是一種設計模式，它在不改變現有對象結 ...
  • Python教程：@符號的用法
    dataclass 到 Python 中的 JSON JavaScript Object Notation或JSON表示使用編程語言中的文本組成的腳本（可執行）文件來存儲和傳輸數據。 Python通過JSON內置模塊支持JSON。因此，我們在Python腳本中導入JSON包，以利用這一能力。 JSO ...
  • 使用 redis 實現分散式介面限流註解 RedisLimit
    前言 很多時候,由於種種不可描述的原因,我們需要針對單個介面實現介面限流,防止訪問次數過於頻繁。這裡就用 redis+aop 實現一個限流介面註解 @RedisLimit 代碼 點擊查看RedisLimit註解代碼 import java.lang.annotation.*; /** * 功能:分佈 ...
  • 10 年程式員的告誡：千萬不要重寫代碼！!
    對重寫代碼說不。 以下為譯文： 1、重寫代碼消耗了12個月！ 我們從頭開始重寫代碼浪費的時間。 你能想象在軟體行業，12個月的時間沒有任何新產品推出，沒有任何新版本更新嗎？ 真的，我不由自主地問自己這個問題： 在這個快速發展的世界里，12月的時間能讓我們做多少事情？ “2015年1月20日，星期二， ...
  • Nacos源碼 (6) Grpc概述與Nacos集成
    Nacos 2.x版本增加了GRPC服務介面和客戶端，極大的提升了Nacos的性能，本文將簡單介紹grpc-java的使用方式以及Nacos中集成GRPC的方式。 grpc-java GRPC是google開源的、以protobuf作為序列化方式、以http2作為通信協議的高性能rpc框架。 grp ...
  • 帶有權重的隨機演算法
    1.什麼是權重比例 權重比例計算即將各數值乘以相應的權數，然後加總求和得到總體值，再除以總的單位數。 如何計算 有一個對象集合為[A,B,C,D,E,F,G,H,I,J]，其對象的全紅 總權重為10 每一個對象的權重為1/10=0.1 2.什麼是權重覆蓋區域 權重覆蓋區域是對象在整體權重範圍中的鎖分 ...
  • JDK21來了！附重要更新說明
    JDK21 計劃23年9月19日正式發佈，雖然一直以來都是“版本隨便出，換 8 算我輸”，但這麼多年這麼多版本的折騰，如果說之前的 LTS版本JDK17你還覺得不香，那 JDK21還是有必要關註一下，因為會有一批重要更新發佈到生產環境中，特別是千呼萬喚的虛擬線程，雖然說這東西我感覺不需要的用不到，需 ...
  • 2023年了，複習了一下spring boot配置使用mongodb
    前言 MongoDB是一個基於分散式文件存儲的開源資料庫系統，使用C++語言編寫。它是一個介於關係資料庫和非關係資料庫之間的產品，具有類似關係資料庫的功能，但又有一些非關係資料庫的特點。MongoDB的數據模型比較鬆散，採用類似json的bson格式，可以靈活地存儲各種類型的數據 MongoDB的優 ...
一周排行
    -Advertisement-
    Play Games
  • 移動開發（一）：使用.NET MAUI開發第一個安卓APP
    移動開發（一）：使用.NET MAUI開發第一個安卓APP 對於工作多年的C#程式員來說，近來想嘗試開發一款安卓APP，考慮了很久最終選擇使用.NET MAUI這個微軟官方的框架來嘗試體驗開發安卓APP，畢竟是使用Visual Studio開發工具，使用起來也比較的順手，結合微軟官方的教程進行了安卓 ...
  • wpf ToggleButton選中效果和一個登錄界面
    前言 QuestPDF 是一個開源 .NET 庫，用於生成 PDF 文檔。使用了C# Fluent API方式可簡化開發、減少錯誤並提高工作效率。利用它可以輕鬆生成 PDF 報告、發票、導出文件等。 項目介紹 QuestPDF 是一個革命性的開源 .NET 庫，它徹底改變了我們生成 PDF 文檔的方 ...
  • 彈幕樹洞項目功能新增篇
    項目地址 項目後端地址： https://github.com/ZyPLJ/ZYTteeHole 項目前端頁面地址： ZyPLJ/TreeHoleVue (github.com) https://github.com/ZyPLJ/TreeHoleVue 目前項目測試訪問地址： http://tree ...
  • 第27篇 sqlserver2022詳細安裝步驟
    話不多說，直接開乾 一.下載 1.官方鏈接下載： https://www.microsoft.com/zh-cn/sql-server/sql-server-downloads 2.在下載目錄中找到下麵這個小的安裝包 SQL2022-SSEI-Dev.exe，運行開始下載SQL server； 二. ...
  • .NET 開源高性能 MQTT 類庫
    前言 隨著物聯網（IoT）技術的迅猛發展，MQTT（消息隊列遙測傳輸）協議憑藉其輕量級和高效性，已成為眾多物聯網應用的首選通信標準。 MQTTnet 作為一個高性能的 .NET 開源庫，為 .NET 平臺上的 MQTT 客戶端與伺服器開發提供了強大的支持。 本文將全面介紹 MQTTnet 的核心功能 ...
  • Serilog文檔翻譯系列（六） - 可用的接收器、增強器、格式化輸出
    Serilog支持多種接收器用於日誌存儲，增強器用於添加屬性，LogContext管理動態屬性，支持多種輸出格式包括純文本、JSON及ExpressionTemplate。還提供了自定義格式化選項，適用於不同需求。 ...
  • 警惕 Visual Studio 屬性求值副作用導致邏輯不符合預期
    目錄簡介獲取 HTML 文檔解析 HTML 文檔測試參考文章 簡介 動態內容網站使用 JavaScript 腳本動態檢索和渲染數據，爬取信息時需要模擬瀏覽器行為，否則獲取到的源碼基本是空的。 本文使用的爬取步驟如下： 使用 Selenium 獲取渲染後的 HTML 文檔 使用 HtmlAgility ...
  • [使用目前最新版]HybridCLR6.9.0+YooAsset2.2.4實現純C# Unity熱更新方案 (一)
    1.前言 什麼是熱更新 游戲或者軟體更新時，無需重新下載客戶端進行安裝，而是在應用程式啟動的情況下，在內部進行資源或者代碼更新 Unity目前常用熱更新解決方案 HybridCLR，Xlua，ILRuntime等 Unity目前常用資源管理解決方案 AssetBundles，Addressable， ...
  • 在 ASP.NET Core Web API 中使用操作篩選器統一處理通用操作
    本文章主要是在C# ASP.NET Core Web API框架實現向手機發送驗證碼簡訊功能。這裡我選擇是一個互億無線簡訊驗證碼平臺，其實像阿裡雲，騰訊雲上面也可以。 首先我們先去 互億無線 https://www.ihuyi.com/api/sms.html 去註冊一個賬號 註冊完成賬號後，它會送 ...
  • 第28篇 如何.net中實現高效可靠數據同步api
    通過以下方式可以高效，並保證數據同步的可靠性 1.API設計 使用RESTful設計，確保API端點明確，並使用適當的HTTP方法（如POST用於創建，PUT用於更新）。 設計清晰的請求和響應模型，以確保客戶端能夠理解預期格式。 2.數據驗證 在伺服器端進行嚴格的數據驗證，確保接收到的數據符合預期格 ...
所有分類