shishengliang/lingniu-vehicle-ingest
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

docs: add detailed 32960 pipeline comments

Browse Source
This commit is contained in:
kkfluous
2026-06-23 13:17:37 +08:00
parent ba68ffe061
commit a096e4ce0e
125 changed files with 493 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
modules/core/ingest-core/src/main/java/com/lingniu/ingest/core/concurrency/AsyncBatchExecutor.java
View File

@@ -135,6 +135,8 @@ public final class AsyncBatchExecutor implements AutoCloseable {
private void flush(List<BatchItem> buf) {
if (requiresPerItemTransform(buf)) {
// rawArchiveUri/rawSize 等元数据是每帧唯一的。为了不把第一条帧的元数据错误复制给整批事件，
// 一旦存在逐条 transformer，就按单条调用 Handler，再逐条发布。
for (BatchItem item : buf) {
flushTransformedItem(item);
}
@@ -161,6 +163,7 @@ public final class AsyncBatchExecutor implements AutoCloseable {
private List<VehicleEvent> invoke(List<Object> messages) {
List<VehicleEvent> events;
try {
// Handler 仍然只暴露一个批量签名；是否批满或超时由 Batcher 统一处理。
Object result = def.method().invoke(def.bean(), messages);
events = switch (result) {
case null -> List.of();

2
modules/core/ingest-core/src/main/java/com/lingniu/ingest/core/concurrency/DisruptorEventBus.java
View File

@@ -43,6 +43,7 @@ public final class DisruptorEventBus implements AutoCloseable {
EventHandler<VehicleEventSlot>[] handlers = sinks.stream()
.map(this::toHandler)
.toArray(EventHandler[]::new);
// 每个 sink 一个独立 handler，事件按扇出模式同时写 Kafka、event-file-store 等目标。
disruptor.handleEventsWith(handlers);
disruptor.start();
log.info("DisruptorEventBus started ringBuffer={} wait={} sinks={}",
@@ -66,6 +67,7 @@ public final class DisruptorEventBus implements AutoCloseable {
if (e == null || !sink.accepts(e)) return;
try {
sink.publish(e).exceptionally(ex -> {
// sink 异步失败只计数和打日志，不反向阻塞入站连接；生产侧靠 sink 自身重试/DLQ 保证可观测。
failed.incrementAndGet();
log.warn("sink {} publish failed", sink.name(), ex);
return null;

5
modules/core/ingest-core/src/main/java/com/lingniu/ingest/core/config/IngestCoreAutoConfiguration.java
View File

@@ -21,6 +21,10 @@ import java.util.List;
/**
* ingest-core 的自动装配入口。所有 Bean 都是 {@code @ConditionalOnMissingBean}，便于下游替换。
*
* <p>生产链路顺序：入口收到 {@code RawFrame} → {@link InterceptorChain} 做准入 →
* {@link Dispatcher} 定位协议 Handler → Handler 产出 {@code VehicleEvent} →
* {@link DisruptorEventBus} 扇出到 Archive、DuckDB、Kafka 等 {@code EventSink}。
*/
@AutoConfiguration
@EnableConfigurationProperties(IngestCoreProperties.class)
@@ -46,6 +50,7 @@ public class IngestCoreAutoConfiguration {
@Bean
@ConditionalOnProperty(prefix = "lingniu.ingest.pipeline.dedup", name = "enabled", havingValue = "true", matchIfMissing = true)
public DedupInterceptor dedupInterceptor(IngestCoreProperties props) {
// 去重尽量放在限流前，重复包不应消耗后续每 VIN QPS 配额。
return new DedupInterceptor(props.getDedup().getCacheSize(), props.getDedup().getTtlSeconds());
}

6
modules/core/ingest-core/src/main/java/com/lingniu/ingest/core/config/IngestCoreProperties.java
View File

@@ -19,7 +19,9 @@ public class IngestCoreProperties {
public void setRateLimit(RateLimit rateLimit) { this.rateLimit = rateLimit; }
public static class Disruptor {
/** 事件总线 ring buffer 大小；生产环境应保持 2 的幂，避免 Disruptor 初始化失败。 */
private int ringBufferSize = 131072;
/** waiting 策略直接影响延迟和 CPU 占用，默认 yielding 偏低延迟。 */
private String waitStrategy = "yielding";
private String producerType = "multi";
@@ -33,7 +35,9 @@ public class IngestCoreProperties {
public static class Dedup {
private boolean enabled = true;
/** 单实例缓存的幂等键数量上限，按 VIN 数和重发窗口估算。 */
private int cacheSize = 200000;
/** 幂等键保留窗口；过短会放过重发包，过长会增加内存压力。 */
private long ttlSeconds = 600;
public boolean isEnabled() { return enabled; }
@@ -45,7 +49,9 @@ public class IngestCoreProperties {
}
public static class RateLimit {
/** 每 VIN 每秒进入 Handler 的最大帧数，默认按 32960 高频实时上报预留。 */
private int perVinQps = 50;
/** VIN limiter 缓存上限；覆盖活跃车辆数即可，淘汰后会重新创建 limiter。 */
private int maxVins = 100000;
public int getPerVinQps() { return perVinQps; }

1
modules/core/ingest-core/src/main/java/com/lingniu/ingest/core/dispatcher/AnnotationHandlerBeanPostProcessor.java
View File

@@ -43,6 +43,7 @@ public class AnnotationHandlerBeanPostProcessor implements BeanPostProcessor {
int[] commands = mapping.command().length == 0 ? new int[]{0} : mapping.command();
int[] infoTypes = mapping.infoType().length == 0 ? new int[]{0} : mapping.infoType();
// 这些注解只声明策略，实际限流、幂等、异步批处理由 Dispatcher/Interceptor 层解释执行。
RateLimited rl = AnnotatedElementUtils.findMergedAnnotation(method, RateLimited.class);
IdempotentKey ik = AnnotatedElementUtils.findMergedAnnotation(method, IdempotentKey.class);
AsyncBatch ab = AnnotatedElementUtils.findMergedAnnotation(method, AsyncBatch.class);

2
modules/core/ingest-core/src/main/java/com/lingniu/ingest/core/dispatcher/Dispatcher.java
View File

@@ -122,6 +122,7 @@ public final class Dispatcher {
}
private static String nextRawArchiveEventId(Instant ingestTime) {
// 文件名需要可排序且单进程内唯一：毫秒时间放大到微秒量级，再用 AtomicLong 递增兜底。
long base = (ingestTime == null ? Instant.now() : ingestTime).toEpochMilli() * 1000L;
long next = RAW_ARCHIVE_SEQUENCE.updateAndGet(previous -> Math.max(previous + 1, base));
return Long.toString(next);
@@ -141,6 +142,7 @@ public final class Dispatcher {
if (event == null || rawArchive.isEmpty() || event instanceof VehicleEvent.RawArchive) {
return event;
}
// 派生事件携带 rawArchiveUri，便于后续服务从 Kafka 事件回查原始报文。
Map<String, String> metadata = addRawArchiveMetadata(event.metadata(), rawArchive.eventId(), rawArchive.key());
return switch (event) {
case VehicleEvent.Realtime e -> new VehicleEvent.Realtime(

1
modules/core/ingest-core/src/main/java/com/lingniu/ingest/core/dispatcher/HandlerInvoker.java
View File

@@ -20,6 +20,7 @@ public class HandlerInvoker {
return switch (result) {
case null -> Collections.emptyList();
case VehicleEvent e -> List.of(e);
// Handler 可以一次产出多个领域事件，例如 GB32960 0x02 同时产出 Realtime/Location/Alarm。
case List<?> list -> (List<VehicleEvent>) list;
default -> throw new IllegalStateException(
"Handler return type must be VehicleEvent or List<VehicleEvent>: " + def.method());

6
modules/core/ingest-core/src/main/java/com/lingniu/ingest/core/pipeline/InterceptorChain.java
View File

@@ -11,6 +11,11 @@ import java.util.List;
/**
* 顺序执行的拦截器链，通过 Spring {@code @Order} 或 {@code Ordered} 排序。
*
* <p>{@link #before(RawFrame, IngestContext)} 是接入链路的准入闸门：任何拦截器返回
* {@code false} 或调用 {@link IngestContext#abort(String)} 都会停止后续协议解析/Handler 调用。
* {@link #after(VehicleEvent, IngestContext)} 与 {@link #onError(Throwable, IngestContext)}
* 则保持广播语义，让指标、审计等横切逻辑都能观察到同一条处理结果。
*/
public final class InterceptorChain {
@@ -24,6 +29,7 @@ public final class InterceptorChain {
public boolean before(RawFrame frame, IngestContext ctx) {
for (IngestInterceptor i : interceptors) {
// before 阶段短路是有意的：去重、限流失败后不应继续消耗解析和存储资源。
if (!i.before(frame, ctx) || ctx.aborted()) {
return false;
}

9
modules/core/ingest-core/src/main/java/com/lingniu/ingest/core/pipeline/builtin/DedupInterceptor.java
View File

@@ -13,8 +13,12 @@ import java.time.Duration;
/**
* 基于 Caffeine 的本地幂等去重。
*
* <p>Key 构造：{@code protocolId + command + sourceMeta.seq} —— 真实实现可以接入 Redis 做多节点一致性，
* 本实现只覆盖单节点场景，满足第一阶段 PoC 需求。
* <p>Key 构造：{@code protocolId + vin + command + seq/fingerprint}。如果上游已经在
* {@link RawFrame#sourceMeta()} 里带了 {@code seq}，优先用真实流水号；否则退化为 raw bytes
* fingerprint，避免设备重连或 TCP 重发导致同一帧重复进入 DuckDB/Archive。
*
* <p>当前缓存是进程内的，只能保证单实例去重。32960 若做多副本水平扩容，需要把这个位置替换成
* Redis/集中式幂等键，否则不同实例仍可能各自接收一次相同原始包。
*/
public class DedupInterceptor implements IngestInterceptor, Ordered {
@@ -32,6 +36,7 @@ public class DedupInterceptor implements IngestInterceptor, Ordered {
String vin = frame.sourceMeta().getOrDefault("vin", "unknown");
String seq = frame.sourceMeta().get("seq");
if (seq == null || seq.isBlank()) {
// 某些入口没有硬件流水号，使用原始字节哈希作为保底键，至少能挡住同连接内的重复帧。
seq = rawFingerprint(frame);
}
String key = frame.protocolId() + ":" + vin + ":" + frame.command() + ":" + seq;

4
modules/core/ingest-core/src/main/java/com/lingniu/ingest/core/pipeline/builtin/RateLimitInterceptor.java
View File

@@ -13,6 +13,9 @@ import java.time.Duration;
/**
* 单 VIN 速率限制。每个 VIN 一个独立的 Resilience4j RateLimiter，由 Caffeine 按 LRU 管理。
*
* <p>这里限的是进入业务 Handler 前的原始帧，不区分命令类型；32960 生产侧如需允许登入/登出穿透，
* 应在此处扩展白名单或改为按命令配置，而不是在存储层丢弃。
*/
public class RateLimitInterceptor implements IngestInterceptor, Ordered {
@@ -34,6 +37,7 @@ public class RateLimitInterceptor implements IngestInterceptor, Ordered {
@Override
public boolean before(RawFrame frame, IngestContext ctx) {
String vin = frame.sourceMeta().getOrDefault("vin", "unknown");
// VIN 粒度隔离，避免某一辆车的高频/异常上报挤占其他车辆处理额度。
RateLimiter rl = limiters.get(vin, k -> RateLimiter.of("vin-" + k, defaultConfig));
if (!rl.acquirePermission()) {
ctx.abort("rate-limited:" + vin);