diff --git a/docs/operations/vehicle-ingest-tdengine-verification.md b/docs/operations/vehicle-ingest-tdengine-verification.md index f17cde8e..d5b18c1d 100644 --- a/docs/operations/vehicle-ingest-tdengine-verification.md +++ b/docs/operations/vehicle-ingest-tdengine-verification.md @@ -115,6 +115,7 @@ JT808 注册身份绑定： - 生产需要把 0x0100 注册信息维护到 MySQL 时，设置 `VEHICLE_IDENTITY_STORE=mysql`。 - 需要提供 `VEHICLE_IDENTITY_MYSQL_JDBC_URL`、`VEHICLE_IDENTITY_MYSQL_USERNAME`、`VEHICLE_IDENTITY_MYSQL_PASSWORD`。 - 服务启动时自动创建 `vehicle_identity_binding` 表，按 `protocol + identifier_type + identifier_value` 维护 VIN 绑定。 +- MySQL 绑定在启动时加载到内存索引，注册/补写时写穿 MySQL 并刷新内存索引；帧解析热路径只查内存，不按帧访问 MySQL。 - 绑定类型包括 `PHONE`、`DEVICE_ID`、`PLATE`；后续同一终端上报 raw/event 时会优先解析成已绑定 VIN。 健康检查： diff --git a/modules/core/vehicle-identity/src/main/java/com/lingniu/ingest/identity/MySqlVehicleIdentityService.java b/modules/core/vehicle-identity/src/main/java/com/lingniu/ingest/identity/MySqlVehicleIdentityService.java index 66a2da5a..634819d2 100644 --- a/modules/core/vehicle-identity/src/main/java/com/lingniu/ingest/identity/MySqlVehicleIdentityService.java +++ b/modules/core/vehicle-identity/src/main/java/com/lingniu/ingest/identity/MySqlVehicleIdentityService.java @@ -14,6 +14,7 @@ public final class MySqlVehicleIdentityService implements VehicleIdentityResolve private final DataSource dataSource; private final String table; + private final InMemoryVehicleIdentityService index = new InMemoryVehicleIdentityService(); public MySqlVehicleIdentityService(DataSource dataSource, String table) { if (dataSource == null) { @@ -22,6 +23,7 @@ public final class MySqlVehicleIdentityService implements VehicleIdentityResolve this.dataSource = dataSource; this.table = sanitizeTable(table); initializeSchema(); + loadIndex(); } @Override @@ -32,6 +34,7 @@ public final class MySqlVehicleIdentityService implements VehicleIdentityResolve upsert(binding.protocol(), "PHONE", binding.phone(), binding.vin()); upsert(binding.protocol(), "DEVICE_ID", binding.deviceId(), binding.vin()); upsert(binding.protocol(), "PLATE", binding.plate(), binding.vin()); + index.bind(binding); } @Override @@ -39,25 +42,7 @@ public final class MySqlVehicleIdentityService implements VehicleIdentityResolve if (lookup == null) { return new VehicleIdentity("unknown", false, VehicleIdentitySource.UNKNOWN); } - if (!lookup.vin().isBlank()) { - return new VehicleIdentity(lookup.vin(), true, VehicleIdentitySource.EXPLICIT_VIN); - } - VehicleIdentity phone = resolveBound(lookup.protocol(), "PHONE", lookup.phone(), VehicleIdentitySource.BOUND_PHONE); - if (phone != null) return phone; - VehicleIdentity device = resolveBound(lookup.protocol(), "DEVICE_ID", lookup.deviceId(), VehicleIdentitySource.BOUND_DEVICE_ID); - if (device != null) return device; - VehicleIdentity plate = resolveBound(lookup.protocol(), "PLATE", lookup.plate(), VehicleIdentitySource.BOUND_PLATE); - if (plate != null) return plate; - if (!lookup.deviceId().isBlank()) { - return new VehicleIdentity(lookup.deviceId(), false, VehicleIdentitySource.FALLBACK_DEVICE_ID); - } - if (!lookup.phone().isBlank()) { - return new VehicleIdentity(lookup.phone(), false, VehicleIdentitySource.FALLBACK_PHONE); - } - if (!lookup.plate().isBlank()) { - return new VehicleIdentity(lookup.plate(), false, VehicleIdentitySource.FALLBACK_PLATE); - } - return new VehicleIdentity("unknown", false, VehicleIdentitySource.UNKNOWN); + return index.resolve(lookup); } private void initializeSchema() { @@ -81,6 +66,39 @@ public final class MySqlVehicleIdentityService implements VehicleIdentityResolve } } + private void loadIndex() { + String sql = "SELECT protocol, identifier_type, identifier_value, vin FROM " + table; + try (Connection connection = dataSource.getConnection(); + PreparedStatement statement = connection.prepareStatement(sql); + ResultSet resultSet = statement.executeQuery()) { + while (resultSet.next()) { + bindLoadedRow( + protocol(resultSet.getString("protocol")), + resultSet.getString("identifier_type"), + resultSet.getString("identifier_value"), + resultSet.getString("vin")); + } + } catch (SQLException e) { + throw new IllegalStateException("vehicle identity mysql index load failed: " + table, e); + } + } + + private void bindLoadedRow(ProtocolId protocol, String identifierType, String identifierValue, String vin) { + if (vin == null || vin.isBlank()) { + return; + } + String type = identifierType == null ? "" : identifierType.trim().toUpperCase(Locale.ROOT); + VehicleIdentityBinding binding = switch (type) { + case "PHONE" -> new VehicleIdentityBinding(protocol, vin, identifierValue, "", ""); + case "DEVICE_ID" -> new VehicleIdentityBinding(protocol, vin, "", identifierValue, ""); + case "PLATE" -> new VehicleIdentityBinding(protocol, vin, "", "", identifierValue); + default -> null; + }; + if (binding != null) { + index.bind(binding); + } + } + private void upsert(ProtocolId protocol, String identifierType, String identifierValue, String vin) { String normalizedIdentifier = normalize(identifierValue); if (normalizedIdentifier.isBlank()) { @@ -101,32 +119,6 @@ public final class MySqlVehicleIdentityService implements VehicleIdentityResolve } } - private VehicleIdentity resolveBound(ProtocolId protocol, - String identifierType, - String identifierValue, - VehicleIdentitySource source) { - String normalizedIdentifier = normalize(identifierValue); - if (normalizedIdentifier.isBlank()) { - return null; - } - String sql = "SELECT vin FROM " + table - + " WHERE protocol = ? AND identifier_type = ? AND identifier_value = ? LIMIT 1"; - try (Connection connection = dataSource.getConnection(); - PreparedStatement statement = connection.prepareStatement(sql)) { - statement.setString(1, protocolName(protocol)); - statement.setString(2, identifierType); - statement.setString(3, normalizedIdentifier); - try (ResultSet resultSet = statement.executeQuery()) { - if (resultSet.next()) { - return new VehicleIdentity(resultSet.getString(1), true, source); - } - } - } catch (SQLException e) { - throw new IllegalStateException("vehicle identity mysql resolve failed: " + table, e); - } - return null; - } - private static String sanitizeTable(String table) { String value = table == null || table.isBlank() ? "vehicle_identity_binding" : table.trim(); if (!value.matches("[A-Za-z0-9_]+")) { @@ -139,6 +131,17 @@ public final class MySqlVehicleIdentityService implements VehicleIdentityResolve return protocol == null ? "UNKNOWN" : protocol.name(); } + private static ProtocolId protocol(String value) { + if (value == null || value.isBlank()) { + return ProtocolId.UNKNOWN; + } + try { + return ProtocolId.valueOf(value.trim().toUpperCase(Locale.ROOT)); + } catch (IllegalArgumentException ignored) { + return ProtocolId.UNKNOWN; + } + } + private static String normalize(String value) { return value == null ? "" : value.trim().toUpperCase(Locale.ROOT); } diff --git a/modules/core/vehicle-identity/src/test/java/com/lingniu/ingest/identity/MySqlVehicleIdentityServiceTest.java b/modules/core/vehicle-identity/src/test/java/com/lingniu/ingest/identity/MySqlVehicleIdentityServiceTest.java index a5ac95b2..f4326f47 100644 --- a/modules/core/vehicle-identity/src/test/java/com/lingniu/ingest/identity/MySqlVehicleIdentityServiceTest.java +++ b/modules/core/vehicle-identity/src/test/java/com/lingniu/ingest/identity/MySqlVehicleIdentityServiceTest.java @@ -62,9 +62,37 @@ class MySqlVehicleIdentityServiceTest { assertThat(jdbc.bindings).isEmpty(); } + @Test + void resolvesFromStartupCacheWithoutPerFrameDatabaseLookup() { + RecordingIdentityJdbc jdbc = new RecordingIdentityJdbc(); + jdbc.seed("JT808", "PHONE", "13079960001", "LNVIN000000000123"); + jdbc.seed("JT808", "DEVICE_ID", "DEV001", "LNVIN000000000123"); + jdbc.seed("JT808", "PLATE", "粤B00001", "LNVIN000000000123"); + MySqlVehicleIdentityService service = new MySqlVehicleIdentityService(jdbc.dataSource(), "vehicle_identity_binding"); + + VehicleIdentity byPhone = service.resolve(new VehicleIdentityLookup( + ProtocolId.JT808, "", "13079960001", "", "")); + VehicleIdentity byDevice = service.resolve(new VehicleIdentityLookup( + ProtocolId.JT808, "", "", "DEV001", "")); + VehicleIdentity byPlate = service.resolve(new VehicleIdentityLookup( + ProtocolId.JT808, "", "", "", "粤B00001")); + + assertThat(byPhone.vin()).isEqualTo("LNVIN000000000123"); + assertThat(byDevice.vin()).isEqualTo("LNVIN000000000123"); + assertThat(byPlate.vin()).isEqualTo("LNVIN000000000123"); + assertThat(jdbc.loadAllSelects).isEqualTo(1); + assertThat(jdbc.lookupSelects).isZero(); + } + private static final class RecordingIdentityJdbc { private final List createdTables = new ArrayList<>(); private final Map bindings = new HashMap<>(); + private int loadAllSelects; + private int lookupSelects; + + private void seed(String protocol, String identifierType, String identifierValue, String vin) { + bindings.put(protocol + "|" + identifierType + "|" + normalize(identifierValue), vin); + } private DataSource dataSource() { return new DataSource() { @@ -122,6 +150,11 @@ class MySqlVehicleIdentityServiceTest { return 1; } case "executeQuery" -> { + if (sql.startsWith("SELECT protocol, identifier_type, identifier_value, vin FROM ")) { + loadAllSelects++; + return allBindingsResultSet(); + } + lookupSelects++; return resultSet(resolve(params)); } case "close" -> { @@ -151,6 +184,40 @@ class MySqlVehicleIdentityServiceTest { return bindings.get(protocol + "|" + identifierType + "|" + identifierValue); } + private ResultSet allBindingsResultSet() { + List> entries = new ArrayList<>(bindings.entrySet()); + InvocationHandler handler = new InvocationHandler() { + private int index = -1; + + @Override + public Object invoke(Object proxy, java.lang.reflect.Method method, Object[] args) { + return switch (method.getName()) { + case "next" -> { + index++; + yield index < entries.size(); + } + case "getString" -> getColumn(entries.get(index), args[0]); + case "close" -> null; + default -> defaultValue(method.getReturnType()); + }; + } + }; + return (ResultSet) Proxy.newProxyInstance( + ResultSet.class.getClassLoader(), new Class[]{ResultSet.class}, handler); + } + + private static String getColumn(Map.Entry entry, Object column) { + String[] parts = entry.getKey().split("\\|", 3); + String name = column.toString(); + return switch (name) { + case "protocol" -> parts[0]; + case "identifier_type" -> parts[1]; + case "identifier_value" -> parts[2]; + case "vin" -> entry.getValue(); + default -> throw new IllegalArgumentException("unsupported column " + name); + }; + } + private static ResultSet resultSet(String vin) { InvocationHandler handler = new InvocationHandler() { private boolean read; diff --git a/modules/core/vehicle-identity/src/test/java/com/lingniu/ingest/identity/config/VehicleIdentityAutoConfigurationTest.java b/modules/core/vehicle-identity/src/test/java/com/lingniu/ingest/identity/config/VehicleIdentityAutoConfigurationTest.java index 1bc8a183..72588866 100644 --- a/modules/core/vehicle-identity/src/test/java/com/lingniu/ingest/identity/config/VehicleIdentityAutoConfigurationTest.java +++ b/modules/core/vehicle-identity/src/test/java/com/lingniu/ingest/identity/config/VehicleIdentityAutoConfigurationTest.java @@ -10,8 +10,10 @@ import org.springframework.boot.test.context.runner.ApplicationContextRunner; import javax.sql.DataSource; import java.nio.file.Path; import java.sql.Connection; -import java.sql.Statement; +import java.sql.PreparedStatement; +import java.sql.ResultSet; import java.sql.SQLException; +import java.sql.Statement; import java.lang.reflect.Proxy; import static org.assertj.core.api.Assertions.assertThat; @@ -78,6 +80,7 @@ class VehicleIdentityAutoConfigurationTest { return (Connection) Proxy.newProxyInstance(Connection.class.getClassLoader(), new Class[]{Connection.class}, (proxy, method, args) -> switch (method.getName()) { case "createStatement" -> statement(); + case "prepareStatement" -> preparedStatement(); case "close" -> null; case "isClosed" -> false; default -> defaultValue(method.getReturnType()); @@ -93,6 +96,25 @@ class VehicleIdentityAutoConfigurationTest { }); } + private static PreparedStatement preparedStatement() { + return (PreparedStatement) Proxy.newProxyInstance( + PreparedStatement.class.getClassLoader(), new Class[]{PreparedStatement.class}, + (proxy, method, args) -> switch (method.getName()) { + case "executeQuery" -> emptyResultSet(); + case "close" -> null; + default -> defaultValue(method.getReturnType()); + }); + } + + private static ResultSet emptyResultSet() { + return (ResultSet) Proxy.newProxyInstance(ResultSet.class.getClassLoader(), new Class[]{ResultSet.class}, + (proxy, method, args) -> switch (method.getName()) { + case "next" -> false; + case "close" -> null; + default -> defaultValue(method.getReturnType()); + }); + } + private static Object defaultValue(Class type) { if (type == Void.TYPE) return null; if (type == Boolean.TYPE) return false;