数据库流处理后端数据工程【免费下载链接】risingwaveEvent streaming platform for agentic AI. Continuously ingest, transform, and serve event streams in real time, at scale.项目地址https://gitcode.com/gh_mirrors/ri/risingwave点击查看免费下载RisingWave 的各个组件meta、compute、frontend 等之间通过基于 tonicRust 的 gRPC 框架的 RPC 服务进行通信。本文以官方开发文档 how-to-write-a-rpc-service.md 为主线结合仓库内的真实源码完整讲解在 RisingWave 中新增一个 RPC 服务的全部步骤定义 proto 消息与 service、用 Rust 实现服务端逻辑、接入prost代码生成体系、注册模块并在服务端启动监听。读完本文你可以照着同样的套路为任意组件新增一个可用的 tonic 服务并理解其背后的代码生成与启动机制。整体流程一览在 RisingWave 中新增一个 RPC 服务本质上是把写一个 gRPC 服务这件事拆成五个互相独立、又通过代码生成串联起来的步骤定义 proto 文件在仓库根目录的 proto/ 下新增或修改*.proto声明消息message和服务service实现服务结构体在对应组件的源码目录下新建*_service.rs编写impl实现 tonic 根据 proto 生成的Xxxtrait接入代码生成在 src/prost/build.rs 的 proto 文件清单中加入新文件并在 src/prost/src/lib.rs 中声明生成的模块注册模块在组件 crate 的模块文件如 meta 的 src/meta/service/src/lib.rs中声明pub mod xxx_service;挂载启动在组件的 server 启动入口如 src/meta/node/src/server.rs用Server::builder().add_service(XxxServer::new(xxx_srv))注册并serve。运行risedev d或执行一次cargo build会触发 src/prost/build.rs 里的tonic_build自动从 proto 生成 trait、Server 与消息类型的样板代码剩下的手写工作就只有实现业务逻辑这一件事。第一步定义 proto 消息与服务所有 proto 文件统一放在仓库根目录的 proto/ 目录下。官方文档以健康检查服务为例其完整定义见 proto/health.protosyntax proto3; package health; option java_package com.risingwave.proto; // This proto is copied from https://github.com/grpc/grpc/blob/v1.15.0/doc/health-checking.md message HealthCheckRequest { string service 1; } message HealthCheckResponse { enum ServingStatus { UNKNOWN 0; SERVING 1; NOT_SERVING 2; } ServingStatus status 1; } service Health { rpc Check(HealthCheckRequest) returns (HealthCheckResponse); }几点需要留意的细节文件顶部必须写syntax proto3;proto3 是 tonic/prost 生成的默认语法版本package health;决定了生成的 Rust 模块名与类型名如risingwave_pb::health消息字段需要显式编号 1、 2字段编号是 proto 二进制格式的一部分一旦发布给客户端使用就不应随意改动枚举的第一个成员必须是 0UNKNOWN 0这是 proto3 对零值的约定也方便用0表示未设置/未知状态该文件来源于 gRPC 官方健康检查协议RisingWave 的 meta、compute、frontend 三个组件都复用了这一份定义用来对外暴露自己的存活状态。文档同时提醒编写完成后用buf对 proto 文件做 lint 检查如buf lint保证命名、字段编号等符合规范。第二步用 Rust 实现服务端逻辑服务实现文件放哪里文档给出的路径模式是src/component/src/rpc/service/service_name.rs例如src/meta/src/rpc/service/health_service.rs。需要说明的是当前仓库的 meta 模块已经过目录重构服务实现统一放在 src/meta/service/src/ 下健康检查实现位于 src/meta/service/src/health_service.rs。因此新增服务时优先把实现文件放进对应组件自己的src/目录meta 放src/meta/service/src/compute 放src/compute/src/rpc/service/等保持与仓库现状一致。服务实现代码以 meta 的健康检查服务为例完整实现见 src/meta/service/src/health_service.rsuse risingwave_pb::health::health_check_response::ServingStatus; use risingwave_pb::health::health_server::Health; use risingwave_pb::health::{HealthCheckRequest, HealthCheckResponse}; use tonic::{Request, Response, Status}; pub struct HealthServiceImpl {} impl Default for HealthServiceImpl { fn default() - Self { Self::new() } } impl HealthServiceImpl { pub fn new() - Self { Self {} } } #[async_trait::async_trait] impl Health for HealthServiceImpl { async fn check( self, _request: RequestHealthCheckRequest, ) - ResultResponseHealthCheckResponse, Status { // Reply serving as long as tonic service is started Ok(Response::new(HealthCheckResponse { status: ServingStatus::Serving as i32, })) } }代码结构拆解HealthServiceImpl是服务端的业务容器通常内部持有该服务需要的依赖如元数据管理、存储句柄等健康检查示例因为不需要任何状态所以是一个空结构体impl Health for HealthServiceImpl实现了 tonic 从 proto 中service Health生成的 traitHealth。trait 的每个方法对应 proto 中的一条rpc方法签名遵循 tonic 约定入参是RequestHealthCheckRequestgRPC 请求可以携带 metadata 元数据所以 tonic 用RequestT包装返回ResultResponseHealthCheckResponse, StatusServingStatus::Serving as i32把枚举值转为 proto 生成的 int32 字段——proto 枚举在 prost 生成的代码中表现为 Rust 枚举 i32 的互转#[async_trait::async_trait]是必须的宏tonic 生成的 trait 方法返回的是async fn语义需要 async_trait 才能在 trait 中安全地使用async fnRust 稳定版尚未原生支持 trait 内的 async fn。同一份 proto 在多个组件中的复用health.proto在仓库中被三个组件同时复用它们是理解一个 proto 服务如何在多组件落地的极佳范本src/meta/service/src/health_service.rsmeta 节点实现额外提供了Defaultsrc/compute/src/rpc/service/health_service.rscompute 节点实现src/frontend/src/health_service.rsfrontend 节点实现。三份实现的业务逻辑完全一致进程活着就返回Serving区别仅在于所在 crate 的目录结构与对 clippy 注解的处理方式。这也说明同一份 proto 定义可以被任意组件导入使用每个组件只需在自己的目录下实现对应 trait 即可。第三步把 proto 接入 prost 代码生成体系proto 文件本身不会自动变成 Rust 代码需要手动接入 src/prost crate 的构建脚本 src/prost/build.rs。在 build.rs 的文件清单中登记build.rs的main()里维护了一个proto_files字符串数组src/prost/build.rs把新 proto 的文件名不含.proto后缀加入其中let proto_files vec![ backup_service, batch_plan, catalog, cloud_service, common, compactor, compute, connector_service, data, ddl_service, expr, health, // - 新增的 proto 文件登记在这里 hummock, iceberg_compaction, java_binding, meta, // ... ];在 lib.rs 中声明生成模块随后在 src/prost/src/lib.rs 中声明对应的模块源码中 health 模块声明于第 112-117 行#[rustfmt::skip] #[cfg_attr(madsim, path sim/health.rs)] pub mod health;这里有两个关键注解#[rustfmt::skip]生成代码不接受 rustfmt 格式化避免格式冲突#[cfg_attr(madsim, path sim/health.rs)]当以madsimRisingWave 用于确定性模拟测试的运行时模拟器编译时改用sim/health.rs这个 mock 版本保证 gRPC 调用在模拟环境如 e2e_test 的确定性测试中无需真实网络即可运行。生成机制tonic_build 与增量拷贝src/prost/build.rs 的核心逻辑第 1143-1310 行值得展开用tonic_build::configure()构造配置设置file_descriptor_set_path用于反射/序列化支持的 FileDescriptorSet、compile_well_known_types、各类type_attribute/boxed派生项调用compile_protos_with_config(...)src/prost/build.rs编译全部 proto 文件输出到OUT_DIR同时生成 tonic 的XxxServer/XxxClient/trait 以及 prost 的消息结构体随后pbjson_build生成 serde 序列化支持最后compare_and_copysrc/prost/build.rs把OUT_DIR中的产物只在内容发生变化时才拷贝回src/prost/src/——这是为了避免每次构建因时间戳变化触发不必要的重编译详见代码中引用的 issue #11449 说明。因此新增服务后第一次构建时risedev d或cargo build会自动生成HealthServer、HealthClient、Healthtrait、HealthCheckRequest/HealthCheckResponse等样板代码你不需要手写它们直接按第 2 节的方式impl Health即可。第四步在组件 crate 中注册模块生成好的实现文件还需要在组件 crate 的模块根文件中登记否则 Rust 编译器看不到它。对于 meta 组件登记位置是 src/meta/service/src/lib.rspub mod backup_service; pub mod cloud_service; pub mod cluster_limit_service; pub mod cluster_service; pub mod ddl_service; pub mod event_log_service; pub mod health_service; // - 在这里注册 pub mod heartbeat_service; pub mod hosted_iceberg_catalog_service; pub mod hummock_service; pub mod meta_member_service; pub mod monitor_service; pub mod notification_service; // ...原文档提到的src/meta/src/rpc/service/mod.rs在当前仓库中已随目录重构被替换为上述 lib.rs。compute、frontend 等其他组件也遵循同样的模式在各自的模块文件中追加pub mod xxx_service;。第五步在 server 启动入口挂载服务最后一步是把服务挂到 gRPC 服务器上并开始监听。文档给出的示例是let health_srv HealthServiceImpl::new(); tokio::spawn(async move { tonic::transport::Server::builder() .layer(MetricsMiddlewareLayer::new(meta_metrics.clone())) .add_service(HealthServer::new(health_srv)) .serve(address_info.listen_addr) .await .unwrap(); });当前仓库的真实启动代码随着仓库演进meta 的 RPC 服务器启动逻辑目前位于 src/meta/node/src/server.rs。以 meta 从节点election follower的启动函数start_service_as_election_follower第 270-299 行为例真实写法是let meta_member_srv MetaMemberServiceImpl::new(election_client); let health_srv HealthServiceImpl::new(); let server tonic::transport::Server::builder() .layer(MetricsMiddlewareLayer::new(Arc::new( GLOBAL_META_METRICS.clone(), ))) .layer(TracingExtractLayer::new()) .add_service(MetaMemberServiceServer::new(meta_member_srv)) .add_service(HealthServer::new(health_srv)) .monitored_serve_with_shutdown( address_info.listen_addr, grpc-meta-follower-service, TcpConfig { tcp_nodelay: true, keepalive_duration: None, }, shutdown.clone().cancelled_owned(), );相比文档示例真实代码多出几处值得说明的演进点TracingExtractLayer配合tracing日志体系从 gRPC 请求的 metadata 中提取链路追踪上下文如 trace ID让一次跨组件调用的日志可以串联monitored_serve_with_shutdownRisingWave 封装的服务启动函数在tonic::transport::Server基础上额外支持优雅关停接收shutdowntoken、对监听线程的服务命名便于监控面板展示以及TcpConfig如tcp_nodelay: true降低流式小消息的延迟MetricsMiddlewareLayer这是 RisingWave 自定义的 tower 中间件定义在 src/meta/src/rpc/intercept.rs实现tower::Layer为每个 gRPC 请求注入指标采集MetaMetrics使服务调用可以被 Prometheus 监控到。其他组件的挂载对照健康检查服务在其他组件的挂载位置也印证了这套写法的通用性compute 节点在 src/compute/src/server.rs 中通过.add_service(HealthServer::new(health_srv))与 ConfigService 等一起注册约第 510-513 行frontend 节点在 src/frontend/src/session.rs 中用tokio::spawn启动一个包含 HealthServer、FrontendServiceServer、monitor 服务的 gRPC 服务器约第 453-457 行meta 主节点在 src/meta/node/src/server.rs 中与 CloudService、ScaleService、BackupService 等十余个服务并列注册约第 814-818 行。一个 proto 定义 各自组件的实现与挂载就构成了完整的服务落地。生成与验证完成以上五步后在仓库根目录运行risedev d开发命令或直接cargo buildsrc/prost/build.rs会自动生成HealthServer等样板代码如果构建提示找不到生成的类型如health_server::Health检查 build.rs 的proto_files是否已登记、src/prost/src/lib.rs 是否已声明模块服务启动后可用任意支持 gRPC 的客户端如grpcurl使用 proto/health.proto 的 import 描述调用Check验证返回SERVING由于 meta 的主/从节点都对外提供健康检查集群的探活、监控面板的存活状态展示均依赖该服务返回的ServingStatus。小结在 RisingWave 中新增 RPC 服务是一条清晰的流水线proto 定义proto/health.proto→ Rust 实现src/meta/service/src/health_service.rs→ 生成接入src/prost/build.rs src/prost/src/lib.rs→ 模块注册src/meta/service/src/lib.rs→ 服务器挂载src/meta/node/src/server.rs。其中样板代码全部由 tonic/prost 构建脚本自动产出开发者只需要写好业务逻辑的 trait 实现再通过risedev d或cargo build触发生成即可。如果需要编写更复杂的服务例如带流式响应的server_streamingRPC、携带大量 metadata 或超大消息体在同样的骨架上追加 proto 定义、调整tonic_build的配置如max_decoding_message_size、boxed注解即可相关配置模式都能在 src/prost/build.rs 中找到现成参考。赞分享数据库流处理后端数据工程【免费下载链接】risingwaveEvent streaming platform for agentic AI. Continuously ingest, transform, and serve event streams in real time, at scale.项目地址https://gitcode.com/gh_mirrors/ri/risingwave点击查看免费下载相关推荐Qdrant 如何扩展 gRPC 服务proto 定义、tonic 服务实现与集成测试验证Qdrant 如何扩展 gRPC 服务proto 定义、tonic 服务实现与集成测试验证 如果你要在 Qdrant 源码上给 gRPC API 增加新的请求向量数据库数据库后端搜索引擎gRPC服务定义booking-microservices中的.proto文件设计gRPC服务定义booking microservices中的.proto文件设计 在微服务架构中gRPCGoogle Remote ProcedureTalos Linux gRPC API 全面参考从 proto 定义到机器服务的实战指南Talos Linux gRPC API 全面参考从 proto 定义到机器服务的实战指南 本文基于仓库中的 API 参考文档 https://link.gi云原生操作系统容器编排上一篇Llama-medx_v3性能优化指南提升昇腾平台推理速度的5个关键技巧下一篇如何在VS Code中实现Office文件预览这款免费插件让你告别软件切换创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考