diff --git a/deploy/local/launchctl/README.md b/deploy/local/launchctl/README.md index b10abbca..b7067c6e 100644 --- a/deploy/local/launchctl/README.md +++ b/deploy/local/launchctl/README.md @@ -62,7 +62,7 @@ history 消费者会按协议和 raw/event 自动拆分 consumer group。设置 - `vehicle-history-live-jt808-event` - `vehicle-history-live-jt808-raw` -JT808 身份绑定本地模板默认使用 `file`，可以在没有 MySQL 时持久化人工绑定，避免重启后退回 `unknown`。需要把 0x0100 注册信息维护到 MySQL 并支持真实 VIN 反写时，将 `com.lingniu.jt808.plist.template` 里的 `VEHICLE_IDENTITY_STORE` 改为 `mysql`，并提供 `VEHICLE_IDENTITY_MYSQL_*` 连接参数。服务会自动创建 `vehicle_identity_binding` 和 `vehicle_identity_binding_registration` 两张表：前者保存已确认 VIN 的外部标识绑定，后者保存 0x0100 注册字段；外部系统可把真实 VIN 反写到 registration 表，服务按刷新周期加载后，后续 raw/event 会带真实 VIN。`memory` 仅适合一次性联调，不建议用于生产验证。 +JT808 身份绑定本地模板默认使用 `file`，可以在没有 MySQL 时持久化人工绑定，避免重启后退回 `unknown`。需要把 0x0100 注册信息维护到 MySQL 并支持真实 VIN 反写时，将 `com.lingniu.jt808.plist.template` 里的 `VEHICLE_IDENTITY_STORE` 改为 `mysql`，并提供 `VEHICLE_IDENTITY_MYSQL_*` 连接参数。服务会自动创建 `vehicle_identity_binding` 和 `vehicle_identity_binding_registration` 两张表：前者只保存 `plate`、`vin`，后者保存 0x0100 注册字段；服务按刷新周期用注册表里的车牌关联 VIN，后续 raw/event 会带真实 VIN。`memory` 仅适合一次性联调，不建议用于生产验证。 ## 启停 diff --git a/docs/operations/vehicle-ingest-tdengine-verification.md b/docs/operations/vehicle-ingest-tdengine-verification.md index 863c37be..82c1e3d8 100644 --- a/docs/operations/vehicle-ingest-tdengine-verification.md +++ b/docs/operations/vehicle-ingest-tdengine-verification.md @@ -121,10 +121,10 @@ JT808 注册身份绑定： - 默认可以使用 `VEHICLE_IDENTITY_STORE=file` 或 `memory`。 - 生产需要把 0x0100 注册信息维护到 MySQL 时，设置 `VEHICLE_IDENTITY_STORE=mysql`。 - 需要提供 `VEHICLE_IDENTITY_MYSQL_JDBC_URL`、`VEHICLE_IDENTITY_MYSQL_USERNAME`、`VEHICLE_IDENTITY_MYSQL_PASSWORD`。 -- 服务启动时自动创建 `vehicle_identity_binding` 和 `vehicle_identity_binding_registration` 两张表。`vehicle_identity_binding` 按 `protocol + identifier_type + identifier_value` 维护已确认 VIN 绑定；`vehicle_identity_binding_registration` 按 `protocol + phone` 保存 JT808 0x0100 注册字段。 -- 注册帧无真实 VIN 时也会写入 registration 表，`vin` 默认为 `unknown`。外部系统识别车辆后可反写真实 VIN，例如：`UPDATE vehicle_identity_binding_registration SET vin = 'LNVIN000000000001' WHERE protocol = 'JT808' AND phone = '13079960003';` -- MySQL 绑定在启动时加载到内存索引，registration 表反写 VIN 后会按 `VEHICLE_IDENTITY_MYSQL_REFRESH_INTERVAL` 周期刷新，默认 `60s`；帧解析热路径只查内存，不按帧访问 MySQL。 -- 绑定/注册类型包括 `PHONE`、`DEVICE_ID`、`PLATE`；后续同一终端上报 raw/event 时会优先解析成已绑定或已反写 VIN。 +- 服务启动时自动创建 `vehicle_identity_binding` 和 `vehicle_identity_binding_registration` 两张表。`vehicle_identity_binding` 只维护 `plate`、`vin` 两列；`vehicle_identity_binding_registration` 按 `protocol + phone` 保存 JT808 0x0100 注册字段。 +- 注册帧无真实 VIN 时也会写入 registration 表，`vin` 默认为 `unknown`。外部系统识别车辆后只需要维护车牌和 VIN，例如：`INSERT INTO vehicle_identity_binding (plate, vin) VALUES ('沪A61559F', 'LNVIN000000000001') ON DUPLICATE KEY UPDATE vin = VALUES(vin);` +- MySQL 绑定在启动时加载到内存索引，`plate/vin` 反写后会按 `VEHICLE_IDENTITY_MYSQL_REFRESH_INTERVAL` 周期刷新，默认 `60s`；帧解析热路径只查内存，不按帧访问 MySQL。 +- 后续同一终端上报 raw/event 时，服务会用 registration 表里的 `phone/device_id/plate` 加绑定表里的 `plate/vin` 解析成真实 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 44610e45..5785304b 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 @@ -47,9 +47,7 @@ public final class MySqlVehicleIdentityService implements VehicleIdentityResolve if (binding == null || !binding.hasResolvedVin()) { return; } - 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()); + upsert(binding.plate(), binding.vin()); index.bind(binding); } @@ -91,13 +89,9 @@ public final class MySqlVehicleIdentityService implements VehicleIdentityResolve private void initializeSchema() { String sql = """ CREATE TABLE IF NOT EXISTS %s ( - protocol VARCHAR(32) NOT NULL, - identifier_type VARCHAR(32) NOT NULL, - identifier_value VARCHAR(128) NOT NULL, + plate VARCHAR(128) NOT NULL, vin VARCHAR(64) NOT NULL, - updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, - created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, - PRIMARY KEY (protocol, identifier_type, identifier_value), + PRIMARY KEY (plate), KEY idx_vehicle_identity_vin (vin) ) """.formatted(table); @@ -132,17 +126,17 @@ public final class MySqlVehicleIdentityService implements VehicleIdentityResolve private InMemoryVehicleIdentityService loadIndex() { InMemoryVehicleIdentityService loaded = new InMemoryVehicleIdentityService(); - String sql = "SELECT protocol, identifier_type, identifier_value, vin FROM " + table; + String sql = "SELECT plate, vin FROM " + table; try (Connection connection = dataSource.getConnection(); PreparedStatement statement = connection.prepareStatement(sql); ResultSet resultSet = statement.executeQuery()) { while (resultSet.next()) { - bindLoadedRow( - loaded, - protocol(resultSet.getString("protocol")), - resultSet.getString("identifier_type"), - resultSet.getString("identifier_value"), - resultSet.getString("vin")); + loaded.bind(new VehicleIdentityBinding( + ProtocolId.JT808, + resultSet.getString("vin"), + "", + "", + resultSet.getString("plate"))); } loadRegistrationIndex(loaded, connection); return loaded; @@ -152,6 +146,24 @@ public final class MySqlVehicleIdentityService implements VehicleIdentityResolve } private void loadRegistrationIndex(InMemoryVehicleIdentityService loaded, Connection connection) throws SQLException { + String sql = "SELECT r.protocol, r.phone, r.device_id, r.plate, b.vin FROM " + registrationTable + " r " + + "JOIN " + table + " b ON b.plate = r.plate " + + "WHERE b.vin IS NOT NULL AND b.vin <> '' AND LOWER(b.vin) <> 'unknown'"; + try (PreparedStatement statement = connection.prepareStatement(sql); + ResultSet resultSet = statement.executeQuery()) { + while (resultSet.next()) { + loaded.bind(new VehicleIdentityBinding( + protocol(resultSet.getString("protocol")), + resultSet.getString("vin"), + resultSet.getString("phone"), + resultSet.getString("device_id"), + resultSet.getString("plate"))); + } + } + loadLegacyRegistrationVinIndex(loaded, connection); + } + + private void loadLegacyRegistrationVinIndex(InMemoryVehicleIdentityService loaded, Connection connection) throws SQLException { String sql = "SELECT protocol, phone, device_id, plate, vin FROM " + registrationTable + " WHERE vin IS NOT NULL AND vin <> '' AND LOWER(vin) <> 'unknown'"; try (PreparedStatement statement = connection.prepareStatement(sql); @@ -167,26 +179,6 @@ public final class MySqlVehicleIdentityService implements VehicleIdentityResolve } } - private void bindLoadedRow(InMemoryVehicleIdentityService loaded, - 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) { - loaded.bind(binding); - } - } - private ScheduledExecutorService startRefresher(Duration refreshInterval) { if (refreshInterval == null || refreshInterval.isZero() || refreshInterval.isNegative()) { return null; @@ -209,20 +201,18 @@ public final class MySqlVehicleIdentityService implements VehicleIdentityResolve } } - private void upsert(ProtocolId protocol, String identifierType, String identifierValue, String vin) { - String normalizedIdentifier = normalize(identifierValue); - if (normalizedIdentifier.isBlank()) { + private void upsert(String plate, String vin) { + String normalizedPlate = normalize(plate); + if (normalizedPlate.isBlank()) { return; } - String sql = "INSERT INTO " + table + " (protocol, identifier_type, identifier_value, vin) " - + "VALUES (?, ?, ?, ?) " - + "ON DUPLICATE KEY UPDATE vin = VALUES(vin), updated_at = CURRENT_TIMESTAMP"; + String sql = "INSERT INTO " + table + " (plate, vin) " + + "VALUES (?, ?) " + + "ON DUPLICATE KEY UPDATE vin = VALUES(vin)"; try (Connection connection = dataSource.getConnection(); PreparedStatement statement = connection.prepareStatement(sql)) { - statement.setString(1, protocolName(protocol)); - statement.setString(2, identifierType); - statement.setString(3, normalizedIdentifier); - statement.setString(4, vin.trim()); + statement.setString(1, normalizedPlate); + statement.setString(2, vin.trim()); statement.executeUpdate(); } catch (SQLException e) { throw new IllegalStateException("vehicle identity mysql bind failed: " + table, e); diff --git a/modules/core/vehicle-identity/src/test/java/com/lingniu/ingest/identity/MySqlVehicleIdentityServiceJdbcTest.java b/modules/core/vehicle-identity/src/test/java/com/lingniu/ingest/identity/MySqlVehicleIdentityServiceJdbcTest.java index 04d2f4b9..0dd1a7d9 100644 --- a/modules/core/vehicle-identity/src/test/java/com/lingniu/ingest/identity/MySqlVehicleIdentityServiceJdbcTest.java +++ b/modules/core/vehicle-identity/src/test/java/com/lingniu/ingest/identity/MySqlVehicleIdentityServiceJdbcTest.java @@ -7,6 +7,8 @@ import org.junit.jupiter.api.Test; import java.sql.Connection; import java.sql.ResultSet; import java.sql.Statement; +import java.util.ArrayList; +import java.util.List; import java.util.UUID; import static org.assertj.core.api.Assertions.assertThat; @@ -14,7 +16,7 @@ import static org.assertj.core.api.Assertions.assertThat; class MySqlVehicleIdentityServiceJdbcTest { @Test - void registrationVinWriteBackIsQueryableThroughRealJdbc() throws Exception { + void plateVinBindingIsQueryableThroughRealJdbc() throws Exception { JdbcDataSource dataSource = new JdbcDataSource(); dataSource.setURL("jdbc:h2:mem:" + UUID.randomUUID() + ";MODE=MySQL;DATABASE_TO_UPPER=false;DB_CLOSE_DELAY=-1"); @@ -50,10 +52,17 @@ class MySqlVehicleIdentityServiceJdbcTest { assertThat(registered.getString("device_type")).isEqualTo("TYPE-A"); assertThat(registered.getInt("plate_color")).isEqualTo(1); + List columns = new ArrayList<>(); + ResultSet bindingColumns = connection.getMetaData() + .getColumns(null, null, "vehicle_identity_binding", null); + while (bindingColumns.next()) { + columns.add(bindingColumns.getString("COLUMN_NAME")); + } + assertThat(columns).containsExactly("plate", "vin"); + statement.executeUpdate(""" - UPDATE vehicle_identity_binding_registration - SET vin = 'LNVIN00000061001' - WHERE protocol = 'JT808' AND phone = '13079961001' + INSERT INTO vehicle_identity_binding (plate, vin) + VALUES ('粤B61001', 'LNVIN00000061001') """); } 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 a007db99..e7815d41 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 @@ -41,7 +41,7 @@ class MySqlVehicleIdentityServiceTest { assertThat(jdbc.createdTables) .anySatisfy(sql -> assertThat(sql).contains("CREATE TABLE IF NOT EXISTS vehicle_identity_binding")); - assertThat(jdbc.bindings).containsKey("JT808|PHONE|13079960001"); + assertThat(jdbc.bindings).containsEntry("粤B00001", "LNVIN000000000123"); assertThat(byPhone.vin()).isEqualTo("LNVIN000000000123"); assertThat(byPhone.resolved()).isTrue(); assertThat(byPhone.source()).isEqualTo(VehicleIdentitySource.BOUND_PHONE); @@ -65,9 +65,8 @@ class MySqlVehicleIdentityServiceTest { @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"); + jdbc.seedRegistration("JT808", "13079960001", "DEV001", "粤B00001"); + jdbc.seed("粤B00001", "LNVIN000000000123"); MySqlVehicleIdentityService service = new MySqlVehicleIdentityService(jdbc.dataSource(), "vehicle_identity_binding"); VehicleIdentity byPhone = service.resolve(new VehicleIdentityLookup( @@ -87,12 +86,13 @@ class MySqlVehicleIdentityServiceTest { @Test void refreshLoadsExternallyInsertedBindingWithoutPerFrameDatabaseLookup() { RecordingIdentityJdbc jdbc = new RecordingIdentityJdbc(); + jdbc.seedRegistration("JT808", "13079960002", "DEV002", "粤B00002"); MySqlVehicleIdentityService service = new MySqlVehicleIdentityService(jdbc.dataSource(), "vehicle_identity_binding"); VehicleIdentity before = service.resolve(new VehicleIdentityLookup( ProtocolId.JT808, "", "13079960002", "", "")); - jdbc.seed("JT808", "PHONE", "13079960002", "LNVIN000000000456"); + jdbc.seed("粤B00002", "LNVIN000000000456"); service.refresh(); VehicleIdentity after = service.resolve(new VehicleIdentityLookup( ProtocolId.JT808, "", "13079960002", "", "")); @@ -106,7 +106,7 @@ class MySqlVehicleIdentityServiceTest { } @Test - void storesRegistrationWithoutVinAndResolvesAfterVinIsWrittenBack() { + void storesRegistrationWithoutVinAndResolvesAfterPlateVinIsWrittenBack() { RecordingIdentityJdbc jdbc = new RecordingIdentityJdbc(); MySqlVehicleIdentityService service = new MySqlVehicleIdentityService(jdbc.dataSource(), "vehicle_identity_binding"); @@ -125,7 +125,7 @@ class MySqlVehicleIdentityServiceTest { VehicleIdentity before = service.resolve(new VehicleIdentityLookup( ProtocolId.JT808, "", "13079960003", "", "")); - jdbc.writeBackRegistrationVin("JT808", "13079960003", "LNVIN000000000789"); + jdbc.seed("粤B00003", "LNVIN000000000789"); service.refresh(); VehicleIdentity byPhone = service.resolve(new VehicleIdentityLookup( ProtocolId.JT808, "", "13079960003", "", "")); @@ -137,7 +137,7 @@ class MySqlVehicleIdentityServiceTest { assertThat(jdbc.createdTables) .anySatisfy(sql -> assertThat(sql).contains("CREATE TABLE IF NOT EXISTS vehicle_identity_binding_registration")); assertThat(jdbc.registrations).containsKey("JT808|13079960003"); - assertThat(jdbc.bindings).isEmpty(); + assertThat(jdbc.bindings).containsEntry("粤B00003", "LNVIN000000000789"); assertThat(before.resolved()).isFalse(); assertThat(byPhone.vin()).isEqualTo("LNVIN000000000789"); assertThat(byPhone.source()).isEqualTo(VehicleIdentitySource.BOUND_PHONE); @@ -152,16 +152,13 @@ class MySqlVehicleIdentityServiceTest { 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 void seed(String plate, String vin) { + bindings.put(normalize(plate), vin); } - private void writeBackRegistrationVin(String protocol, String phone, String vin) { - RegistrationRow row = registrations.get(protocol + "|" + normalize(phone)); - if (row == null) { - throw new IllegalArgumentException("missing registration " + phone); - } - row.vin = vin; + private void seedRegistration(String protocol, String phone, String deviceId, String plate) { + registrations.put(protocol + "|" + normalize(phone), + new RegistrationRow(protocol, normalize(phone), "unknown", normalize(deviceId), normalize(plate))); } private DataSource dataSource() { @@ -224,13 +221,20 @@ class MySqlVehicleIdentityServiceTest { return 1; } case "executeQuery" -> { - if (sql.startsWith("SELECT protocol, identifier_type, identifier_value, vin FROM ")) { + if (sql.startsWith("SELECT plate, vin FROM ")) { loadAllSelects++; return allBindingsResultSet(); } - if (sql.startsWith("SELECT protocol, phone, device_id, plate, vin FROM ")) { + if (sql.startsWith("SELECT protocol, identifier_type, identifier_value, vin FROM ")) { + loadAllSelects++; + return legacyAllBindingsResultSet(); + } + if (sql.startsWith("SELECT r.protocol, r.phone, r.device_id, r.plate, b.vin FROM ")) { return registrationResultSet(); } + if (sql.startsWith("SELECT protocol, phone, device_id, plate, vin FROM ")) { + return legacyRegistrationResultSet(); + } lookupSelects++; return resultSet(resolve(params)); } @@ -247,11 +251,16 @@ class MySqlVehicleIdentityServiceTest { } private void upsert(List params) { - String protocol = (String) params.get(0); - String identifierType = (String) params.get(1); - String identifierValue = normalize((String) params.get(2)); - String vin = (String) params.get(3); - bindings.put(protocol + "|" + identifierType + "|" + identifierValue, vin); + String plate; + String vin; + if (params.size() >= 4) { + plate = normalize((String) params.get(2)); + vin = (String) params.get(3); + } else { + plate = normalize((String) params.get(0)); + vin = (String) params.get(1); + } + bindings.put(plate, vin); } private void upsertRegistration(List params) { @@ -273,7 +282,10 @@ class MySqlVehicleIdentityServiceTest { String protocol = (String) params.get(0); String identifierType = (String) params.get(1); String identifierValue = normalize((String) params.get(2)); - return bindings.get(protocol + "|" + identifierType + "|" + identifierValue); + if (!"PLATE".equals(identifierType)) { + return null; + } + return bindings.get(identifierValue); } private ResultSet allBindingsResultSet() { @@ -298,22 +310,63 @@ class MySqlVehicleIdentityServiceTest { ResultSet.class.getClassLoader(), new Class[]{ResultSet.class}, handler); } + private ResultSet legacyAllBindingsResultSet() { + 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" -> legacyColumn(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 "plate" -> entry.getKey(); case "vin" -> entry.getValue(); default -> throw new IllegalArgumentException("unsupported column " + name); }; } + private static String legacyColumn(Map.Entry entry, Object column) { + return switch (column.toString()) { + case "protocol" -> "JT808"; + case "identifier_type" -> "PLATE"; + case "identifier_value" -> entry.getKey(); + case "vin" -> entry.getValue(); + default -> throw new IllegalArgumentException("unsupported column " + column); + }; + } + private ResultSet registrationResultSet() { + List rows = registrations.values().stream() + .filter(row -> bindings.containsKey(row.plate)) + .map(row -> new RegistrationRow(row.protocol, row.phone, bindings.get(row.plate), row.deviceId, row.plate)) + .toList(); + return registrationRowsResultSet(rows); + } + + private ResultSet legacyRegistrationResultSet() { List rows = registrations.values().stream() .filter(row -> row.vin != null && !row.vin.isBlank() && !"unknown".equalsIgnoreCase(row.vin)) .toList(); + return registrationRowsResultSet(rows); + } + + private ResultSet registrationRowsResultSet(List rows) { InvocationHandler handler = new InvocationHandler() { private int index = -1;