# gRPC 和协议缓冲区模式 ## 原始文件结构语法=“proto3”；包 orca.environment.v1; 选项 java_package = "orca.server.grpc.environment"; 选项 java_multiple_files = true; 导入“google/protobuf/timestamp.proto”；导入“google/protobuf/empty.proto”； // 服务定义服务环境服务{ // 一元 RPC rpc CreateEnvironment(CreateEnvironmentRequest) 返回 (Environment); rpc GetEnvironment(GetEnvironmentRequest) 返回 (Environment)； rpc ListEnvironments(ListEnvironmentsRequest) 返回 (ListEnvironmentsResponse)； rpc DeleteEnvironment(DeleteEnvironmentRequest) 返回 (google.protobuf.Empty); // 服务器流式传输 rpc WatchEnvironment（WatchEnvironmentRequest）返回（流EnvironmentEvent）； // 双向流 rpc StreamLogs(stream LogRequest) returns (stream LogResponse); } // 消息消息环境{ 字符串 ID = 1; 字符串名称=2；环境状态=3； google.protobuf.时间戳created_at = 4; google.protobuf.Timestamp Updated_at = 5；地图<字符串，字符串>标签= 6； } 枚举环境状态{ 环境_状态_未指定= 0； ENVIRONMENT_STATUS_PENDING = 1；环境状态运行=2； ENVIRONMENT_STATUS_STOPPED = 3；环境状态失败= 4； } 消息创建环境请求{ 字符串名称=1；字符串描述=2；地图<字符串，字符串>标签= 3； } 消息获取环境请求{ 字符串 ID = 1; } 消息列表环境请求{ int32 page_size = 1; 字符串 page_token = 2; 字符串过滤器=3； // 例如，“status=RUNNING” } 消息列表环境响应{ 重复环境environments = 1; 字符串 next_page_token = 2; int32 总计数 = 3; } 消息删除环境请求{ 字符串 ID = 1; } 消息 WatchEnvironmentRequest { 字符串 ID = 1; } 消息环境事件{ 环境环境=1；事件类型类型 = 2; google.protobuf.Timestamp 时间戳 = 3; 枚举事件类型{ 事件类型未指定= 0； EVENT_TYPE_CREATED = 1； EVENT_TYPE_UPDATED = 2； EVENT_TYPE_DELETED = 3； } } ## Spring gRPC 服务器实现 @GrpcService 类环境GrpcService( 私有val环境服务：EnvironmentService ) : EnvironmentServiceGrpcKt.EnvironmentServiceCoroutineImplBase() { 覆盖挂起乐趣createEnvironment（请求：创建环境请求 ): 环境 { val 结果 =environmentService.create(request.toDomain()) 返回结果.toProto() } 覆盖挂起乐趣 getEnvironment( 请求：获取环境请求 ): 环境 { val id = UUID.fromString(request.id) val env =environmentService.findById(id) ?: 抛出 StatusException(Status.NOT_FOUND.withDescription("未找到环境：${request.id}")) 返回 env.toProto() } 覆盖挂起乐趣列表环境（请求：列出环境请求 ): 列出环境响应 { val页面=environmentService.findAll( pageSize = 请求.pageSize, pageToken = request.pageToken.ifEmpty { null } ）返回 ListEnvironmentsResponse.newBuilder() .addAllEnvironments(page.items.map { it.toProto() }) .setNextPageToken(page.nextToken ?: "") .setTotalCount(页数.总计) .build() } 覆盖有趣的watchEnvironment（请求：WatchEnvironmentRequest ): Flow<环境事件> = flow { val id = UUID.fromString(request.id) environmentService.watchEvents(id).collect { 事件 -> 发出（事件.toProto（）） } } } ## gRPC 客户端实现 @组件类计算客户端（ @GrpcClient(“计算”) 私有 val 存根：ComputeServiceGrpcKt.ComputeServiceCoroutineStub ）{ 暂停乐趣createInstance（请求：CreateInstanceRequest）：实例{ 返回尝试{ 存根.createInstance(请求) } catch (e: StatusException) { 当（e.status.code）{ Status.Code.NOT_FOUND -> 抛出 ResourceNotFoundRestException("未找到实例模板") Status.Code.ALREADY_EXISTS -> 抛出 ConflictRestException("实例已存在") Status.Code.RESOURCE_EXHAUSTED -> 抛出 ServiceUnavailableException("超出配额") else -> throw InternalServerException("计算服务错误: ${e.message}") } } } 挂起 fun getInstance(id: String): 实例？ { 返回尝试{ Stub.getInstance(GetInstanceRequest.newBuilder().setId(id).build()) } catch (e: StatusException) { if (e.status.code == Status.Code.NOT_FOUND) null 否则抛出 e } } } ## gRPC 状态的错误处理 // 将域异常映射到 gRPC 状态 @GrpcAdvice 类 GrpcExceptionHandler { @GrpcExceptionHandler(ResourceNotFoundRestException::class) 有趣的handleNotFound（e：ResourceNotFoundRestException）：StatusException = 状态异常（状态.NOT_FOUND .withDescription(e.message) .withCause(e) ） @GrpcExceptionHandler(ValidationRestException::class) 有趣的handleValidation（e：ValidationRestException）：StatusException = 状态异常（状态.INVALID_ARGUMENT .withDescription(e.message) .withCause(e) ） @GrpcExceptionHandler(ConflictRestException::class) 有趣的handleConflict（e：ConflictRestException）：StatusException = 状态异常（状态.ALREADY_EXISTS .withDescription(e.message) .withCause(e) ） } ## 拦截器 // 用于记录/指标的服务器拦截器 @组件类 LoggingServerInterceptor : ServerInterceptor { 重写 fun 拦截Call( 调用：ServerCall, 标题：元数据，下一步：ServerCallHandler ): ServerCall.Listener { val method = call.methodDescriptor.fullMethodName val startTime = System.currentTimeMillis() logger.info("gRPC 调用开始：$method") 返回对象：ForwardingServerCallListener.SimpleForwardingServerCallListener( next.startCall（调用，标题））{ 重写 fun onComplete() { val 持续时间 = System.currentTimeMillis() - startTime logger.info("gRPC 调用完成：$method (${duration}ms)") super.onComplete() } } } } // 用于验证的客户端拦截器 @组件类 AuthClientInterceptor( private val tokenProvider: TokenProvider ) : 客户端拦截器 { 重写 fun 拦截Call( 方法：MethodDescriptor，呼叫选项：呼叫选项，下一篇：频道 ): ClientCall { 返回对象：ForwardingClientCall.SimpleForwardingClientCall( next.newCall(方法, callOptions) ）{ 覆盖 fun start(responseListener: Listener, headers: 元数据) { 标题.put( Metadata.Key.of("授权", Metadata.ASCII_STRING_MARSHALLER), “承载${tokenProvider.getToken()}” ） super.start(responseListener, headers) } } } } ## 原型到域的映射 // 域到原型有趣的 orca.server.domain.Environment.toProto(): 环境 = 环境.newBuilder() .setId(id.toString()) .setName(名称) .setStatus(status.toProto()) .setCreatedAt(createdAt.toProtoTimestamp()) .apply { updateAt?.let { setUpdatedAt(it.toProtoTimestamp()) } } .putAllLabels(标签) .build() 有趣的EnvironmentStatus.toProto(): orca.grpc.EnvironmentStatus = 当（这个）{ 环境状态.PENDING -> orca.grpc.EnvironmentStatus.ENVIRONMENT_STATUS_PENDING EnvironmentStatus.RUNNING -> orca.grpc.EnvironmentStatus.ENVIRONMENT_STATUS_RUNNING EnvironmentStatus.STOPPED -> orca.grpc.EnvironmentStatus.ENVIRONMENT_STATUS_STOPPED EnvironmentStatus.FAILED -> orca.grpc.EnvironmentStatus.ENVIRONMENT_STATUS_FAILED } 有趣的 Instant.toProtoTimestamp(): 时间戳 = 时间戳.newBuilder() .setSeconds(epochSecond) .setNanos(纳米) .build() // 原型到域有趣的 CreateEnvironmentRequest.toDomain() = CreateEnvironmentDomainRequest( 姓名=姓名，描述 = 描述.ifEmpty { null }, 标签=标签映射） ## 流模式 // 服务器流式传输重写 fun watchEnvironment(request: WatchRequest): Flow = flow { val 通道 = eventBus.subscribe(request.id) 尝试{ for（频道中的事件）{ 发出（事件.toProto（）） } }最后{ 频道.取消() } } // 双向流覆盖有趣的聊天（请求：Flow ）：Flow = flow { requests.collect { 消息 -> val 响应 = processMessage(消息) 发出（响应） } }

Grpc 协议缓冲区

包含技能