fix: stabilize jt808 fanout and dedup

2026-06-29 17:57:46 +08:00
parent 1d569b7bf3
commit 705b507f88
4 changed files with 100 additions and 9 deletions
--- a/modules/core/ingest-core/src/main/java/com/lingniu/ingest/core/concurrency/DisruptorEventBus.java
+++ b/modules/core/ingest-core/src/main/java/com/lingniu/ingest/core/concurrency/DisruptorEventBus.java
@@ -103,8 +103,9 @@ public final class DisruptorEventBus implements AutoCloseable {
            if (e == null || !sink.accepts(e)) return;
            try {
                publishToSinkAndTrack(sink, e);
-            } finally {
-                if (endOfBatch) slot.clear();
+            } catch (Throwable t) {
+                failed.incrementAndGet();
+                log.warn("sink {} publish failed", sink.name(), t);
            }
        };
    }
--- a/modules/core/ingest-core/src/main/java/com/lingniu/ingest/core/pipeline/builtin/DedupInterceptor.java
+++ b/modules/core/ingest-core/src/main/java/com/lingniu/ingest/core/pipeline/builtin/DedupInterceptor.java
@@ -9,13 +9,16 @@ import org.springframework.core.Ordered;

 import java.util.Arrays;
 import java.time.Duration;
+import java.util.Map;

 /**
 * 基于 Caffeine 的本地幂等去重。
 *
- * <p>Key 构造：{@code protocolId + vin + command + seq/fingerprint}。如果上游已经在
- * {@link RawFrame#sourceMeta()} 里带了 {@code seq}，优先用真实流水号；否则退化为 raw bytes
- * fingerprint，避免设备重连或 TCP 重发导致同一帧重复进入 DuckDB/Archive。
+ * <p>Key 构造：{@code protocolId + vehicle identity + command + seq/fingerprint}。vehicle identity
+ * 优先使用已解析 VIN；VIN 未解析时使用 phone/deviceId/terminalId/plate，避免 JT808 多终端在
+ * {@code vin=unknown} 时同流水号互相去重。如果上游已经在 {@link RawFrame#sourceMeta()} 里带了
+ * {@code seq}，优先用真实流水号；否则退化为 raw bytes fingerprint，避免设备重连或 TCP 重发导致
+ * 同一帧重复进入 DuckDB/Archive。
 *
 * <p>当前缓存是进程内的，只能保证单实例去重。32960 若做多副本水平扩容，需要把这个位置替换成
 * Redis/集中式幂等键，否则不同实例仍可能各自接收一次相同原始包。
@@ -33,13 +36,14 @@ public class DedupInterceptor implements IngestInterceptor, Ordered {

    @Override
    public boolean before(RawFrame frame, IngestContext ctx) {
-        String vin = frame.sourceMeta().getOrDefault("vin", "unknown");
-        String seq = frame.sourceMeta().get("seq");
+        Map<String, String> meta = frame.sourceMeta() == null ? Map.of() : frame.sourceMeta();
+        String identity = identityKey(frame, meta);
+        String seq = meta.get("seq");
        if (seq == null || seq.isBlank()) {
            // 某些入口没有硬件流水号，使用原始字节哈希作为保底键，至少能挡住同连接内的重复帧。
            seq = rawFingerprint(frame);
        }
-        String key = frame.protocolId() + ":" + vin + ":" + frame.command() + ":" + seq;
+        String key = frame.protocolId() + ":" + identity + ":" + frame.command() + ":" + seq;
        if (seen.asMap().putIfAbsent(key, Boolean.TRUE) != null) {
            ctx.abort("duplicate:" + key);
            return false;
@@ -52,6 +56,20 @@ public class DedupInterceptor implements IngestInterceptor, Ordered {
        return 100;
    }

+    private static String identityKey(RawFrame frame, Map<String, String> meta) {
+        String vin = meta.getOrDefault("vin", "unknown").trim();
+        if (!vin.isBlank() && !"unknown".equalsIgnoreCase(vin)) {
+            return "vin:" + vin;
+        }
+        for (String key : new String[]{"phone", "deviceId", "terminalId", "plate"}) {
+            String value = meta.get(key);
+            if (value != null && !value.isBlank()) {
+                return key + ":" + value.trim();
+            }
+        }
+        return "raw:" + rawFingerprint(frame);
+    }
+
    private static String rawFingerprint(RawFrame frame) {
        byte[] rawBytes = frame.rawBytes();
        if (rawBytes != null && rawBytes.length > 0) {
--- a/modules/core/ingest-core/src/test/java/com/lingniu/ingest/core/concurrency/DisruptorEventBusAwaitTest.java
+++ b/modules/core/ingest-core/src/test/java/com/lingniu/ingest/core/concurrency/DisruptorEventBusAwaitTest.java
@@ -45,10 +45,33 @@ class DisruptorEventBusAwaitTest {
        }
    }

+    @Test
+    void publishFansOutEveryEventToEveryAcceptingSink() throws Exception {
+        int count = 1000;
+        CountingSink rawArchive = new CountingSink("raw-archive", count);
+        CountingSink kafka = new CountingSink("kafka", count);
+
+        try (DisruptorEventBus bus = new DisruptorEventBus(1024, "blocking", List.of(rawArchive, kafka))) {
+            for (int i = 0; i < count; i++) {
+                bus.publish(event("heartbeat-" + i));
+            }
+
+            assertThat(rawArchive.received.await(3, TimeUnit.SECONDS)).isTrue();
+            assertThat(kafka.received.await(3, TimeUnit.SECONDS)).isTrue();
+        }
+
+        assertThat(rawArchive.count.get()).isEqualTo(count);
+        assertThat(kafka.count.get()).isEqualTo(count);
+    }
+
    private static VehicleEvent.Heartbeat event() {
+        return event("heartbeat-1");
+    }
+
+    private static VehicleEvent.Heartbeat event(String eventId) {
        Instant now = Instant.parse("2026-06-23T10:00:00Z");
        return new VehicleEvent.Heartbeat(
-                "heartbeat-1",
+                eventId,
                "VIN001",
                ProtocolId.GB32960,
                now,
@@ -78,4 +101,27 @@ class DisruptorEventBusAwaitTest {
            return CompletableFuture.completedFuture(null);
        }
    }
+
+    private static final class CountingSink implements EventSink {
+        private final String name;
+        private final AtomicLong count = new AtomicLong();
+        private final CountDownLatch received;
+
+        private CountingSink(String name, int expected) {
+            this.name = name;
+            this.received = new CountDownLatch(expected);
+        }
+
+        @Override
+        public String name() {
+            return name;
+        }
+
+        @Override
+        public CompletableFuture<Void> publish(VehicleEvent event) {
+            count.incrementAndGet();
+            received.countDown();
+            return CompletableFuture.completedFuture(null);
+        }
+    }
 }
--- a/modules/core/ingest-core/src/test/java/com/lingniu/ingest/core/pipeline/builtin/DedupInterceptorTest.java
+++ b/modules/core/ingest-core/src/test/java/com/lingniu/ingest/core/pipeline/builtin/DedupInterceptorTest.java
@@ -27,6 +27,21 @@ class DedupInterceptorTest {
        assertThat(ctx.aborted()).isTrue();
    }

+    @Test
+    void doesNotDeduplicateDifferentJt808PhonesWhenVinIsUnknownAndSequenceMatches() {
+        DedupInterceptor interceptor = new DedupInterceptor(100, 60);
+
+        RawFrame firstPhone = jt808Frame("13079960001", "1000");
+        RawFrame secondPhone = jt808Frame("13079960002", "1000");
+        RawFrame duplicateFirstPhone = jt808Frame("13079960001", "1000");
+
+        assertThat(interceptor.before(firstPhone, new IngestContext("trace-1"))).isTrue();
+        assertThat(interceptor.before(secondPhone, new IngestContext("trace-2"))).isTrue();
+        IngestContext duplicateContext = new IngestContext("trace-3");
+        assertThat(interceptor.before(duplicateFirstPhone, duplicateContext)).isFalse();
+        assertThat(duplicateContext.abortReason()).contains("duplicate");
+    }
+
    private static RawFrame frame(byte[] rawBytes) {
        return new RawFrame(
                ProtocolId.GB32960,
@@ -37,4 +52,15 @@ class DedupInterceptorTest {
                Map.of("vin", "TESTSTATVIN000001"),
                Instant.parse("2026-06-22T13:00:00Z"));
    }
+
+    private static RawFrame jt808Frame(String phone, String seq) {
+        return new RawFrame(
+                ProtocolId.JT808,
+                0x0200,
+                0,
+                new Object(),
+                new byte[]{0x02, 0x00, 0x00, 0x1c},
+                Map.of("vin", "unknown", "phone", phone, "seq", seq),
+                Instant.parse("2026-06-22T13:00:00Z"));
+    }
 }