MultiProtocol gRPC API 文档

graph TB
subgraph "客户端层"
APP[应用程序]
CLIENT[gRPC客户端]
end
subgraph "网络层"
GATEWAY[网关/负载均衡器]
LB[负载均衡策略]
end
subgraph "服务端层"
SERVER[gRPC服务端]
SESSION[会话管理]
PROTO[协议处理器]
end
subgraph "存储层"
CACHE[缓存系统]
DB[(数据库)]
end
APP --> CLIENT
CLIENT --> GATEWAY
GATEWAY --> LB
LB --> SERVER
SERVER --> SESSION
SESSION --> PROTO
PROTO --> CACHE
PROTO --> DB

classDiagram
class ProtocolInterface {
<<service>>
+onDownlink(stream RequestMsg) stream ResponseMsg
}
class RequestMsg {
+int64 ts
+TracerProto tracer
+ConnectRequestMsg connectRequestMsg
+DownlinkRequestMessage downlinkRequestMessage
}
class ResponseMsg {
+TracerProto tracer
+ConnectResponseMsg connectResponseMsg
+DownlinkResponseMessage downlinkResponseMsg
}
class ConnectRequestMsg {
+string nodeId
}
class ConnectResponseMsg {
+ConnectResponseCode responseCode
+string errorMsg
}
class TracerProto {
+string id
+string origin
+int64 ts
}
ProtocolInterface --> RequestMsg : "接收"
ProtocolInterface --> ResponseMsg : "返回"
RequestMsg --> ConnectRequestMsg : "包含"
RequestMsg --> DownlinkRequestMessage : "包含"
ResponseMsg --> ConnectResponseMsg : "包含"
ResponseMsg --> DownlinkResponseMessage : "包含"
RequestMsg --> TracerProto : "追踪"
ResponseMsg --> TracerProto : "追踪"

flowchart TD
JAVA_OBJ["Java对象"] --> CONVERTER["ProtoConverter"]
CONVERTER --> PROTO_MSG["Protobuf消息"]
PROTO_MSG --> SERIALIZATION["序列化"]
SERIALIZATION --> NETWORK["网络传输"]
NETWORK --> DESERIALIZATION["反序列化"]
DESERIALIZATION --> PROTO_MSG_REV["Protobuf消息"]
PROTO_MSG_REV --> CONVERTER_REV["ProtoConverter"]
CONVERTER_REV --> JAVA_OBJ_REV["Java对象"]

classDiagram
class ProtoConverter {
<<utility>>
+toTracerProto() TracerProto
+toPricingModel(PricingModel) PricingModelProto
}
class PricingModel {
+PricingModelType type
+PricingModelRule rule
+BigDecimal standardElec
+BigDecimal standardServ
+Map~PricingModelFlag,FlagPrice~ flagPriceList
+Period[] periodsList
+TimePeriodItem[] timePeriodItems
}
class PricingModelProto {
+PricingModelType type
+PricingModelRule rule
+oneof pricing_config
}
class TracerProto {
+string id
+string origin
+int64 ts
}
ProtoConverter --> PricingModel : "转换"
ProtoConverter --> TracerProto : "创建"
PricingModel --> PricingModelProto : "转换"

classDiagram
class DownlinkGrpcClient {
-Map~HostAndPort,ManagedChannel~ channelMap
-Map~HostAndPort,StreamObserver~ inputStreamMap
-Map~HostAndPort,LinkedBlockingQueue~ queueMap
-Map~HostAndPort,ReentrantLock~ msgHandleLocksMap
-Map~HostAndPort,ExecutorService~ msgHandleExecutorMap
-Map~HostAndPort,AtomicInteger~ connectErrTimesMap
-Map~HostAndPort,Boolean~ initializedMap
+connect(HostAndPort) void
+sendDownlinkRequest(HostAndPort, RequestMsg) void
-handleMsgs(HostAndPort, LinkedBlockingQueue) void
}
class ManagedChannel {
+newBuilder() Builder
+shutdownNow() void
}
class StreamObserver {
+onNext(T) void
+onError(Throwable) void
+onCompleted() void
}
class RequestMsg {
+TracerProto tracer
+ConnectRequestMsg connectRequestMsg
+DownlinkRequestMessage downlinkRequestMessage
}
DownlinkGrpcClient --> ManagedChannel : "管理"
DownlinkGrpcClient --> StreamObserver : "使用"
DownlinkGrpcClient --> RequestMsg : "发送"

sequenceDiagram
participant App as 应用程序
participant Client as DownlinkGrpcClient
participant Queue as 消息队列
participant Channel as gRPC通道
participant Server as gRPC服务器
App->>Client : sendDownlinkRequest()
Client->>Queue : 添加到队列
loop 消息处理循环
Queue->>Client : 获取消息
Client->>Channel : 发送消息
Channel->>Server : onDownlink()
Server-->>Channel : 响应
Channel-->>Client : 处理响应
end

classDiagram
class DownlinkGrpcService {
-int grpcPort
-int grpcBoss
-int grpcWorker
-int grpcNettySoRcvbuf
-int grpcNettySoSndbuf
-boolean grpcNettyNoDelay
-int maxInboundMessageSize
-int maxConcurrentCallsPerConnection
-int clientMaxKeepAliveTimeSec
-Server server
-ReentrantLock replyLock
+init() void
+destroy() void
+onDownlink(StreamObserver) StreamObserver
}
class ProtocolInterfaceImplBase {
<<abstract>>
+onDownlink(StreamObserver) StreamObserver
}
class StreamObserver {
+onNext(T) void
+onError(Throwable) void
+onCompleted() void
}
class RequestMsg {
+TracerProto tracer
+ConnectRequestMsg connectRequestMsg
+DownlinkRequestMessage downlinkRequestMessage
}
class ResponseMsg {
+TracerProto tracer
+ConnectResponseMsg connectResponseMsg
+DownlinkResponseMessage downlinkResponseMsg
}
DownlinkGrpcService --|> ProtocolInterfaceImplBase
DownlinkGrpcService --> StreamObserver : "实现"
DownlinkGrpcService --> RequestMsg : "处理"
DownlinkGrpcService --> ResponseMsg : "返回"

flowchart TD
START([接收请求]) --> VALIDATE["验证请求格式"]
VALIDATE --> TRACE["设置追踪信息"]
TRACE --> CHECK_TYPE{"检查消息类型"}
CHECK_TYPE --> |连接请求| HANDLE_CONNECT["处理连接请求"]
CHECK_TYPE --> |下行请求| HANDLE_DOWNLINK["处理下行请求"]
HANDLE_CONNECT --> LOCK["获取锁"]
LOCK --> RESPOND["发送连接响应"]
RESPOND --> UNLOCK["释放锁"]
HANDLE_DOWNLINK --> CREATE_SESSION["创建协议会话"]
CREATE_SESSION --> PROCESS_CMD["处理命令"]
PROCESS_CMD --> LOG_RESULT["记录处理结果"]
UNLOCK --> END([处理完成])
LOG_RESULT --> END

flowchart TD
ERROR[捕获异常] --> CLASSIFY{"分类异常"}
CLASSIFY --> |网络异常| RETRY["重试机制"]
CLASSIFY --> |认证异常| AUTH["重新认证"]
CLASSIFY --> |业务异常| BUSINESS["业务处理"]
CLASSIFY --> |系统异常| FATAL["致命错误"]
RETRY --> BACKOFF["指数退避"]
BACKOFF --> MAX_RETRY{"达到最大重试次数?"}
MAX_RETRY --> |否| RETRY
MAX_RETRY --> |是| FALLBACK["降级处理"]
AUTH --> RECONNECT["重新连接"]
BUSINESS --> LOG_ERROR["记录错误"]
FATAL --> SHUTDOWN["优雅关闭"]
FALLBACK --> NOTIFY["通知监控"]
LOG_ERROR --> NOTIFY
RECONNECT --> NOTIFY
SHUTDOWN --> NOTIFY

graph LR
subgraph "连接生命周期"
A[连接建立] --> B[心跳检测]
B --> C[消息传输]
C --> D[空闲检测]
D --> E{是否空闲?}
E --> |是| F[保持连接]
E --> |否| C
F --> G[连接关闭]
end
subgraph "优化策略"
H[连接池] --> I[复用连接]
J[心跳保活] --> K[及时发现断连]
L[批量处理] --> M[提高吞吐量]
N[异步处理] --> O[降低延迟]
end

graph TB
subgraph "性能指标对比"
A[序列化速度] --> B[gRPC: 200MB/s<br/>REST: 100MB/s]
C[包体积] --> D[gRPC: 30%<br/>REST: 100%]
E[延迟] --> F[gRPC: 1ms<br/>REST: 10ms]
G[吞吐量] --> H[gRPC: 10K+ TPS<br/>REST: 1K+ TPS]
end
subgraph "资源消耗"
I[CPU使用] --> J[gRPC: 60%<br/>REST: 80%]
K[内存使用] --> L[gRPC: 70%<br/>REST: 90%]
M[网络带宽] --> N[gRPC: 50%<br/>REST: 100%]
end