mirror of
https://gitee.com/san-bing/JChargePointProtocol
synced 2026-07-18 04:37:55 +08:00
638 lines
22 KiB
Markdown
638 lines
22 KiB
Markdown
# 上行消息处理
|
||
|
||
<cite>
|
||
**本文档引用的文件**
|
||
- [YunKuaiChongProtocolMessageProcessor.java](file://jcpp-protocol-yunkuaichong/src/main/java/sanbing/jcpp/protocol/yunkuaichong/YunKuaiChongProtocolMessageProcessor.java)
|
||
- [ProtocolCmd.java](file://jcpp-protocol-api/src/main/java/sanbing/jcpp/protocol/annotation/ProtocolCmd.java)
|
||
- [ProtocolCommandRouter.java](file://jcpp-protocol-api/src/main/java/sanbing/jcpp/protocol/routing/ProtocolCommandRouter.java)
|
||
- [YunKuaiChongUplinkCmdExe.java](file://jcpp-protocol-yunkuaichong/src/main/java/sanbing/jcpp/protocol/yunkuaichong/YunKuaiChongUplinkCmdExe.java)
|
||
- [YunKuaiChongUplinkMessage.java](file://jcpp-protocol-yunkuaichong/src/main/java/sanbing/jcpp/protocol/yunkuaichong/YunKuaiChongUplinkMessage.java)
|
||
- [KafkaForwarder.java](file://jcpp-protocol-api/src/main/java/sanbing/jcpp/protocol/forwarder/KafkaForwarder.java)
|
||
- [YunKuaiChongV150HeartbeatULCmd.java](file://jcpp-protocol-yunkuaichong/src/main/java/sanbing/jcpp/protocol/yunkuaichong/v150/cmd/YunKuaiChongV150HeartbeatULCmd.java)
|
||
- [YunKuaiChongV150LoginULCmd.java](file://jcpp-protocol-yunkuaichong/src/main/java/sanbing/jcpp/protocol/yunkuaichong/v150/cmd/YunKuaiChongV150LoginULCmd.java)
|
||
- [YunKuaiChongV150StartChargeULCmd.java](file://jcpp-protocol-yunkuaichong/src/main/java/sanbing/jcpp/protocol/yunkuaichong/v150/cmd/YunKuaiChongV150StartChargeULCmd.java)
|
||
- [YunKuaiChongV150RealTimeDataULCmd.java](file://jcpp-protocol-yunkuaichong/src/main/java/sanbing/jcpp/protocol/yunkuaichong/v150/cmd/YunKuaiChongV150RealTimeDataULCmd.java)
|
||
- [YunKuaiChongV150TransactionRecordULCmd.java](file://jcpp-protocol-yunkuaichong/src/main/java/sanbing/jcpp/protocol/yunkuaichong/v150/cmd/YunKuaiChongV150TransactionRecordULCmd.java)
|
||
- [YunKuaiChongProtocolConstants.java](file://jcpp-protocol-yunkuaichong/src/main/java/sanbing/jcpp/protocol/yunkuaichong/YunKuaiChongProtocolConstants.java)
|
||
</cite>
|
||
|
||
## 目录
|
||
|
||
1. [概述](#概述)
|
||
2. [系统架构](#系统架构)
|
||
3. [消息处理流程](#消息处理流程)
|
||
4. [协议命令注解机制](#协议命令注解机制)
|
||
5. [关键上行命令详解](#关键上行命令详解)
|
||
6. [消息转发机制](#消息转发机制)
|
||
7. [数据持久化与状态管理](#数据持久化与状态管理)
|
||
8. [性能优化策略](#性能优化策略)
|
||
9. [总结](#总结)
|
||
|
||
## 概述
|
||
|
||
云快充协议上行消息处理系统是一个高度模块化的分布式消息处理框架，负责接收、解析、分发和处理来自充电桩的各种上行消息。该系统采用事件驱动架构，通过反射机制和命令路由模式实现灵活的消息处理。
|
||
|
||
核心特性包括：
|
||
|
||
- **高性能TCP监听**：基于Netty的异步I/O处理
|
||
- **智能协议解析**：支持多种协议版本的统一处理
|
||
- **反射命令路由**：基于注解的自动化命令分发
|
||
- **消息转发队列**：通过Kafka实现异步消息传递
|
||
- **状态管理**：实时更新充电桩状态和属性
|
||
|
||
## 系统架构
|
||
|
||
```mermaid
|
||
graph TB
|
||
subgraph "网络层"
|
||
TCP[TCP监听器]
|
||
Session[TCP会话管理]
|
||
end
|
||
subgraph "协议处理层"
|
||
Processor[云快充协议处理器]
|
||
Router[命令路由器]
|
||
Message[消息解析器]
|
||
end
|
||
subgraph "命令执行层"
|
||
Heartbeat[心跳命令]
|
||
Login[登录命令]
|
||
Charge[充电命令]
|
||
RealTime[实时数据]
|
||
Transaction[交易记录]
|
||
end
|
||
subgraph "消息转发层"
|
||
Forwarder[Kafka转发器]
|
||
Queue[消息队列]
|
||
end
|
||
subgraph "应用服务层"
|
||
App[jcpp-app服务]
|
||
DB[数据库存储]
|
||
end
|
||
TCP --> Session
|
||
Session --> Processor
|
||
Processor --> Router
|
||
Router --> Message
|
||
Message --> Heartbeat
|
||
Message --> Login
|
||
Message --> Charge
|
||
Message --> RealTime
|
||
Message --> Transaction
|
||
Heartbeat --> Forwarder
|
||
Login --> Forwarder
|
||
Charge --> Forwarder
|
||
RealTime --> Forwarder
|
||
Transaction --> Forwarder
|
||
Forwarder --> Queue
|
||
Queue --> App
|
||
App --> DB
|
||
```
|
||
|
||
**图表来源**
|
||
|
||
- [YunKuaiChongProtocolMessageProcessor.java](file://jcpp-protocol-yunkuaichong/src/main/java/sanbing/jcpp/protocol/yunkuaichong/YunKuaiChongProtocolMessageProcessor.java#L27-L61)
|
||
- [ProtocolCommandRouter.java](file://jcpp-protocol-api/src/main/java/sanbing/jcpp/protocol/routing/ProtocolCommandRouter.java#L25-L40)
|
||
|
||
## 消息处理流程
|
||
|
||
### TCP监听器接收数据
|
||
|
||
系统从TcpListener接收到原始数据包开始，首先进行基本的完整性检查：
|
||
|
||
```mermaid
|
||
flowchart TD
|
||
Start([接收数据包]) --> CheckLength{数据包长度检查}
|
||
CheckLength --> |长度不足| Reject[拒绝处理]
|
||
CheckLength --> |长度足够| CheckHeader{头部标识检查}
|
||
CheckHeader --> |头部无效| Reject
|
||
CheckHeader --> |头部有效| ParseHeader[解析协议头]
|
||
ParseHeader --> CheckBoundary{边界检查}
|
||
CheckBoundary --> |边界无效| Reject
|
||
CheckBoundary --> |边界有效| ParseFields[解析字段]
|
||
ParseFields --> CheckCRC{校验和验证}
|
||
CheckCRC --> |校验失败| Reject
|
||
CheckCRC --> |校验成功| BuildMessage[构建消息对象]
|
||
BuildMessage --> RouteCommand[路由命令]
|
||
RouteCommand --> ExecuteCommand[执行命令]
|
||
ExecuteCommand --> End([处理完成])
|
||
Reject --> End
|
||
```
|
||
|
||
**图表来源**
|
||
|
||
- [YunKuaiChongProtocolMessageProcessor.java](file://jcpp-protocol-yunkuaichong/src/main/java/sanbing/jcpp/protocol/yunkuaichong/YunKuaiChongProtocolMessageProcessor.java#L63-L154)
|
||
|
||
### 消息解析与验证
|
||
|
||
协议处理器执行严格的验证流程：
|
||
|
||
1. **长度验证**：确保数据包包含最小长度（8字节）
|
||
2. **头部验证**：检查起始标识符（0x68）
|
||
3. **边界检查**：验证数据长度字段的合理性
|
||
4. **CRC校验**：双重校验和验证（小端和大端）
|
||
|
||
**章节来源**
|
||
|
||
- [YunKuaiChongProtocolMessageProcessor.java](file://jcpp-protocol-yunkuaichong/src/main/java/sanbing/jcpp/protocol/yunkuaichong/YunKuaiChongProtocolMessageProcessor.java#L63-L154)
|
||
|
||
### 命令路由与分发
|
||
|
||
解析完成后，消息通过命令路由器进行分发：
|
||
|
||
```mermaid
|
||
sequenceDiagram
|
||
participant Processor as 协议处理器
|
||
participant Router as 命令路由器
|
||
participant Executor as 命令执行器
|
||
participant Forwarder as 消息转发器
|
||
Processor->>Router : 查找命令执行器
|
||
Router->>Router : 构建路由键(protocol : cmd)
|
||
Router->>Router : 查询执行器映射表
|
||
Router-->>Processor : 返回执行器实例
|
||
Processor->>Executor : 调用execute方法
|
||
Executor->>Executor : 处理业务逻辑
|
||
Executor->>Forwarder : 发送消息到队列
|
||
Forwarder-->>Executor : 确认发送
|
||
Executor-->>Processor : 返回处理结果
|
||
```
|
||
|
||
**图表来源**
|
||
|
||
- [YunKuaiChongProtocolMessageProcessor.java](file://jcpp-protocol-yunkuaichong/src/main/java/sanbing/jcpp/protocol/yunkuaichong/YunKuaiChongProtocolMessageProcessor.java#L170-L184)
|
||
- [ProtocolCommandRouter.java](file://jcpp-protocol-api/src/main/java/sanbing/jcpp/protocol/routing/ProtocolCommandRouter.java#L85-L95)
|
||
|
||
**章节来源**
|
||
|
||
- [YunKuaiChongProtocolMessageProcessor.java](file://jcpp-protocol-yunkuaichong/src/main/java/sanbing/jcpp/protocol/yunkuaichong/YunKuaiChongProtocolMessageProcessor.java#L170-L184)
|
||
|
||
## 协议命令注解机制
|
||
|
||
### @ProtocolCmd注解设计
|
||
|
||
系统采用基于注解的命令映射机制，通过`@ProtocolCmd`注解实现声明式命令注册：
|
||
|
||
```mermaid
|
||
classDiagram
|
||
class ProtocolCmd {
|
||
+int value()
|
||
+String[] protocolNames()
|
||
}
|
||
class ProtocolCommandRouter {
|
||
-Map~String,T~ executorMap
|
||
+ProtocolCommandRouter(Class, Predicate)
|
||
+getExecutor(String, int) T
|
||
-initializeRoutes(Class, Predicate)
|
||
-registerExecutor(Class)
|
||
}
|
||
class YunKuaiChongUplinkCmdExe {
|
||
+execute(TcpSession, YunKuaiChongUplinkMessage, ProtocolContext)
|
||
#uplinkMessageBuilder(String, TcpSession, YunKuaiChongUplinkMessage)
|
||
}
|
||
ProtocolCmd --> YunKuaiChongUplinkCmdExe : 注解标记
|
||
ProtocolCommandRouter --> YunKuaiChongUplinkCmdExe : 管理实例
|
||
```
|
||
|
||
**图表来源**
|
||
|
||
- [ProtocolCmd.java](file://jcpp-protocol-api/src/main/java/sanbing/jcpp/protocol/annotation/ProtocolCmd.java#L20-L32)
|
||
- [ProtocolCommandRouter.java](file://jcpp-protocol-api/src/main/java/sanbing/jcpp/protocol/routing/ProtocolCommandRouter.java#L25-L40)
|
||
|
||
### 反射机制工作原理
|
||
|
||
命令路由器通过以下步骤实现反射机制：
|
||
|
||
1. **类扫描**：扫描指定包路径下带有`@ProtocolCmd`注解的类
|
||
2. **注解解析**：提取命令字和协议名称数组
|
||
3. **实例化**：动态创建命令执行器实例
|
||
4. **路由注册**：构建`protocol:cmd`键值对存储到映射表
|
||
|
||
**章节来源**
|
||
|
||
- [ProtocolCommandRouter.java](file://jcpp-protocol-api/src/main/java/sanbing/jcpp/protocol/routing/ProtocolCommandRouter.java#L42-L85)
|
||
|
||
### 多版本协议支持
|
||
|
||
系统通过协议名称数组支持多个协议版本的统一处理：
|
||
|
||
| 命令类型 | 协议版本 | 命令字 | 功能描述 |
|
||
|------|----------------|------|---------|
|
||
| 心跳命令 | V150/V160/V170 | 0x03 | 充电桩状态报告 |
|
||
| 登录命令 | V150/V160/V170 | 0x01 | 充电桩认证注册 |
|
||
| 充电启动 | V150/V160/V170 | 0x31 | 主动充电请求 |
|
||
| 实时数据 | V150/V160/V170 | 0x13 | 充电过程监控 |
|
||
| 交易记录 | V150/V160 | 0x3B | 充电完成记录 |
|
||
|
||
**章节来源**
|
||
|
||
- [YunKuaiChongProtocolConstants.java](file://jcpp-protocol-yunkuaichong/src/main/java/sanbing/jcpp/protocol/yunkuaichong/YunKuaiChongProtocolConstants.java#L25-L40)
|
||
|
||
## 关键上行命令详解
|
||
|
||
### 心跳包处理
|
||
|
||
心跳命令（0x03）负责维持连接状态和报告充电桩状态：
|
||
|
||
```mermaid
|
||
sequenceDiagram
|
||
participant Pile as 充电桩
|
||
participant Processor as 协议处理器
|
||
participant HeartbeatCmd as 心跳命令
|
||
participant Session as 会话管理
|
||
participant Forwarder as 消息转发器
|
||
Pile->>Processor : 发送心跳数据
|
||
Processor->>HeartbeatCmd : 路由到心跳处理器
|
||
HeartbeatCmd->>HeartbeatCmd : 解析桩编号和枪状态
|
||
HeartbeatCmd->>Session : 刷新会话激活状态
|
||
HeartbeatCmd->>HeartbeatCmd : 构建心跳请求消息
|
||
HeartbeatCmd->>Forwarder : 发送到Kafka队列
|
||
HeartbeatCmd->>Pile : 发送心跳响应
|
||
```
|
||
|
||
**图表来源**
|
||
|
||
- [YunKuaiChongV150HeartbeatULCmd.java](file://jcpp-protocol-yunkuaichong/src/main/java/sanbing/jcpp/protocol/yunkuaichong/v150/cmd/YunKuaiChongV150HeartbeatULCmd.java#L30-L84)
|
||
|
||
**业务逻辑特点**：
|
||
|
||
- **状态更新**：刷新会话激活时间
|
||
- **响应机制**：自动回复心跳确认
|
||
- **信息提取**：解析桩编号、枪号、状态等关键信息
|
||
- **转发处理**：将心跳信息发送到后端服务
|
||
|
||
**章节来源**
|
||
|
||
- [YunKuaiChongV150HeartbeatULCmd.java](file://jcpp-protocol-yunkuaichong/src/main/java/sanbing/jcpp/protocol/yunkuaichong/v150/cmd/YunKuaiChongV150HeartbeatULCmd.java#L30-L84)
|
||
|
||
### 登录请求处理
|
||
|
||
登录命令（0x01）负责充电桩的身份认证和会话建立：
|
||
|
||
```mermaid
|
||
flowchart TD
|
||
LoginReq[登录请求] --> ParseInfo[解析设备信息]
|
||
ParseInfo --> ExtractPileCode[提取桩编号]
|
||
ExtractPileCode --> ExtractType[提取桩类型]
|
||
ExtractType --> ExtractGuns[提取枪数量]
|
||
ExtractGuns --> ExtractVersion[提取软件版本]
|
||
ExtractVersion --> ExtractSIM[提取SIM卡信息]
|
||
ExtractSIM --> RegisterSession[注册会话]
|
||
RegisterSession --> BuildLoginMsg[构建登录消息]
|
||
BuildLoginMsg --> ForwardToBackend[转发到后端]
|
||
ForwardToBackend --> Complete[处理完成]
|
||
```
|
||
|
||
**图表来源**
|
||
|
||
- [YunKuaiChongV150LoginULCmd.java](file://jcpp-protocol-yunkuaichong/src/main/java/sanbing/jcpp/protocol/yunkuaichong/v150/cmd/YunKuaiChongV150LoginULCmd.java#L30-L84)
|
||
|
||
**认证流程**：
|
||
|
||
1. **身份验证**：通过桩编号验证设备合法性
|
||
2. **会话注册**：建立和维护设备会话状态
|
||
3. **信息收集**：收集设备硬件和软件信息
|
||
4. **后端通知**：通知应用服务新设备上线
|
||
|
||
**章节来源**
|
||
|
||
- [YunKuaiChongV150LoginULCmd.java](file://jcpp-protocol-yunkuaichong/src/main/java/sanbing/jcpp/protocol/yunkuaichong/v150/cmd/YunKuaiChongV150LoginULCmd.java#L30-L84)
|
||
|
||
### 启动充电处理
|
||
|
||
充电启动命令（0x31）处理充电桩主动发起的充电请求：
|
||
|
||
```mermaid
|
||
flowchart TD
|
||
StartCharge[充电启动请求] --> ParseBasic[解析基础信息]
|
||
ParseBasic --> ParseCard[解析卡片信息]
|
||
ParseCard --> ParseVIN[解析VIN码]
|
||
ParseVIN --> ValidatePassword{需要密码?}
|
||
ValidatePassword --> |是| HashPassword[密码哈希处理]
|
||
ValidatePassword --> |否| SkipPassword[跳过密码处理]
|
||
HashPassword --> ReverseVIN[反转VIN码]
|
||
SkipPassword --> ReverseVIN
|
||
ReverseVIN --> BuildRequest[构建启动请求]
|
||
BuildRequest --> ForwardMessage[转发消息]
|
||
ForwardMessage --> Complete[处理完成]
|
||
```
|
||
|
||
**图表来源**
|
||
|
||
- [YunKuaiChongV150StartChargeULCmd.java](file://jcpp-protocol-yunkuaichong/src/main/java/sanbing/jcpp/protocol/yunkuaichong/v150/cmd/YunKuaiChongV150StartChargeULCmd.java#L35-L124)
|
||
|
||
**处理要点**：
|
||
|
||
- **卡片验证**：处理物理卡片号和密码验证
|
||
- **VIN码处理**：特殊格式的VIN码反转处理
|
||
- **启动方式**：支持多种启动方式（APP、卡片、离线卡等）
|
||
- **安全机制**：密码MD5哈希处理
|
||
|
||
**章节来源**
|
||
|
||
- [YunKuaiChongV150StartChargeULCmd.java](file://jcpp-protocol-yunkuaichong/src/main/java/sanbing/jcpp/protocol/yunkuaichong/v150/cmd/YunKuaiChongV150StartChargeULCmd.java#L35-L124)
|
||
|
||
### 实时数据处理
|
||
|
||
实时数据命令（0x13）负责传输充电过程中的监控数据：
|
||
|
||
```mermaid
|
||
classDiagram
|
||
class RealTimeDataProcessor {
|
||
+execute(TcpSession, YunKuaiChongUplinkMessage, ProtocolContext)
|
||
-parseGunStatus(int, int, String) GunRunStatus
|
||
-parseFaults(byte[]) boolean[]
|
||
-getFaultDescriptions(boolean[]) String[]
|
||
-reduceMagnification(long, int) BigDecimal
|
||
}
|
||
class FaultDescription {
|
||
+急停按钮动作故障
|
||
+无可用整流模块
|
||
+出风口温度过高
|
||
+交流防雷故障
|
||
+交直流模块通信中断
|
||
+绝缘检测模块通信中断
|
||
+电度表通信中断
|
||
+读卡器通信中断
|
||
+RC10通信中断
|
||
+风扇调速板故障
|
||
+直流熔断器故障
|
||
+高压接触器故障
|
||
+门打开
|
||
}
|
||
RealTimeDataProcessor --> FaultDescription : 使用
|
||
```
|
||
|
||
**图表来源**
|
||
|
||
- [YunKuaiChongV150RealTimeDataULCmd.java](file://jcpp-protocol-yunkuaichong/src/main/java/sanbing/jcpp/protocol/yunkuaichong/v150/cmd/YunKuaiChongV150RealTimeDataULCmd.java#L35-L239)
|
||
|
||
**数据处理能力**：
|
||
|
||
- **状态解析**：解析充电枪的各种运行状态
|
||
- **故障检测**：14种硬件故障的bit位解析
|
||
- **数值转换**：精密的数值放大倍率处理
|
||
- **多消息转发**：同时发送状态和进度消息
|
||
|
||
**章节来源**
|
||
|
||
- [YunKuaiChongV150RealTimeDataULCmd.java](file://jcpp-protocol-yunkuaichong/src/main/java/sanbing/jcpp/protocol/yunkuaichong/v150/cmd/YunKuaiChongV150RealTimeDataULCmd.java#L35-L239)
|
||
|
||
### 交易记录处理
|
||
|
||
交易记录命令（0x3B）处理充电完成后的结算信息：
|
||
|
||
```mermaid
|
||
flowchart TD
|
||
TransactionRecord[交易记录] --> ParseTradeNo[解析交易流水号]
|
||
ParseTradeNo --> ParsePileGun[解析桩号和枪号]
|
||
ParsePileGun --> ParseTimes[解析时间信息]
|
||
ParseTimes --> ParseEnergy[解析电量数据]
|
||
ParseEnergy --> ParseAmount[解析金额信息]
|
||
ParseAmount --> ParseMeter[解析电表数据]
|
||
ParseMeter --> ParseVIN[解析VIN码]
|
||
ParseVIN --> ParseStartFlag[解析启动方式]
|
||
ParseStartFlag --> ParseStopReason[解析停止原因]
|
||
ParseStopReason --> ParseCard[解析卡片信息]
|
||
ParseCard --> BuildTransaction[构建交易记录]
|
||
BuildTransaction --> ForwardToBackend[转发到后端]
|
||
```
|
||
|
||
**图表来源**
|
||
|
||
- [YunKuaiChongV150TransactionRecordULCmd.java](file://jcpp-protocol-yunkuaichong/src/main/java/sanbing/jcpp/protocol/yunkuaichong/v150/cmd/YunKuaiChongV150TransactionRecordULCmd.java#L35-L305)
|
||
|
||
**结算信息完整性**：
|
||
|
||
- **分时电价**：支持尖峰平谷四种电价模式
|
||
- **电量统计**：精确到4位小数的电量计量
|
||
- **费用明细**：详细的电费和服务费计算
|
||
- **异常处理**：128种充电异常原因的完整映射
|
||
|
||
**章节来源**
|
||
|
||
- [YunKuaiChongV150TransactionRecordULCmd.java](file://jcpp-protocol-yunkuaichong/src/main/java/sanbing/jcpp/protocol/yunkuaichong/v150/cmd/YunKuaiChongV150TransactionRecordULCmd.java#L35-L305)
|
||
|
||
## 消息转发机制
|
||
|
||
### Kafka转发器架构
|
||
|
||
系统采用Kafka作为消息中间件，实现异步消息传递：
|
||
|
||
```mermaid
|
||
graph TB
|
||
subgraph "消息生产者"
|
||
Processor[协议处理器]
|
||
CmdExe[命令执行器]
|
||
end
|
||
subgraph "Kafka集群"
|
||
Broker1[Kafka Broker 1]
|
||
Broker2[Kafka Broker 2]
|
||
Broker3[Kafka Broker 3]
|
||
end
|
||
subgraph "消息消费者"
|
||
Consumer1[jcpp-app服务1]
|
||
Consumer2[jcpp-app服务2]
|
||
Consumer3[jcpp-app服务3]
|
||
end
|
||
Processor --> CmdExe
|
||
CmdExe --> KafkaForwarder[Kafka转发器]
|
||
KafkaForwarder --> Broker1
|
||
KafkaForwarder --> Broker2
|
||
KafkaForwarder --> Broker3
|
||
Broker1 --> Consumer1
|
||
Broker2 --> Consumer2
|
||
Broker3 --> Consumer3
|
||
```
|
||
|
||
**图表来源**
|
||
|
||
- [KafkaForwarder.java](file://jcpp-protocol-api/src/main/java/sanbing/jcpp/protocol/forwarder/KafkaForwarder.java#L40-L80)
|
||
|
||
### 消息序列化与传输
|
||
|
||
转发器支持多种消息格式和传输模式：
|
||
|
||
| 传输模式 | 格式类型 | 性能特点 | 使用场景 |
|
||
|-----------|----------|------|-------|
|
||
| Monolith | Protobuf | 高性能 | 单体部署 |
|
||
| Partition | JSON | 易调试 | 分布式部署 |
|
||
| Custom | 自定义 | 灵活扩展 | 特殊需求 |
|
||
|
||
**章节来源**
|
||
|
||
- [KafkaForwarder.java](file://jcpp-protocol-api/src/main/java/sanbing/jcpp/protocol/forwarder/KafkaForwarder.java#L120-L180)
|
||
|
||
### 消息路由策略
|
||
|
||
系统实现了智能的消息路由机制：
|
||
|
||
```mermaid
|
||
flowchart TD
|
||
Message[上行消息] --> CheckMode{检查传输模式}
|
||
CheckMode --> |Monolith| DirectSend[直接发送]
|
||
CheckMode --> |Partition| PartitionSend[分区发送]
|
||
CheckMode --> |Custom| CustomSend[自定义发送]
|
||
DirectSend --> MonolithQueue[单体队列]
|
||
PartitionSend --> PartitionKey[分区键]
|
||
PartitionKey --> KafkaTopic[Kafka主题]
|
||
CustomSend --> CustomLogic[自定义逻辑]
|
||
MonolithQueue --> Consumer[jcpp-app消费者]
|
||
KafkaTopic --> Consumer
|
||
CustomLogic --> Consumer
|
||
```
|
||
|
||
**图表来源**
|
||
|
||
- [KafkaForwarder.java](file://jcpp-protocol-api/src/main/java/sanbing/jcpp/protocol/forwarder/KafkaForwarder.java#L120-L180)
|
||
|
||
## 数据持久化与状态管理
|
||
|
||
### 属性系统架构
|
||
|
||
系统采用属性（Attribute）系统进行数据持久化：
|
||
|
||
```mermaid
|
||
erDiagram
|
||
ENTITY {
|
||
uuid id PK
|
||
string entity_type
|
||
timestamp created_at
|
||
timestamp last_update_ts
|
||
}
|
||
ATTRIBUTE {
|
||
uuid entity_id FK
|
||
string attr_key
|
||
string str_value
|
||
double dbl_value
|
||
bigint long_value
|
||
boolean bool_value
|
||
json json_value
|
||
timestamp last_update_ts
|
||
int version
|
||
}
|
||
ENTITY ||--o{ ATTRIBUTE : contains
|
||
```
|
||
|
||
**图表来源**
|
||
|
||
- [DefaultAttributeRepository.java](file://jcpp-app/src/main/java/sanbing/jcpp/app/dal/repository/attribute/DefaultAttributeRepository.java#L112-L142)
|
||
|
||
### 状态更新流程
|
||
|
||
系统实现了高效的状态更新机制：
|
||
|
||
```mermaid
|
||
sequenceDiagram
|
||
participant Cmd as 命令处理器
|
||
participant AttrSvc as 属性服务
|
||
participant Repo as 属性仓库
|
||
participant Queue as 批量队列
|
||
participant DB as 数据库
|
||
Cmd->>AttrSvc : 请求状态更新
|
||
AttrSvc->>AttrSvc : 构建属性列表
|
||
AttrSvc->>Repo : 添加到批量队列
|
||
Repo->>Queue : 排队等待
|
||
Queue->>DB : 批量插入/更新
|
||
DB-->>Queue : 确认写入
|
||
Queue-->>Repo : 返回版本号
|
||
Repo-->>AttrSvc : 返回操作结果
|
||
AttrSvc-->>Cmd : 返回更新状态
|
||
```
|
||
|
||
**图表来源**
|
||
|
||
- [DefaultAttributeRepository.java](file://jcpp-app/src/main/java/sanbing/jcpp/app/dal/repository/attribute/DefaultAttributeRepository.java#L71-L110)
|
||
|
||
### 批量处理优化
|
||
|
||
系统采用批量处理策略提升性能：
|
||
|
||
| 优化策略 | 实现方式 | 性能提升 |
|
||
|------|------------------|-----------|
|
||
| 批量队列 | SqlBlockingQueue | 减少数据库连接开销 |
|
||
| 异步处理 | Future模式 | 提升并发处理能力 |
|
||
| 版本控制 | 冲突检测 | 确保数据一致性 |
|
||
| 缓存机制 | 多级缓存 | 减少重复查询 |
|
||
|
||
**章节来源**
|
||
|
||
- [DefaultAttributeRepository.java](file://jcpp-app/src/main/java/sanbing/jcpp/app/dal/repository/attribute/DefaultAttributeRepository.java#L71-L110)
|
||
|
||
## 性能优化策略
|
||
|
||
### 异步处理架构
|
||
|
||
系统采用完全异步的处理架构：
|
||
|
||
```mermaid
|
||
graph LR
|
||
subgraph "网络层"
|
||
Netty[Netty I/O线程]
|
||
end
|
||
subgraph "处理层"
|
||
Processor[协议处理器]
|
||
Router[命令路由器]
|
||
Executor[命令执行器]
|
||
end
|
||
subgraph "异步层"
|
||
AsyncPool[异步线程池]
|
||
Future[Future模式]
|
||
end
|
||
subgraph "存储层"
|
||
DB[数据库]
|
||
Cache[缓存]
|
||
end
|
||
Netty --> Processor
|
||
Processor --> Router
|
||
Router --> Executor
|
||
Executor --> AsyncPool
|
||
AsyncPool --> Future
|
||
Future --> DB
|
||
Future --> Cache
|
||
```
|
||
|
||
### 内存优化策略
|
||
|
||
系统实现了多层次的内存优化：
|
||
|
||
| 优化层级 | 策略 | 效果 |
|
||
|------|---------------|---------|
|
||
| 对象池 | ByteBuf复用 | 减少GC压力 |
|
||
| 缓存 | Caffeine本地缓存 | 提升访问速度 |
|
||
| 序列化 | Protobuf二进制格式 | 减少序列化开销 |
|
||
| 批量 | 批量操作 | 减少网络往返 |
|
||
|
||
### 并发控制机制
|
||
|
||
系统采用多种并发控制策略：
|
||
|
||
```mermaid
|
||
flowchart TD
|
||
Request[请求到达] --> Lock{获取锁}
|
||
Lock --> |成功| Process[处理请求]
|
||
Lock --> |失败| Queue[加入队列]
|
||
Process --> Release[释放锁]
|
||
Queue --> Wait[等待执行]
|
||
Wait --> Process
|
||
Release --> Complete[完成处理]
|
||
```
|
||
|
||
## 总结
|
||
|
||
云快充协议上行消息处理系统是一个高度优化的分布式消息处理框架，具有以下核心优势：
|
||
|
||
### 技术亮点
|
||
|
||
1. **模块化设计**：通过注解驱动的命令路由实现高度可扩展的架构
|
||
2. **高性能处理**：基于Netty的异步I/O和批量处理策略
|
||
3. **强一致性**：通过事务性和批量操作保证数据一致性
|
||
4. **可观测性**：完善的日志记录和监控指标体系
|
||
|
||
### 业务价值
|
||
|
||
1. **实时响应**：毫秒级的消息处理延迟
|
||
2. **高可靠性**：多重校验和容错机制
|
||
3. **可扩展性**：支持多种协议版本和未来扩展
|
||
4. **运维友好**：清晰的日志和监控界面
|
||
|
||
### 应用场景
|
||
|
||
该系统适用于大规模充电桩管理平台，能够处理数万甚至数十万充电桩的并发消息处理需求，为电动车充电服务提供稳定可靠的技术支撑。 |