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

perf: cache mysql vehicle identity lookups
All checks were successful
ci/woodpecker/push/woodpecker Pipeline was successful
Details

Browse Source
This commit is contained in:
lingniu
2026-06-29 17:00:54 +08:00
parent 114e707a5b
commit f00d851f2a
4 changed files with 139 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
docs/operations/vehicle-ingest-tdengine-verification.md
View File

@@ -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。
健康检查：

93
modules/core/vehicle-identity/src/main/java/com/lingniu/ingest/identity/MySqlVehicleIdentityService.java
View File

@@ -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);
}

67
modules/core/vehicle-identity/src/test/java/com/lingniu/ingest/identity/MySqlVehicleIdentityServiceTest.java
View File

@@ -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<String> createdTables = new ArrayList<>();
private final Map<String, String> 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<Map.Entry<String, String>> 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<String, String> 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;

24
modules/core/vehicle-identity/src/test/java/com/lingniu/ingest/identity/config/VehicleIdentityAutoConfigurationTest.java
View File

@@ -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;