From 74e01acc7a64deab61d84212259e38a1bec74aef Mon Sep 17 00:00:00 2001 From: lingniu Date: Wed, 1 Jul 2026 10:12:52 +0800 Subject: [PATCH] perf: reuse mqtt identity resolution --- .../mqtt/client/MqttEndpointManager.java | 1 + .../mqtt/mapper/YutongEventMapper.java | 11 +++ .../inbound/mqtt/model/MqttPayload.java | 41 +++++++++++ .../inbound/mqtt/MqttEndpointManagerTest.java | 73 +++++++++++++++++++ 4 files changed, 126 insertions(+) diff --git a/modules/inbound/inbound-mqtt/src/main/java/com/lingniu/ingest/inbound/mqtt/client/MqttEndpointManager.java b/modules/inbound/inbound-mqtt/src/main/java/com/lingniu/ingest/inbound/mqtt/client/MqttEndpointManager.java index c710e8ce..efe435bf 100644 --- a/modules/inbound/inbound-mqtt/src/main/java/com/lingniu/ingest/inbound/mqtt/client/MqttEndpointManager.java +++ b/modules/inbound/inbound-mqtt/src/main/java/com/lingniu/ingest/inbound/mqtt/client/MqttEndpointManager.java @@ -141,6 +141,7 @@ public final class MqttEndpointManager implements AutoCloseable { MqttPayload parsed = profileRegistry.parse(ep.getName(), ep.getProfile(), topic, payload); String externalDeviceId = firstNonBlank(parsed.deviceId(), parsed.vin()); IdentityResolution identity = resolveIdentity(parsed); + parsed = parsed.withIdentity(identity.identity(), identity.errorMessage()); Map meta = new HashMap<>(4); // metadata 同时保留外部设备号和内部 VIN，便于排查绑定表错误导致的车辆归属问题。 meta.put("vin", identity.identity().vin()); diff --git a/modules/inbound/inbound-mqtt/src/main/java/com/lingniu/ingest/inbound/mqtt/mapper/YutongEventMapper.java b/modules/inbound/inbound-mqtt/src/main/java/com/lingniu/ingest/inbound/mqtt/mapper/YutongEventMapper.java index b8537222..01b9e1ef 100644 --- a/modules/inbound/inbound-mqtt/src/main/java/com/lingniu/ingest/inbound/mqtt/mapper/YutongEventMapper.java +++ b/modules/inbound/inbound-mqtt/src/main/java/com/lingniu/ingest/inbound/mqtt/mapper/YutongEventMapper.java @@ -109,6 +109,17 @@ public final class YutongEventMapper implements EventMapper { } private IdentityResolution resolveIdentity(MqttPayload payload) { + if (payload.hasIdentitySnapshot()) { + VehicleIdentitySource source = VehicleIdentitySource.UNKNOWN; + try { + source = VehicleIdentitySource.valueOf(payload.identitySource()); + } catch (IllegalArgumentException ignored) { + // 保守兜底，外部化的 payload 如果带了未知 source 不应影响接收主链路。 + } + return new IdentityResolution( + new VehicleIdentity(payload.resolvedVin(), payload.identityResolved(), source), + payload.identityErrorMessage().isBlank() ? null : payload.identityErrorMessage()); + } String externalDeviceId = firstNonBlank(payload.deviceId(), payload.vin()); String externalVin = payload.externalVin() == null ? "" : payload.externalVin(); String phone = payload.phone() == null ? "" : payload.phone(); diff --git a/modules/inbound/inbound-mqtt/src/main/java/com/lingniu/ingest/inbound/mqtt/model/MqttPayload.java b/modules/inbound/inbound-mqtt/src/main/java/com/lingniu/ingest/inbound/mqtt/model/MqttPayload.java index 7b382ff9..ba454c64 100644 --- a/modules/inbound/inbound-mqtt/src/main/java/com/lingniu/ingest/inbound/mqtt/model/MqttPayload.java +++ b/modules/inbound/inbound-mqtt/src/main/java/com/lingniu/ingest/inbound/mqtt/model/MqttPayload.java @@ -1,6 +1,9 @@ package com.lingniu.ingest.inbound.mqtt.model; +import com.fasterxml.jackson.annotation.JsonIgnore; import com.fasterxml.jackson.databind.JsonNode; +import com.lingniu.ingest.identity.VehicleIdentity; +import com.lingniu.ingest.identity.VehicleIdentitySource; import java.time.Instant; @@ -15,6 +18,7 @@ import java.time.Instant; * @param phone 外部手机号 / SIM 字段 * @param deviceId 外部设备号 / 终端 ID / IMEI 字段 * @param plateNo 外部车牌字段 + * @param resolvedVin 已解析的内部 VIN，供 RAW 和业务事件复用同一次身份查询 * @param deviceTime 设备上报时刻 * @param data 原始 JSON 树，供 Mapper 按 profile 提取字段 */ @@ -27,9 +31,33 @@ public record MqttPayload( String phone, String deviceId, String plateNo, + @JsonIgnore String resolvedVin, + @JsonIgnore boolean identityResolved, + @JsonIgnore String identitySource, + @JsonIgnore String identityErrorMessage, Instant deviceTime, JsonNode data ) { + public MqttPayload { + resolvedVin = resolvedVin == null ? "" : resolvedVin.trim(); + identitySource = identitySource == null ? "" : identitySource.trim(); + identityErrorMessage = identityErrorMessage == null ? "" : identityErrorMessage.trim(); + } + + public MqttPayload(String endpointName, + String profile, + String topic, + String vin, + String externalVin, + String phone, + String deviceId, + String plateNo, + Instant deviceTime, + JsonNode data) { + this(endpointName, profile, topic, vin, externalVin, phone, deviceId, plateNo, + "", false, "", "", deviceTime, data); + } + public MqttPayload(String endpointName, String profile, String topic, @@ -38,4 +66,17 @@ public record MqttPayload( JsonNode data) { this(endpointName, profile, topic, vin, "", "", vin, "", deviceTime, data); } + + public MqttPayload withIdentity(VehicleIdentity identity, String errorMessage) { + VehicleIdentity safe = identity == null + ? new VehicleIdentity("unknown", false, VehicleIdentitySource.UNKNOWN) + : identity; + return new MqttPayload(endpointName, profile, topic, vin, externalVin, phone, deviceId, plateNo, + safe.vin(), safe.resolved(), safe.source().name(), errorMessage, deviceTime, data); + } + + @JsonIgnore + public boolean hasIdentitySnapshot() { + return !identitySource.isBlank(); + } } diff --git a/modules/inbound/inbound-mqtt/src/test/java/com/lingniu/ingest/inbound/mqtt/MqttEndpointManagerTest.java b/modules/inbound/inbound-mqtt/src/test/java/com/lingniu/ingest/inbound/mqtt/MqttEndpointManagerTest.java index 4efa29d2..e4bfcb10 100644 --- a/modules/inbound/inbound-mqtt/src/test/java/com/lingniu/ingest/inbound/mqtt/MqttEndpointManagerTest.java +++ b/modules/inbound/inbound-mqtt/src/test/java/com/lingniu/ingest/inbound/mqtt/MqttEndpointManagerTest.java @@ -12,7 +12,11 @@ import com.lingniu.ingest.core.dispatcher.HandlerInvoker; import com.lingniu.ingest.core.dispatcher.HandlerRegistry; import com.lingniu.ingest.core.pipeline.InterceptorChain; import com.lingniu.ingest.identity.InMemoryVehicleIdentityService; +import com.lingniu.ingest.identity.VehicleIdentity; import com.lingniu.ingest.identity.VehicleIdentityBinding; +import com.lingniu.ingest.identity.VehicleIdentityLookup; +import com.lingniu.ingest.identity.VehicleIdentityResolver; +import com.lingniu.ingest.identity.VehicleIdentitySource; import com.lingniu.ingest.inbound.mqtt.client.MqttEndpointManager; import com.lingniu.ingest.inbound.mqtt.config.MqttInboundProperties; import com.lingniu.ingest.inbound.mqtt.handler.MqttRealtimeHandler; @@ -30,6 +34,7 @@ import java.util.concurrent.CompletableFuture; import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; @@ -86,6 +91,55 @@ class MqttEndpointManagerTest { eventBus.close(); } + @Test + void inboundMessageReusesResolvedIdentityForRawArchiveAndMappedEvents() throws Exception { + MqttInboundProperties.Endpoint endpoint = new MqttInboundProperties.Endpoint(); + endpoint.setName("endpoint-a"); + endpoint.setTopic("/ytforward/#"); + endpoint.setProfile("yutong"); + byte[] payload = """ + {"device":"YT-DEVICE-001","time":"20260622150000","data":{"METER_SPEED":31,"TOTAL_MILEAGE":1024}} + """.trim().getBytes(StandardCharsets.UTF_8); + CountingIdentityResolver identities = new CountingIdentityResolver(new VehicleIdentity( + "LNVIN000000MQTT01", true, VehicleIdentitySource.BOUND_DEVICE_ID)); + RecordingSink sink = new RecordingSink(2); + DisruptorEventBus eventBus = new DisruptorEventBus(1024, "blocking", List.of(sink)); + AsyncBatchExecutor batchExecutor = new AsyncBatchExecutor(eventBus::publish); + MqttProfileRegistry profileRegistry = new MqttProfileRegistry(List.of( + new YutongMqttProfile( + new MqttPayloadParser(new ObjectMapper()), + new YutongEventMapper(identities)))); + Dispatcher dispatcher = new Dispatcher( + registry(profileRegistry), + new InterceptorChain(List.of()), + new HandlerInvoker(), + eventBus, + batchExecutor); + MqttEndpointManager manager = new MqttEndpointManager( + new MqttInboundProperties(), profileRegistry, identities, dispatcher); + + invokeOnMessage(manager, endpoint, "/ytforward/vehicle/YT-DEVICE-001", payload); + + assertThat(sink.await()).isTrue(); + assertThat(identities.calls()).isEqualTo(1); + assertThat(sink.events) + .extracting(VehicleEvent::vin) + .contains("LNVIN000000MQTT01"); + assertThat(sink.events).anySatisfy(event -> { + assertThat(event).isInstanceOf(VehicleEvent.RawArchive.class); + VehicleEvent.RawArchive archive = (VehicleEvent.RawArchive) event; + assertThat(archive.parsedJson()) + .contains("\"deviceId\":\"YT-DEVICE-001\"") + .doesNotContain("resolvedVin") + .doesNotContain("identityResolved") + .doesNotContain("identitySource") + .doesNotContain("identityErrorMessage"); + }); + + batchExecutor.close(); + eventBus.close(); + } + @Test void malformedInboundMessageArchivesParseFailureMetadataAndPassthrough() throws Exception { MqttInboundProperties.Endpoint endpoint = new MqttInboundProperties.Endpoint(); @@ -381,6 +435,25 @@ class MqttEndpointManagerTest { } } + private static final class CountingIdentityResolver implements VehicleIdentityResolver { + private final VehicleIdentity identity; + private final AtomicInteger calls = new AtomicInteger(); + + private CountingIdentityResolver(VehicleIdentity identity) { + this.identity = identity; + } + + @Override + public VehicleIdentity resolve(VehicleIdentityLookup lookup) { + calls.incrementAndGet(); + return identity; + } + + private int calls() { + return calls.get(); + } + } + private static final class BrokenProfile implements com.lingniu.ingest.inbound.mqtt.profile.MqttProfile { @Override public String name() {