diff --git 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
index f3ab22ec..92760e25 100644
--- 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);
}
};
}
diff --git 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
index 4148a7fe..ecffb669 100644
--- 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 的本地幂等去重。
*
- * Key 构造：{@code protocolId + vin + command + seq/fingerprint}。如果上游已经在
- * {@link RawFrame#sourceMeta()} 里带了 {@code seq}，优先用真实流水号；否则退化为 raw bytes
- * fingerprint，避免设备重连或 TCP 重发导致同一帧重复进入 DuckDB/Archive。
+ *
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。
*
*
当前缓存是进程内的，只能保证单实例去重。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 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 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) {
diff --git 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
index 1c39eab6..d013707a 100644
--- 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 publish(VehicleEvent event) {
+ count.incrementAndGet();
+ received.countDown();
+ return CompletableFuture.completedFuture(null);
+ }
+ }
}
diff --git 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
index 9b3076b0..85e33e23 100644
--- 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"));
+ }
}