From ebbf2ca8fbbceccf7d247d310108048c4c339df4 Mon Sep 17 00:00:00 2001 From: lingniu Date: Wed, 1 Jul 2026 11:25:00 +0800 Subject: [PATCH] docs: clarify kafka-only production channel --- DECISIONS.md | 4 ++-- docs/module-data-flow.html | 10 +++++----- .../vehicle-ingest-tdengine-verification.md | 4 +--- .../PortainerComposeResourceLimitsTest.java | 20 ++++++++++++++++++- 4 files changed, 27 insertions(+), 11 deletions(-) diff --git a/DECISIONS.md b/DECISIONS.md index 37bb8bf4..337926ee 100644 --- a/DECISIONS.md +++ b/DECISIONS.md @@ -2,11 +2,11 @@ > 本文件记录 lingniu-vehicle-ingest v2 的关键架构决策，后续每次重大调整追加条目（不删除旧条目）。 -## ADR-001 MQ 选型：Kafka +## ADR-001 消息通道：Kafka - **Status**: Accepted - **Context**: 需要高吞吐、严格单车有序、成熟生态 - **Decision**: 使用 Kafka，分区 key = VIN -- **Consequences**: 下游消费者需使用消费组 + 分区内顺序消费；不使用 RocketMQ 的事务/顺序特性 +- **Consequences**: 下游消费者需使用消费组 + 分区内顺序消费；生产链路只支持 Kafka。 ## ADR-002 线上消息格式：Protobuf - **Status**: Accepted diff --git a/docs/module-data-flow.html b/docs/module-data-flow.html index 104eb444..8bfe2c58 100644 --- a/docs/module-data-flow.html +++ b/docs/module-data-flow.html @@ -414,7 +414,7 @@ sequenceDiagram participant Dispatcher as Dispatcher participant Handler as Protocol Handler
Mapper participant EventBus as DisruptorEventBus - participant MQ as sink-kafka
Kafka Envelope + participant Kafka as sink-kafka
Kafka Envelope participant Archive as sink-archive
ArchiveStore participant History as vehicle-history-app participant TDengine as tdengine-history-store
raw_frames + locations @@ -430,12 +430,12 @@ sequenceDiagram Handler->>Handler: 协议字段映射为内部字段 Dispatcher->>Handler: 绑定 rawArchiveKey / rawArchiveUri Handler->>EventBus: VehicleEvent(Realtime / Location / Alarm / Login ...) - EventBus->>MQ: 投递规整后的业务事件 - MQ->>MQ: EnvelopeMapper 转 Protobuf - MQ->>History: Kafka event/raw topic，key=vin，单车有序 + EventBus->>Kafka: 投递规整后的业务事件 + Kafka->>Kafka: EnvelopeMapper 转 Protobuf + Kafka->>History: event/raw topic，key=vin，单车有序 History->>TDengine: 写 raw_frames、locations 和 raw parsed JSON TDengine->>Consumer: 分页历史查询、原始帧追溯、位置查询 - MQ->>Consumer: Redis 热状态、指标统计、告警工单等下游消费 + Kafka->>Consumer: Redis 热状态、指标统计、告警工单等下游消费
diff --git a/docs/operations/vehicle-ingest-tdengine-verification.md b/docs/operations/vehicle-ingest-tdengine-verification.md index 0a7485b3..04b44a43 100644 --- a/docs/operations/vehicle-ingest-tdengine-verification.md +++ b/docs/operations/vehicle-ingest-tdengine-verification.md @@ -24,13 +24,12 @@ export TDENGINE_USERNAME=root export TDENGINE_PASSWORD='' export TDENGINE_MIN_IDLE=0 export TDENGINE_MAX_POOL_SIZE=32 -export EVENT_FILE_STORE_ENABLED=false export TDENGINE_TELEMETRY_FIELDS_ENABLED=false ``` `TDENGINE_MIN_IDLE=0` 用于减少冷启动时 TDengine 连接重试日志。TDengine 地址稳定后，再按生产并发调大连接池。 -高吞吐生产验收默认关闭 `EVENT_FILE_STORE_ENABLED` 和 `TDENGINE_TELEMETRY_FIELDS_ENABLED`，验证 TDengine `raw_frames`、位置表和分页查询闭环。RAW 帧的 `parsedJson`、`metadataJson`、`rawUri` 直接进入 TDengine，GB32960 snapshot/fields 接口优先基于 `raw_frames` 中的结构化 JSON 返回结果；接入服务写出的原始 `.bin` 只作为冷备复核材料。需要兼容旧 `/api/event-history/records` 低层记录接口或逐字段趋势宽表时，再显式开启对应开关，并单独压测写入放大。 +高吞吐生产验收默认保持 `TDENGINE_TELEMETRY_FIELDS_ENABLED=false`，验证 TDengine `raw_frames`、位置表和分页查询闭环。RAW 帧的 `parsedJson`、`metadataJson`、`rawUri` 直接进入 TDengine，GB32960 snapshot/fields 接口基于 `raw_frames` 与当前解析器即时返回结构化结果；接入服务写出的原始 `.bin` 只作为冷备复核材料。需要逐字段趋势宽表时，再显式开启 `TDENGINE_TELEMETRY_FIELDS_ENABLED=true`，并单独压测写入放大。 ## 构建 @@ -83,7 +82,6 @@ lsof -nP -iTCP:32960 -sTCP:LISTEN ```bash HTTP_PORT=20200 \ KAFKA_CONSUMER_ENABLED=true \ -EVENT_FILE_STORE_ENABLED=false \ EVENT_FILE_STORE_PATH=./target/tdengine-verification/event-store \ TDENGINE_TELEMETRY_FIELDS_ENABLED=false \ java -jar modules/apps/vehicle-history-app/target/vehicle-history-app.jar diff --git a/modules/apps/vehicle-history-app/src/test/java/com/lingniu/ingest/historyapp/PortainerComposeResourceLimitsTest.java b/modules/apps/vehicle-history-app/src/test/java/com/lingniu/ingest/historyapp/PortainerComposeResourceLimitsTest.java index 77ab1871..773835cd 100644 --- a/modules/apps/vehicle-history-app/src/test/java/com/lingniu/ingest/historyapp/PortainerComposeResourceLimitsTest.java +++ b/modules/apps/vehicle-history-app/src/test/java/com/lingniu/ingest/historyapp/PortainerComposeResourceLimitsTest.java @@ -116,7 +116,9 @@ class PortainerComposeResourceLimitsTest { .contains("TDengine `raw_frames`") .doesNotContain("三个服务必须共用同一个 `SINK_ARCHIVE_PATH`") .doesNotContain("再回读共享 archive") - .doesNotContain("读取共享 archive 中的原始 `.bin`"); + .doesNotContain("读取共享 archive 中的原始 `.bin`") + .doesNotContain("EVENT_FILE_STORE_ENABLED") + .doesNotContain("/api/event-history/records"); assertThat(splitRunbook) .contains("TDengine `raw_frames`") .doesNotContain("must share the same `SINK_ARCHIVE_PATH`") @@ -124,6 +126,22 @@ class PortainerComposeResourceLimitsTest { .doesNotContain("shared volume mounted to all three services"); } + @Test + void architectureDocsNameKafkaInsteadOfGenericMq() throws IOException { + String decisions = Files.readString(repositoryRoot().resolve("DECISIONS.md")); + String dataFlow = Files.readString(repositoryRoot().resolve("docs/module-data-flow.html")); + + assertThat(decisions) + .contains("生产链路只支持 Kafka") + .doesNotContain("MQ 选型") + .doesNotContain("RocketMQ"); + assertThat(dataFlow) + .contains("participant Kafka as sink-kafka") + .doesNotContain("participant MQ") + .doesNotContain("->>MQ") + .doesNotContain("MQ->>"); + } + @Test void splitRunbookKeepsAnalyticsAsJt808MetricRuntimeOnly() throws IOException { String splitRunbook = Files.readString(repositoryRoot()