mirror of https://gitee.com/san-bing/JChargePointProtocol synced 2026-05-05 02:19:56 +08:00

Files

三丙 fa9524d302 wiki生成

2025-10-28 14:39:06 +08:00

19 KiB

Raw Blame History

消息队列集成

**本文档引用的文件** - [KafkaAppQueueFactory.java](file://jcpp-infrastructure-queue/src/main/java/sanbing/jcpp/infrastructure/queue/provider/KafkaAppQueueFactory.java) - [AppQueueFactory.java](file://jcpp-infrastructure-queue/src/main/java/sanbing/jcpp/infrastructure/queue/provider/AppQueueFactory.java) - [KafkaConsumerTemplate.java](file://jcpp-infrastructure-queue/src/main/java/sanbing/jcpp/infrastructure/queue/kafka/KafkaConsumerTemplate.java) - [KafkaProducerTemplate.java](file://jcpp-infrastructure-queue/src/main/java/sanbing/jcpp/infrastructure/queue/kafka/KafkaProducerTemplate.java) - [ProtocolUplinkConsumerService.java](file://jcpp-app/src/main/java/sanbing/jcpp/app/service/queue/consumer/ProtocolUplinkConsumerService.java) - [KafkaSettings.java](file://jcpp-infrastructure-queue/src/main/java/sanbing/jcpp/infrastructure/queue/kafka/KafkaSettings.java) - [KafkaTopicConfigs.java](file://jcpp-infrastructure-queue/src/main/java/sanbing/jcpp/infrastructure/queue/kafka/KafkaTopicConfigs.java) - [ProtoQueueMsg.java](file://jcpp-infrastructure-queue/src/main/java/sanbing/jcpp/infrastructure/queue/ProtoQueueMsg.java) - [DefaultQueueMsgHeaders.java](file://jcpp-infrastructure-queue/src/main/java/sanbing/jcpp/infrastructure/queue/DefaultQueueMsgHeaders.java) - [QueueAppSettings.java](file://jcpp-infrastructure-queue/src/main/java/sanbing/jcpp/infrastructure/queue/settings/QueueAppSettings.java) - [KafkaConsumerStatsService.java](file://jcpp-infrastructure-queue/src/main/java/sanbing/jcpp/infrastructure/queue/kafka/KafkaConsumerStatsService.java) - [ProtocolUplinkMsg.java](file://jcpp-protocol-api/src/main/java/sanbing/jcpp/protocol/domain/ProtocolUplinkMsg.java) - [uplink.proto](file://jcpp-infrastructure-proto/src/main/proto/uplink.proto) - [app-service.yml](file://jcpp-app-bootstrap/src/main/resources/app-service.yml) - [protocol-service.yml](file://jcpp-protocol-bootstrap/src/main/resources/protocol-service.yml)

引言

JChargePointProtocol系统采用Kafka作为核心消息中间件，构建了一个高可用、高性能的消息队列集成架构。该系统通过消息队列实现了协议处理服务和业务应用服务之间的解耦，确保了系统的可扩展性和可靠性。

本文档详细阐述了系统如何选择Kafka作为消息中间件，以及Kafka在整体架构中的关键作用，包括消息的生产、消费、序列化、可靠性保证等各个方面。

系统架构概述

系统采用分层架构设计，通过消息队列实现不同服务组件之间的解耦：

graph TB
subgraph "协议处理层"
PC[协议客户端<br/>TCP连接]
PL[协议监听器<br/>TcpListener]
PM[协议消息处理器<br/>ProtocolMessageProcessor]
PF[协议转发器<br/>Forwarder]
end
subgraph "消息队列层"
KQF[KafkaAppQueueFactory]
KP[KafkaProducerTemplate]
KC[KafkaConsumerTemplate]
KT[Kafka主题<br/>protocol_uplink]
end
subgraph "业务应用层"
PUCS[ProtocolUplinkConsumerService]
PS[业务处理器<br/>PileProtocolService]
AS[应用服务<br/>其他服务]
end
PC --> PL
PL --> PM
PM --> PF
PF --> KQF
KQF --> KP
KP --> KT
KT --> KC
KC --> PUCS
PUCS --> PS
PS --> AS

图表来源

Kafka作为核心消息中间件的选择

选择Kafka的核心原因

高吞吐量和低延迟：Kafka能够处理每秒数百万条消息，满足充电桩高频数据传输需求
持久化存储：消息持久化保证数据不丢失，支持离线处理和重放
分布式架构：天然支持水平扩展，适应大规模部署需求
分区和副本机制：提供高可用性和数据冗余
流式处理能力：支持实时数据处理和分析

解耦协议处理和服务业务

Kafka在系统中扮演着关键的解耦角色：

sequenceDiagram
participant P as 协议处理服务
participant K as Kafka队列
participant A as 应用业务服务
P->>K : 生产消息充电桩上行数据
Note over P,K : 异步处理，无需等待
K->>K : 持久化存储
K->>A : 消费消息
A->>A : 处理业务逻辑
A->>K : 确认消费成功

图表来源

章节来源

KafkaAppQueueFactory设计与实现

工厂模式的应用

KafkaAppQueueFactory实现了AppQueueFactory接口，提供了Kafka相关队列组件的创建功能：

classDiagram
class AppQueueFactory {
<<interface>>
+createProtocolUplinkMsgConsumer() QueueConsumer
+createProtocolUplinkMsgProducer(String) QueueProducer
}
class KafkaAppQueueFactory {
-KafkaSettings kafkaSettings
-QueueAppSettings appSettings
-KafkaConsumerStatsService consumerStatsService
-ServiceInfoProvider serviceInfoProvider
-QueueAdmin appAdmin
+createProtocolUplinkMsgConsumer() QueueConsumer
+createProtocolUplinkMsgProducer(String) QueueProducer
-destroy() void
}
class KafkaConsumerTemplate {
-KafkaConsumer consumer
-KafkaDecoder decoder
-KafkaConsumerStatsService statsService
-String groupId
+decode(ConsumerRecord) T
+doPoll(long) List
+doCommit() void
}
class KafkaProducerTemplate {
-KafkaProducer producer
-String topic
-String clientId
-QueueAdmin admin
+send(TopicPartitionInfo, T, QueueCallback) void
+addAnalyticHeaders(List) void
}
AppQueueFactory <|-- KafkaAppQueueFactory
KafkaAppQueueFactory --> KafkaConsumerTemplate
KafkaAppQueueFactory --> KafkaProducerTemplate

图表来源

生产者和消费者的创建

ProtocolUplinkMsgConsumer的创建

KafkaAppQueueFactory根据配置创建ProtocolUplinkMsgConsumer，支持多种解码方式：

Protobuf解码：默认方式，高效且类型安全
JSON解码：兼容性更好，便于调试

ProtocolUplinkMsgProducer的创建

生产者负责将消息发送到指定的Kafka主题，支持动态主题切换和批量发送优化。

章节来源

ProtocolUplinkConsumerService详解

消费者服务的核心职责

ProtocolUplinkConsumerService是系统中处理充电桩上行消息的主要组件，负责：

消息订阅：订阅Kafka主题，接收来自协议处理服务的消息
消息分发：根据消息类型将消息路由到相应的业务处理器
并发处理：支持多线程并发处理提高吞吐量
错误处理：提供完善的异常处理和重试机制

消息处理流程

flowchart TD
Start([开始消费]) --> Poll[Poll消息]
Poll --> Decode[解码消息]
Decode --> Validate{消息验证}
Validate --> |有效| Process[处理消息]
Validate --> |无效| Error[记录错误]
Process --> Route{路由判断}
Route --> |登录请求| Login[处理登录]
Route --> |心跳| Heartbeat[处理心跳]
Route --> |充电状态| Charge[处理充电状态]
Route --> |其他| Other[处理其他类型]
Login --> Callback[回调处理]
Heartbeat --> Callback
Charge --> Callback
Other --> Callback
Callback --> Commit[提交偏移量]
Error --> Commit
Commit --> End([结束])

图表来源

ProtocolUplinkConsumerService.java

消息路由机制

ProtocolUplinkConsumerService根据消息内容的不同类型调用相应的业务处理器：

消息类型	处理方法	功能描述
LoginRequest	pileLogin	处理充电桩登录请求
HeartBeatRequest	heartBeat	处理心跳检测
ChargingProgressProto	postChargingProgress	处理充电进度信息
TransactionRecordRequest	onTransactionRecordRequest	处理交易记录
BmsChargingErrorProto	onBmsChargingErrorProto	处理BMS充电错误
RemoteStartChargingResponse	onRemoteStartChargingResponse	处理远程启动响应

并发处理与性能优化

系统采用多线程并发处理机制：

消费者线程池：支持按分区创建独立的消费者线程
消息批处理：支持批量处理提高吞吐量
超时控制：防止长时间阻塞影响系统性能

章节来源

消息序列化与格式

Protobuf序列化

系统主要采用Google Protobuf进行消息序列化，具有以下优势：

高效压缩：二进制格式，传输效率高
类型安全：强类型定义，减少运行时错误
向前兼容：支持协议版本演进
跨语言支持：Java、Python、Go等多种语言支持

UplinkQueueMessage结构

消息格式定义在Protocol Buffer中，包含以下核心字段：

classDiagram
class UplinkQueueMessage {
+int64 messageIdMSB
+int64 messageIdLSB
+int64 sessionIdMSB
+int64 sessionIdLSB
+string messageKey
+string protocolName
+int64 ts
+bytes requestData
+LoginRequest loginRequest
+HeartBeatRequest heartBeatRequest
+ChargingProgressProto chargingProgressProto
+TransactionRecordRequest transactionRecordRequest
+其他消息类型...
}
class LoginRequest {
+string pileCode
+string credential
+string remoteAddress
+string nodeId
+string nodeHostAddress
+int32 nodeRestPort
+int32 nodeGrpcPort
}
class ChargingProgressProto {
+string pileCode
+string gunNo
+string tradeNo
+string outputVoltage
+string outputCurrent
+int32 soc
+int32 totalChargingDurationMin
+string totalChargingEnergyKWh
+string totalChargingCostYuan
}
UplinkQueueMessage --> LoginRequest
UplinkQueueMessage --> ChargingProgressProto

图表来源

消息头（Headers）的使用

系统通过消息头传递元数据信息：

追踪信息：用于分布式追踪和日志关联
统计信息：性能监控和分析
路由信息：消息路由和过滤

章节来源

消息可靠性保证机制

ACK机制

Kafka提供了多层次的确认机制：

生产者确认：
- acks=all：所有副本都确认
- acks=1：Leader确认
- acks=0：不等待确认
消费者确认：
- 手动提交：处理完成后手动确认
- 自动提交：配置自动提交偏移量

重试策略

系统实现了多层次的重试机制：

flowchart TD
Send[发送消息] --> Success{发送成功?}
Success --> |是| Complete[完成]
Success --> |否| Retry{重试次数检查}
Retry --> |未超限| Delay[延迟重试]
Retry --> |已超限| DLQ[死信队列]
Delay --> Backoff[指数退避]
Backoff --> Send
DLQ --> Manual[人工干预]

死信队列（DLQ）处理

对于无法正常处理的消息，系统提供死信队列机制：

自动转移：达到最大重试次数后自动转移到DLQ
人工干预：提供专门的工具和界面处理DLQ消息
监控告警：DLQ消息数量超过阈值时触发告警

章节来源

Kafka主题配置与管理

主题配置

系统使用统一的主题命名规范和配置策略：

配置项	默认值	说明
主题名称	protocol_uplink	标准主题名称
分区数	10	支持水平扩展
副本因子	1	生产环境建议≥3
保留时间	86400000ms	24小时
分段大小	52428800B	50MB
最大保留大小	1048576000B	1GB

KafkaAdmin管理

系统通过KafkaAdmin类提供主题管理功能：

自动创建主题：首次使用时自动创建主题
配置验证：检查主题配置的有效性
分区管理：支持动态调整分区数量

配置示例

queue:
  kafka:
    topic-properties:
      app: "retention.ms:86400000;segment.bytes:52428800;retention.bytes:1048576000;partitions:10;min.insync.replicas:1"

章节来源

监控与性能指标

消费延迟监控

系统提供了完善的监控指标体系：

graph LR
subgraph "监控指标"
A[消费延迟]
B[消息吞吐量]
C[错误率]
D[分区偏移量]
E[消费者组状态]
end
subgraph "监控工具"
F[KafkaConsumerStatsService]
G[日志系统]
H[指标收集]
end
A --> F
B --> F
C --> F
D --> F
E --> F
F --> G
F --> H

图表来源

KafkaConsumerStatsService.java

关键性能指标

消费延迟：消息从生产到消费的时间差
处理速率：每秒处理的消息数量
错误率：处理失败的消息比例
分区平衡：各分区的负载分布情况

监控配置

queue:
  kafka:
    consumer-stats:
      enabled: true
      print-interval-ms: 60000
      kafka-response-timeout-ms: 1000
  app:
    stats:
      enabled: true
      print-interval-ms: 60000
      timer-top-n: 5

章节来源

故障处理与最佳实践

常见故障场景

网络分区：Kafka集群网络中断
消费者组失效：消费者进程异常退出
消息积压：消费速度跟不上生产速度
数据损坏：消息格式错误或损坏

故障恢复策略

flowchart TD
Detect[检测故障] --> Classify{故障分类}
Classify --> |网络问题| Network[网络恢复]
Classify --> |消费者故障| Consumer[重启消费者]
Classify --> |数据问题| Data[数据修复]
Classify --> |系统故障| System[系统恢复]
Network --> Verify[验证恢复]
Consumer --> Verify
Data --> Verify
System --> Verify
Verify --> Success{恢复成功?}
Success --> |是| Monitor[持续监控]
Success --> |否| Escalate[升级处理]

最佳实践建议

配置优化：
- 合理设置分区数量
- 配置合适的副本因子
- 优化网络和磁盘IO
监控告警：
- 设置关键指标阈值
- 建立自动化告警机制
- 准备应急预案
运维管理：
- 定期备份重要数据
- 制定灾难恢复计划
- 进行定期演练

性能调优指南

参数	推荐值	说明
max.poll.records	8192	单次拉取最大记录数
fetch.max.bytes	134217728	单次拉取最大字节数
session.timeout.ms	10000	消费者会话超时时间
max.poll.interval.ms	300000	最大拉取间隔
linger.ms	1	发送延迟时间

章节来源

总结

JChargePointProtocol系统通过精心设计的消息队列集成方案，实现了协议处理服务和业务应用服务之间的高效解耦。Kafka作为核心消息中间件，不仅提供了高吞吐量、低延迟的消息传递能力，还通过其丰富的功能特性确保了系统的可靠性和可扩展性。

系统的关键优势包括：

架构解耦：通过消息队列实现服务间的松耦合
高可靠性：多重确认机制和容错处理
高性能：优化的序列化格式和并发处理
可观测性：完善的监控和诊断能力
可扩展性：支持水平扩展和动态配置

这种设计模式为大型分布式系统提供了可靠的基础设施支撑，特别适合充电桩等IoT设备密集的场景。通过持续的监控和优化，系统能够稳定运行在高并发环境下，为业务发展提供坚实的技术保障。

19 KiB Raw Blame History Unescape Escape

消息队列集成

目录

引言

系统架构概述

Kafka作为核心消息中间件的选择

选择Kafka的核心原因

解耦协议处理和服务业务

KafkaAppQueueFactory设计与实现

工厂模式的应用

生产者和消费者的创建

ProtocolUplinkMsgConsumer的创建

ProtocolUplinkMsgProducer的创建

ProtocolUplinkConsumerService详解

消费者服务的核心职责

消息处理流程

消息路由机制

并发处理与性能优化

消息序列化与格式

Protobuf序列化

UplinkQueueMessage结构

消息头（Headers）的使用

消息可靠性保证机制

ACK机制

重试策略

死信队列（DLQ）处理

Kafka主题配置与管理

主题配置

KafkaAdmin管理

配置示例

监控与性能指标

消费延迟监控

关键性能指标

监控配置

故障处理与最佳实践

常见故障场景

故障恢复策略

最佳实践建议

性能调优指南

总结

19 KiB

Raw Blame History