文件目录
toml 用的不是新版本都是稳定版本，新版本各种报错不兼容
[dependencies]
tokio = { version = "1", features = ["full"] }
prost = "0.12"
tonic = { version = "0.10" }[build-dependencies]
tonic-build = "0.10"
proto
syntax = "proto3";package biz_activity;// 定义业务活动消息
message BizActivity {optional string id = 1;optional string name = 2;optional int32 status = 3;optional int64 create_time = 4; // 使用时间戳optional string additional_field = 5;
}// 定义请求和响应消息
message GetBizActivityRequest {string id = 1;
}message GetBizActivityResponse {BizActivity activity = 1;
}message CreateBizActivityRequest {BizActivity activity = 1;
}message CreateBizActivityResponse {BizActivity activity = 1;
}message UpdateBizActivityRequest {BizActivity activity = 1;
}message UpdateBizActivityResponse {BizActivity activity = 1;
}message DeleteBizActivityRequest {string id = 1;
}message DeleteBizActivityResponse {bool success = 1;
}message ListBizActivitiesRequest {optional int32 page = 1;optional int32 size = 2;
}message ListBizActivitiesResponse {repeated BizActivity activities = 1;
}// 定义 gRPC 服务
service BizActivityService {rpc GetBizActivity(GetBizActivityRequest) returns (GetBizActivityResponse);rpc CreateBizActivity(CreateBizActivityRequest) returns (CreateBizActivityResponse);rpc UpdateBizActivity(UpdateBizActivityRequest) returns (UpdateBizActivityResponse);rpc DeleteBizActivity(DeleteBizActivityRequest) returns (DeleteBizActivityResponse);rpc ListBizActivities(ListBizActivitiesRequest) returns (ListBizActivitiesResponse);
}
build.rs
fn main() -> Result<(), Box<dyn std::error::Error>> {// 编译 .proto 文件用于 gRPC 服务tonic_build::compile_protos("src/protos/biz_activity.proto")?;Ok(())
}
server
use tonic::{transport::Server, Request, Response, Status};
use std::time::{SystemTime, UNIX_EPOCH};// 引入生成的 gRPC 代码
pub mod biz_activity {tonic::include_proto!("biz_activity");
}use biz_activity::{biz_activity_service_server::{BizActivityService, BizActivityServiceServer},BizActivity, GetBizActivityRequest, GetBizActivityResponse,CreateBizActivityRequest, CreateBizActivityResponse,UpdateBizActivityRequest, UpdateBizActivityResponse,DeleteBizActivityRequest, DeleteBizActivityResponse,ListBizActivitiesRequest, ListBizActivitiesResponse,
};// 模拟数据库存储
use std::collections::HashMap;
use std::sync::{Arc, Mutex};#[derive(Default)]
pub struct BizActivityServiceImpl {// 使用内存存储模拟数据库activities: Arc<Mutex<HashMap<String, BizActivity>>>,
}#[tonic::async_trait]
impl BizActivityService for BizActivityServiceImpl {async fn get_biz_activity(&self,request: Request<GetBizActivityRequest>,) -> Result<Response<GetBizActivityResponse>, Status> {let req = request.into_inner();let activities = self.activities.lock().unwrap();let activity = activities.get(&req.id).cloned();match activity {Some(activity) => {let response = GetBizActivityResponse {activity: Some(activity),};Ok(Response::new(response))}None => Err(Status::not_found("Activity not found")),}}async fn create_biz_activity(&self,request: Request<CreateBizActivityRequest>,) -> Result<Response<CreateBizActivityResponse>, Status> {let req = request.into_inner();let mut activity = req.activity.unwrap();// 如果没有提供 ID，则生成一个if activity.id.is_none() {let id = format!("activity_{}", SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_secs());activity.id = Some(id);}// 设置创建时间if activity.create_time.is_none() {let timestamp = SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_secs() as i64;activity.create_time = Some(timestamp);}let id = activity.id.clone().unwrap();let mut activities = self.activities.lock().unwrap();activities.insert(id.clone(), activity.clone());let response = CreateBizActivityResponse {activity: Some(activity),};Ok(Response::new(response))}async fn update_biz_activity(&self,request: Request<UpdateBizActivityRequest>,) -> Result<Response<UpdateBizActivityResponse>, Status> {let req = request.into_inner();let activity = req.activity.unwrap();let id = activity.id.clone().unwrap();let mut activities = self.activities.lock().unwrap();if !activities.contains_key(&id) {return Err(Status::not_found("Activity not found"));}activities.insert(id.clone(), activity.clone());let response = UpdateBizActivityResponse {activity: Some(activity),};Ok(Response::new(response))}async fn delete_biz_activity(&self,request: Request<DeleteBizActivityRequest>,) -> Result<Response<DeleteBizActivityResponse>, Status> {let req = request.into_inner();let mut activities = self.activities.lock().unwrap();let existed = activities.remove(&req.id);let response = DeleteBizActivityResponse {success: existed.is_some(),};Ok(Response::new(response))}async fn list_biz_activities(&self,request: Request<ListBizActivitiesRequest>,) -> Result<Response<ListBizActivitiesResponse>, Status> {let req = request.into_inner();let activities = self.activities.lock().unwrap();let mut activity_list: Vec<BizActivity> = activities.values().cloned().collect();// 分页处理 - 修复字段访问方式let page = req.page.unwrap_or(1) as usize;let size = req.size.unwrap_or(10) as usize;let start = (page - 1) * size;let end = start + size;if start < activity_list.len() {let end = std::cmp::min(end, activity_list.len());activity_list = activity_list[start..end].to_vec();} else {activity_list = vec![];}let response = ListBizActivitiesResponse {activities: activity_list,};Ok(Response::new(response))}
}pub async fn start_grpc_server() -> Result<(), Box<dyn std::error::Error>> {let addr = "127.0.0.1:50051".parse().unwrap();let service_impl = BizActivityServiceImpl::default();println!("gRPC 服务器运行在 {}", addr);Server::builder().add_service(BizActivityServiceServer::new(service_impl)).serve(addr).await?;Ok(())
}
client
use tonic::transport::Channel;
use std::time::{SystemTime, UNIX_EPOCH};// 引入生成的 gRPC 代码
pub mod biz_activity {tonic::include_proto!("biz_activity");
}use biz_activity::{biz_activity_service_client::BizActivityServiceClient,BizActivity, GetBizActivityRequest, CreateBizActivityRequest, UpdateBizActivityRequest, DeleteBizActivityRequest, ListBizActivitiesRequest,
};pub struct BizActivityGrpcClient {client: BizActivityServiceClient<Channel>,
}impl BizActivityGrpcClient {pub async fn new(address: &str) -> Result<Self, Box<dyn std::error::Error>> {let client = BizActivityServiceClient::connect(address.to_string()).await?;Ok(BizActivityGrpcClient { client })}pub async fn create_activity(&mut self, activity: BizActivity) -> Result<BizActivity, tonic::Status> {let request = tonic::Request::new(CreateBizActivityRequest {activity: Some(activity),});let response = self.client.create_biz_activity(request).await?;Ok(response.into_inner().activity.unwrap())}pub async fn get_activity(&mut self, id: String) -> Result<Option<BizActivity>, tonic::Status> {let request = tonic::Request::new(GetBizActivityRequest { id });match self.client.get_biz_activity(request).await {Ok(response) => Ok(response.into_inner().activity),Err(status) => {if status.code() == tonic::Code::NotFound {Ok(None)} else {Err(status)}}}}pub async fn update_activity(&mut self, activity: BizActivity) -> Result<BizActivity, tonic::Status> {let request = tonic::Request::new(UpdateBizActivityRequest {activity: Some(activity),});let response = self.client.update_biz_activity(request).await?;Ok(response.into_inner().activity.unwrap())}pub async fn delete_activity(&mut self, id: String) -> Result<bool, tonic::Status> {let request = tonic::Request::new(DeleteBizActivityRequest { id });let response = self.client.delete_biz_activity(request).await?;Ok(response.into_inner().success)}pub async fn list_activities(&mut self, page: Option<i32>, size: Option<i32>) -> Result<Vec<BizActivity>, tonic::Status> {let request = tonic::Request::new(ListBizActivitiesRequest {page,size,});let response = self.client.list_biz_activities(request).await?;Ok(response.into_inner().activities)}
}pub async fn run_grpc_client_test() -> Result<(), Box<dyn std::error::Error>> {println!("连接到 gRPC 服务器...");let mut client = BizActivityGrpcClient::new("http://127.0.0.1:50051").await?;// 创建一个活动println!("创建业务活动...");let new_activity = BizActivity {id: Some("test_activity_1".to_string()),name: Some("测试活动".to_string()),status: Some(1),create_time: Some(SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_secs() as i64),additional_field: Some("额外信息".to_string()),};let created_activity = client.create_activity(new_activity).await?;println!("创建的活动: {:?}", created_activity.name);// 获取活动println!("获取业务活动...");if let Some(activity) = client.get_activity("test_activity_1".to_string()).await? {println!("获取的活动: {:?}", activity.name);} else {println!("未找到活动");}// 更新活动println!("更新业务活动...");let mut updated_activity = created_activity.clone();updated_activity.name = Some("更新后的活动".to_string());updated_activity.status = Some(2);let updated_activity = client.update_activity(updated_activity).await?;println!("更新后的活动: {:?}", updated_activity.name);// 列出活动println!("列出业务活动...");let activities = client.list_activities(Some(1), Some(10)).await?;println!("找到 {} 个活动", activities.len());for activity in &activities {println!(" - ID: {:?}, Name: {:?}", activity.id, activity.name);}// 删除活动println!("删除业务活动...");let deleted = client.delete_activity("test_activity_1".to_string()).await?;println!("删除结果: {}", deleted);// 再次尝试获取已删除的活动println!("尝试获取已删除的活动...");if let Some(activity) = client.get_activity("test_activity_1".to_string()).await? {println!("获取的活动: {:?}", activity.name);} else {println!("活动已被删除，未找到");}println!("gRPC 客户端测试完成！");Ok(())
}
lib
pub mod grpc_client;
pub mod grpc_server;
main
mod grpc_server;#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {println!("启动 gRPC 服务器...");// 启动 gRPC 服务器grpc_server::start_grpc_server().await?;Ok(())
}
client_test 测试客户端，对应下面的启动命令
use rbatiss::grpc_client;#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {println!("启动 gRPC 客户端测试...");// 运行客户端测试grpc_client::run_grpc_client_test().await?;Ok(())
}
启动命令
服务器 cargo run --bin 项目名
客户端 cargo run --bin client_test
