使用 Rust 构建 gRPC 服务器
Posted
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了使用 Rust 构建 gRPC 服务器相关的知识，希望对你有一定的参考价值。
参考技术A
弗洛里安·克鲁姆 ( Florian Krumm ) 在Unsplash上拍摄的照片
一旦我了解了gRPC和Thrift，就很难再回到使用更具过渡性的基于 JSON 的 REST API 或SOAP API 了。
两个著名的RPC框架 gRPC 和 Thrift 有很多相似之处。前者源自谷歌，后者源自Facebook。它们都易于使用，对多种编程语言都有很好的支持，并且都具有高性能。
最有价值的功能是多种语言的代码生成以及服务器端反射。这些使 API 本质上是类型安全的。使用服务器端反射，可以更轻松地 探索 API 的模式定义，而无需阅读和理解实现。
Apache Thrift在 历史 上一直是一个流行的选择。然而近年来，由于缺乏来自 Facebook 的持续支持，以及与fbthrift的分叉分叉，慢慢失去了人气。
与此同时，gRPC 已经赶上了越来越多的功能，拥有更 健康 的生态系统。
GRPC（蓝色）与 Apache Thrift（红色）的比较。谷歌趋势
gRPC、fbThrift 和 Apache Thrift 之间的 GitHub 明星 历史 。https://star-history.com
到今天为止，除非您的应用程序以某种方式隶属于 Facebook，否则没有充分的理由考虑 Thrift。
GraphQL是另一个由 Facebook 发起的框架。它与上面的两个 RPC 框架有很多相似之处。
移动 API 开发的最大痛点之一是一些用户从不升级他们的应用程序。因为我们想保持向后兼容性，我们要么必须在 API 中保留旧的未使用字段，要么创建 API 的多个版本。GraphQL 的一个动机就是解决这个问题。它被设计成一种“查询语言”，允许客户端指定它需要的数据字段。这使得处理向后兼容性变得更容易。
GraphQL 在开发移动 API 以及面向公众的 API（例如GitHub）方面具有巨大价值。因为，在这两种情况下，我们都无法轻易控制客户端的行为。
但是，如果我们正在为 Web 前端构建 API 或为内部后端服务构建 API，那么选择 GraphQL 而不是 gRPC 几乎没有什么好处。
以上是迄今为止网络框架的一个小概述。除了网络，我们还需要决定应用服务器的语言。
基于Stack Overflow 调查：“六年来，Rust 是最受欢迎的语言。” 尽管学习曲线相对陡峭，但它的类型安全、优雅的内存管理、广泛的社区支持和性能，都使 Rust 成为一种非常有吸引力和有前途的后端服务开发编程语言。
Rust 是最受欢迎的语言。2021 年 Stackoverflow 调查
我们也开始看到 Rust 在行业中得到越来越广泛的采用：Facebook、Dropbox、Yelp、AWS、Google等。很明显，Rust 将继续增长并继续存在。
这就是我们将在今天的教程中看到的内容——在 Rust 中使用 gRPC 构建一个小型服务器。
使用以下命令安装 Rust：
如果您以前安装过 Rust，我们可以通过以下方式对其进行更新：
让我们仔细检查rustc（Rust 编译器）和cargo（Rust 包管理器）的安装版本：
有关安装的更多信息，请查看https://www.rust-lang.org/tools/install。
运行以下命令创建一个新的“Hello World”项目：
让我们编译并运行程序：
这显示了我们到目前为止的文件结构：
gRPC 使用协议缓冲区来序列化和反序列化数据。.proto让我们在一个文件中定义服务器 API 。
我们定义了一个书店服务，只有一个方法：提供一个书本 id，并返回有关该书的一些详细信息。
我们将使用tonic创建我们的 gRPC 服务。将以下依赖项添加到Cargo.toml文件中：
为了从 生成 Rust 代码bookstore.proto，我们tonic-build在 crate 的build.rs构建脚本中使用。
将以下内容添加到build.rs文件中：
需要特别指出的一件事是，我们添加它.out_dir(“./src”)是为了将默认输出目录更改为该src目录，以便我们可以更轻松地查看生成的文件，以达到本文的目的。
在我们准备编译之前还有一件事。tonic-build依赖于Protocol Buffers 编译器将文件解析.proto为可以转换为 Rust 的表示。让我们安装protobuf：
并仔细检查 protobuf 编译器是否安装正确：
准备编译：
有了这个，我们应该src/bookstore.rs生成一个文件。此时，我们的文件结构应该如下所示：
最后，是时候将服务放在一起了。替换为main.rs以下内容：
正如我们所见，为了简单起见，我们并没有真正的图书设置数据库。在这个端点中，我们只是返回一本假书。
运行服务器的时间：
很好，我们在 Rust 中启动并运行了 gRPC 服务器！
正如开头所说，gRPC 最初给我留下了深刻的印象，因为它具有做服务器反射的能力。它不仅在服务开发过程中很方便，而且还使与前端工程师的沟通变得更加容易。因此，如果不解释如何为 Rust 服务器添加它，那么结束本教程是不完整的。
将以下内容添加到依赖项中：
更新build.rs。需要更改的行用// Add this注释标记。
最后，将其更新main.rs为以下内容。
有许多 GUI 客户端可以使用 gRPC Server，例如Postman、Kreya、bloomrpc、grpcox等。为了简单起见，我们将使用命令行工具grpc_cli。
安装：
并测试我们的第一个 gRPC 端点：
看起来它有效！我的朋友，这就是我们在 Rust 中构建 gRPC 服务器的方式。
这就是今天的内容。感谢阅读和快乐的编码！与往常一样，源代码可在GitHub 上获得。
gRPC 客户端不使用服务器端流
【中文标题】gRPC 客户端不使用服务器端流【英文标题】：gRPC client not consuming server side stream 【发布时间】：2021-03-01 10:13:23 【问题描述】：我创建了一个 gRPC 服务器端流服务（在 Rust 中），它从 cluster A 上的 kafka 主题流式传输消息。每次外部客户端发起 websocket 连接时，cluster B 上的 Go 客户端都会连接到 rpc 服务；消息通过此连接流式传输到外部客户端。
然而，Go 客户端似乎并不能可靠地从 rust 服务器消费（即使消息确实已经提交）。即使 websocket 肯定是从外部客户端读取信息；有时它似乎不会从 grpc 流中消耗。我已尝试阅读文档并确定可能导致此问题的任何资源泄漏，但我无法解决问题。
我认为问题可能在于上下文处理或关闭连接（由于流最初成功，然后如果您关闭 websocket 并重新打开它，则消息无法发送）
Go 客户端代码
type KafkaChannel struct
Client bridge.KafkaStreamClient
func (kc *KafkaChannel) Consume(topic string) (<-chan []byte, func() error)
readChan := make(chan []byte)
// Context with cancellation to close the routine
ctx, cancel := context.WithCancel(context.Background())
stream, err := kc.Client.Consume(ctx, &bridge.ConsumeRequest
Topic: topic,
)
if err != nil
log.Fatalf("Error creating cosumer stream: %v", err)
// Launch listener in new thread
go func(reader *chan []byte, consumeStream *bridge.KafkaStream_ConsumeClient)
// Recover from a panic (resources consumed)
defer func()
if recover() != nil
log.Println("Consumer routine closed")
()
for
response, err := stream.Recv()
if err != nil
log.Printf("Error creating cosumer stream: %v", err)
break
switch data := response.OptionalContent.(type)
case *bridge.KafkaResponse_Content:
*reader <- *&data.Content
default:
break
(&readChan, &stream)
// Create a callback func that frees the resources
closeCallback := func() error
err := stream.CloseSend()
close(readChan)
cancel()
return err
return readChan, closeCallback
网络套接字
type ChatHandler struct
Database *db.Queries
Client *grpc.ClientConn
Context context.Context
SessionChannel chan []byte
func (handler *ChatHandler) GetChatConnection(c *websocket.Conn)
//initialisation...
consumer, closeConsume := kChannel.Consume(topic)
for msg := range consumer
log.Printf("Received message from bridge: %s", string(msg))
writeMessageStart := time.Now()
if err = c.WriteMessage(1, msg); err != nil
log.Printf("Error writing message: %v", err)
writeMessageElapsed := time.Since(writeMessageStart)
log.Printf("Write time elapsed error: %s", writeMessageElapsed)
if errors.Is(err, syscall.EPIPE)
log.Printf("Sys error: %v", err)
//continue
closeConsume()
handler.close(c)
return
writeMessageElapsed := time.Since(writeMessageStart)
log.Printf("Write time elapsed no error: %s", writeMessageElapsed)
Rust 服务器端代码
为了完整性
async fn consume(
&self,
request: Request<ConsumeRequest>,
) -> Result<Response<Self::ConsumeStream>, Status>
let (tx, rx) = mpsc::unbounded_channel();
info!("Initiated read-only stream");
tokio::spawn(async move
let message = match Some(request.get_ref())
Some(x) => x,
None => return,
;
let topic = message.topic.clone();
info!("Consuming on topic: ", topic);
let consumer = create_kafka_consumer(topic);
loop
let result = consumer.stream().next().await;
match result
None =>
warn!("Received none-type from consumer stream");
continue;
Some(Err(e)) =>
error!("Error consuming from kafka broker: :?", e);
continue;
Some(Ok(message)) =>
let payload = match message.payload_view::<str>()
None =>
warn!("Recived none-type when unpacking payload");
continue;
Some(Ok(s)) =>
info!("Received payload: :?", s);
s
Some(Err(e)) =>
error!("Error viewing payload contents: ", e);
return;
;
info!("Received message from broker in read-only stream");
if payload.len() > 0
info!("Sending payload :?", payload);
match tx.send(Ok(KafkaResponse
success: true,
optional_content: Some(
bridge::kafka_response::OptionalContent::Content(
(*payload).as_bytes().to_vec(),
),
),
))
Ok(_) => info!("Successfully sent payload to client"),
Err(e) =>
error!("GRPC error sending message to client :?", e);
return;
else
warn!("No content detected in payload from broker");
match consumer.commit_message(&message, CommitMode::Async)
Ok(_) => (),
Err(e) =>
error!("Error commiting a consumed message: :?", e);
return;
);
Ok(Response::new(Box::pin(
tokio_stream::wrappers::UnboundedReceiverStream::new(rx),
)))
【问题讨论】：
【参考方案1】：问题是 Go 客户端的资源泄漏，在某种程度上，Rust 服务器也是如此。
优雅地关闭 Go 客户端流：
对于名为 Consume 的流式 rpc
ctx, cancel := context.WithCancel(context.Background())
stream, err := handler.Client.Consume(ctx, &bridge.ConsumeRequest
Topic: topic,
)
closeCallback := func()
stream.CloseSend()
cancel()
c.Close() // where c := *websocket.Conn
defer closeCallback()
使用可取消的上下文初始化流很重要，这样运行时可以释放与多路复用 rpc 关联的资源。调用cancel() 会关闭客户端的连接（这将在服务器尝试发送新消息时显示为错误）。
在服务器端检测client.CloseSend()。
在 Go 客户端中调用 stream.CloseSend() 不会通知服务器终止连接，它只是表示客户端将停止发送消息。在这种特殊情况下，我在流 rpc 连接中的异步线程中工作，因此为了在客户端与 websocket 断开连接时优雅地关闭连接，检测CloseSend 数据包并关闭@ 非常重要987654328@频道以关闭连接。
请注意，这样做的动机是服务器将继续尝试将数据包发送到“死”接收器，因为否则它会认为连接仍然存在。
以下代码改编自this thread：
pub struct DropReceiver<T>
rx: mpsc::UnboundedReceiver<T>,
impl<T> Stream for DropReceiver<T>
type Item = T;
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>>
Pin::new(&mut self.rx).poll_recv(cx)
// The Drop trait detects the `CloseSend`
impl<T> Drop for DropReceiver<T>
fn drop(&mut self)
info!("Connection has been droped");
self.rx.close(); // Shutdown the receiver
// Within the rpc
let (tx, rx) = mpsc::unbounded_channel();
let rx = DropReceiver rx: rx ;
tokio::spawn(async move /** work */);
Ok(Response::new(Box::pin(rx))) // return this
【讨论】：
以上是关于使用 Rust 构建 gRPC 服务器的主要内容，如果未能解决你的问题，请参考以下文章
grpc,protoc, protoc-gen-go,rust