docs: align comments with tdengine history path
This commit is contained in:
@@ -11,8 +11,8 @@ import java.util.Map;
|
|||||||
* Shared raw-archive key conventions used by Dispatcher, archive sink, and query services.
|
* Shared raw-archive key conventions used by Dispatcher, archive sink, and query services.
|
||||||
*
|
*
|
||||||
* <p>32960 生产链路里,RAW .bin 文件本体按 {@code 日期/协议/VIN/eventId.bin} 落在 archive 根目录,
|
* <p>32960 生产链路里,RAW .bin 文件本体按 {@code 日期/协议/VIN/eventId.bin} 落在 archive 根目录,
|
||||||
* DuckDB 只保存 {@code archive://...} 引用和查询索引。这个类集中维护 key 规则,避免接收端、落盘端、
|
* TDengine raw_frames 或兼容索引只保存 {@code archive://...} 引用。这个类集中维护 key 规则,
|
||||||
* snapshot 查询端对目录分区的理解不一致。</p>
|
* 避免接收端、落盘端、snapshot 查询端对目录分区的理解不一致。</p>
|
||||||
*/
|
*/
|
||||||
public final class RawArchiveKeys {
|
public final class RawArchiveKeys {
|
||||||
|
|
||||||
|
|||||||
@@ -3,11 +3,11 @@ package com.lingniu.ingest.api.event;
|
|||||||
/**
|
/**
|
||||||
* One normalized telemetry field value.
|
* One normalized telemetry field value.
|
||||||
*
|
*
|
||||||
* <p>Protocol mappers use stable internal field keys so Kafka, Parquet, Redis,
|
* <p>Protocol mappers use stable internal field keys so Kafka, TDengine, Redis,
|
||||||
* and statistics do not depend on protocol-specific names.
|
* and statistics do not depend on protocol-specific names.
|
||||||
*
|
*
|
||||||
* <p>{@code value} 统一用字符串承载,真实类型由 {@code valueType} 表达。这样 Kafka
|
* <p>{@code value} 统一用字符串承载,真实类型由 {@code valueType} 表达。这样 Kafka
|
||||||
* protobuf、CSV 导出、DuckDB JSON 字段和前端展示可以共用同一份字段结构;数值精度/格式化
|
* protobuf、CSV 导出、JSON 字段和前端展示可以共用同一份字段结构;数值精度/格式化
|
||||||
* 由查询层或前端按 {@code valueType + unit} 决定。
|
* 由查询层或前端按 {@code valueType + unit} 决定。
|
||||||
*/
|
*/
|
||||||
public record TelemetryFieldValue(
|
public record TelemetryFieldValue(
|
||||||
|
|||||||
@@ -12,9 +12,9 @@ import java.util.Optional;
|
|||||||
/**
|
/**
|
||||||
* Normalized full-field telemetry snapshot for one parsed vehicle event.
|
* Normalized full-field telemetry snapshot for one parsed vehicle event.
|
||||||
*
|
*
|
||||||
* <p>This is the shared contract for Kafka, Parquet history, Redis hot state,
|
* <p>This is the shared contract for Kafka, TDengine history, Redis hot state,
|
||||||
* and statistics. It intentionally lives in {@code ingest-api} and has no
|
* and statistics. It intentionally lives in {@code ingest-api} and has no
|
||||||
* Spring, Kafka, Redis, or DuckDB dependency.
|
* Spring, Kafka, Redis, or TDengine dependency.
|
||||||
*/
|
*/
|
||||||
public record TelemetrySnapshot(
|
public record TelemetrySnapshot(
|
||||||
String eventId,
|
String eventId,
|
||||||
|
|||||||
@@ -52,7 +52,7 @@ public final class DisruptorEventBus implements AutoCloseable {
|
|||||||
EventHandler<VehicleEventSlot>[] handlers = this.sinks.stream()
|
EventHandler<VehicleEventSlot>[] handlers = this.sinks.stream()
|
||||||
.map(this::toHandler)
|
.map(this::toHandler)
|
||||||
.toArray(EventHandler[]::new);
|
.toArray(EventHandler[]::new);
|
||||||
// 每个 sink 一个独立 handler,事件按扇出模式同时写 Kafka、event-file-store 等目标。
|
// 每个 sink 一个独立 handler,事件按扇出模式同时写 Kafka、archive、历史索引等目标。
|
||||||
disruptor.handleEventsWith(handlers);
|
disruptor.handleEventsWith(handlers);
|
||||||
disruptor.start();
|
disruptor.start();
|
||||||
log.info("DisruptorEventBus started ringBuffer={} wait={} sinks={}",
|
log.info("DisruptorEventBus started ringBuffer={} wait={} sinks={}",
|
||||||
|
|||||||
@@ -24,7 +24,7 @@ import java.util.List;
|
|||||||
*
|
*
|
||||||
* <p>生产链路顺序:入口收到 {@code RawFrame} → {@link InterceptorChain} 做准入 →
|
* <p>生产链路顺序:入口收到 {@code RawFrame} → {@link InterceptorChain} 做准入 →
|
||||||
* {@link Dispatcher} 定位协议 Handler → Handler 产出 {@code VehicleEvent} →
|
* {@link Dispatcher} 定位协议 Handler → Handler 产出 {@code VehicleEvent} →
|
||||||
* {@link DisruptorEventBus} 扇出到 Archive、DuckDB、Kafka 等 {@code EventSink}。
|
* {@link DisruptorEventBus} 扇出到 Archive、Kafka、历史索引等 {@code EventSink}。
|
||||||
*/
|
*/
|
||||||
@AutoConfiguration
|
@AutoConfiguration
|
||||||
@EnableConfigurationProperties(IngestCoreProperties.class)
|
@EnableConfigurationProperties(IngestCoreProperties.class)
|
||||||
|
|||||||
@@ -18,7 +18,7 @@ import java.util.Map;
|
|||||||
* 优先使用已解析 VIN;VIN 未解析时使用 phone/deviceId/terminalId/plate,避免 JT808 多终端在
|
* 优先使用已解析 VIN;VIN 未解析时使用 phone/deviceId/terminalId/plate,避免 JT808 多终端在
|
||||||
* {@code vin=unknown} 时同流水号互相去重。如果上游已经在 {@link RawFrame#sourceMeta()} 里带了
|
* {@code vin=unknown} 时同流水号互相去重。如果上游已经在 {@link RawFrame#sourceMeta()} 里带了
|
||||||
* {@code seq},优先用真实流水号;否则退化为 raw bytes fingerprint,避免设备重连或 TCP 重发导致
|
* {@code seq},优先用真实流水号;否则退化为 raw bytes fingerprint,避免设备重连或 TCP 重发导致
|
||||||
* 同一帧重复进入 DuckDB/Archive。
|
* 同一帧重复进入 Archive/Kafka/历史索引。
|
||||||
*
|
*
|
||||||
* <p>当前缓存是进程内的,只能保证单实例去重。32960 若做多副本水平扩容,需要把这个位置替换成
|
* <p>当前缓存是进程内的,只能保证单实例去重。32960 若做多副本水平扩容,需要把这个位置替换成
|
||||||
* Redis/集中式幂等键,否则不同实例仍可能各自接收一次相同原始包。
|
* Redis/集中式幂等键,否则不同实例仍可能各自接收一次相同原始包。
|
||||||
|
|||||||
@@ -18,7 +18,7 @@ import java.util.List;
|
|||||||
* <p>这是 GB32960 协议在通用 Dispatcher 管线里的**唯一入口** Bean——移除将导致所有
|
* <p>这是 GB32960 协议在通用 Dispatcher 管线里的**唯一入口** Bean——移除将导致所有
|
||||||
* GB32960 帧产出的事件丢失。构造器注入 {@link Gb32960EventMapper},全部业务逻辑限制
|
* GB32960 帧产出的事件丢失。构造器注入 {@link Gb32960EventMapper},全部业务逻辑限制
|
||||||
* 在纯函数里,不触碰数据库 / 不做 IO。产出的事件由 Dispatcher 统一投递到 Disruptor;
|
* 在纯函数里,不触碰数据库 / 不做 IO。产出的事件由 Dispatcher 统一投递到 Disruptor;
|
||||||
* 下游 sink 再决定写 raw archive、DuckDB/Parquet 历史库或转发 Kafka。
|
* 下游 sink 再决定写 raw archive、TDengine 历史库或转发 Kafka。
|
||||||
*
|
*
|
||||||
* <p>当前 32960 专用 snapshot 查询不依赖 Realtime/Location 事件表,而是通过 raw archive URI
|
* <p>当前 32960 专用 snapshot 查询不依赖 Realtime/Location 事件表,而是通过 raw archive URI
|
||||||
* 回读原始 .bin 并即时解码合并,避免不同车辆或不同子包之间做大范围 join。
|
* 回读原始 .bin 并即时解码合并,避免不同车辆或不同子包之间做大范围 join。
|
||||||
|
|||||||
@@ -29,7 +29,7 @@ import java.util.Map;
|
|||||||
* Netty 入站处理器:字节 → Gb32960Message → 认证 → RawFrame → Dispatcher。
|
* Netty 入站处理器:字节 → Gb32960Message → 认证 → RawFrame → Dispatcher。
|
||||||
*
|
*
|
||||||
* <p>这是 32960 生产链路的入口边界。它只做协议层职责:解码、鉴权、应答、组装
|
* <p>这是 32960 生产链路的入口边界。它只做协议层职责:解码、鉴权、应答、组装
|
||||||
* {@link RawFrame} 并投递 Dispatcher;原始包归档、DuckDB/Parquet 索引、Kafka 转发都在
|
* {@link RawFrame} 并投递 Dispatcher;原始包归档、TDengine 历史索引、Kafka 转发都在
|
||||||
* 下游 sink 中完成,避免 Netty EventLoop 被存储 IO 阻塞。
|
* 下游 sink 中完成,避免 Netty EventLoop 被存储 IO 阻塞。
|
||||||
*
|
*
|
||||||
* <p>认证逻辑:
|
* <p>认证逻辑:
|
||||||
|
|||||||
@@ -111,7 +111,7 @@ public class Jt808ChannelHandler extends SimpleChannelInboundHandler<Object> {
|
|||||||
processDecodedFrame(ctx, frame, msg);
|
processDecodedFrame(ctx, frame, msg);
|
||||||
} catch (RuntimeException e) {
|
} catch (RuntimeException e) {
|
||||||
// 解码成功但业务处理失败时仍派发一条 malformed RawFrame,
|
// 解码成功但业务处理失败时仍派发一条 malformed RawFrame,
|
||||||
// 保证原始帧能被 archive/event-file-store 留痕,方便事后排查。
|
// 保证原始帧能被 archive、Kafka 和历史索引留痕,方便事后排查。
|
||||||
log.warn("[jt808] processing failed peer={} phone={} msgId=0x{} error={}",
|
log.warn("[jt808] processing failed peer={} phone={} msgId=0x{} error={}",
|
||||||
addr(ctx), msg.header().phone(), Integer.toHexString(msg.header().messageId()), e.getMessage());
|
addr(ctx), msg.header().phone(), Integer.toHexString(msg.header().messageId()), e.getMessage());
|
||||||
log.debug("[jt808] processing failed stack peer={}", addr(ctx), e);
|
log.debug("[jt808] processing failed stack peer={}", addr(ctx), e);
|
||||||
|
|||||||
@@ -28,7 +28,7 @@ public final class VehicleStateController {
|
|||||||
|
|
||||||
@GetMapping("/{vin}")
|
@GetMapping("/{vin}")
|
||||||
public ResponseEntity<String> state(@PathVariable String vin) {
|
public ResponseEntity<String> state(@PathVariable String vin) {
|
||||||
// 查询的是 Redis 最新状态快照,不是 event-file-store 的历史 snapshot。
|
// 查询的是 Redis 最新状态快照,不是历史明细库。
|
||||||
return json(repository.getState(vin));
|
return json(repository.getState(vin));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user