diff --git a/README.md b/README.md index 9f5b84cd..a8c7b401 100644 --- a/README.md +++ b/README.md @@ -37,7 +37,7 @@ lingniu-vehicle-ingest/ │ │ ├── ingest-codec-common/ 公共编解码工具（BCD/CRC/BCC/bit utils） │ │ ├── ingest-core/ Pipeline / Dispatcher / Disruptor / Session 桥 │ │ ├── session-core/ 设备会话 + 鉴权 + Token + Redis/Memory SessionStore -│ │ ├── vehicle-identity/ 跨协议车辆身份解析 + 外部标识绑定（memory/file） +│ │ ├── vehicle-identity/ 跨协议车辆身份解析 + 外部标识绑定（memory/file/mysql） │ │ └── observability/ metrics / tracing / health │ ├── protocols/ │ │ ├── protocol-gb32960/ GB/T 32960 diff --git a/deploy/local/launchctl/README.md b/deploy/local/launchctl/README.md index 51d68cad..bedafe94 100644 --- a/deploy/local/launchctl/README.md +++ b/deploy/local/launchctl/README.md @@ -48,7 +48,7 @@ done 如果生产配置由 Nacos 下发，可以把模板里的 `NACOS_CONFIG_ENABLED` 改成 `true`，并补充 `NACOS_SERVER_ADDR`、`NACOS_NAMESPACE`、`NACOS_GROUP`、`NACOS_USERNAME`、`NACOS_PASSWORD`。端口、Kafka topic、archive root、TDengine 连接建议仍保留为启动环境变量，便于 Portainer、launchctl 和临时压测保持一致。 -JT808 身份绑定默认使用内存或文件。需要把注册信息维护到 MySQL 时，将 `com.lingniu.jt808.plist.template` 里的 `VEHICLE_IDENTITY_STORE` 改为 `mysql`，并提供 `VEHICLE_IDENTITY_MYSQL_*` 连接参数。服务会自动创建 `vehicle_identity_binding` 表并按 phone/deviceId/plate 维护 VIN 绑定。 +JT808 身份绑定默认使用内存或文件。需要把注册信息维护到 MySQL 时，将 `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。 ## 启停 diff --git a/docs/operations/vehicle-ingest-tdengine-verification.md b/docs/operations/vehicle-ingest-tdengine-verification.md index 3a36fd27..649ad1a7 100644 --- a/docs/operations/vehicle-ingest-tdengine-verification.md +++ b/docs/operations/vehicle-ingest-tdengine-verification.md @@ -114,9 +114,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` 表，按 `protocol + identifier_type + identifier_value` 维护 VIN 绑定。 -- MySQL 绑定在启动时加载到内存索引，注册/补写时写穿 MySQL 并刷新内存索引；帧解析热路径只查内存，不按帧访问 MySQL。外部系统更新绑定后，后台按 `VEHICLE_IDENTITY_MYSQL_REFRESH_INTERVAL` 周期刷新，默认 `60s`。 -- 绑定类型包括 `PHONE`、`DEVICE_ID`、`PLATE`；后续同一终端上报 raw/event 时会优先解析成已绑定 VIN。 +- 服务启动时自动创建 `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。 健康检查： @@ -295,4 +296,4 @@ curl -sS 'http://127.0.0.1:20200/api/event-history/gb32960/snapshots/fields?vin= - 当前本机使用 TDengine WebSocket JDBC，例如 `jdbc:TAOS-WS://:6041/vehicle_ts`。 - `KAFKA_CONSUMER_AUTO_OFFSET_RESET=latest` 适合生产接入新流量；如果要回放历史 Kafka 数据，需要切换 consumer group 或重置 offset。 - `vehicle-history-app` raw 和 event 使用不同 consumer binding；raw consumer 必须开启，否则 `raw_frames` 不会持续增长。 -- JT808 设备没有 VIN 映射时，`vehicle_key` 使用 `jt808:`；启用 MySQL identity store 后，注册维护的 phone/deviceId/plate 绑定会用于后续 raw/event 的 VIN 解析。 +- JT808 设备没有 VIN 映射时，`vehicle_key` 使用 `jt808:`；启用 MySQL identity store 后，registration 表中已反写 VIN 的 phone/deviceId/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 4f31745e..44610e45 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 @@ -22,6 +22,7 @@ public final class MySqlVehicleIdentityService implements VehicleIdentityResolve private final DataSource dataSource; private final String table; + private final String registrationTable; private final ScheduledExecutorService refresher; private volatile InMemoryVehicleIdentityService index; @@ -35,6 +36,7 @@ public final class MySqlVehicleIdentityService implements VehicleIdentityResolve } this.dataSource = dataSource; this.table = sanitizeTable(table); + this.registrationTable = this.table + "_registration"; initializeSchema(); this.index = loadIndex(); this.refresher = startRefresher(refreshInterval); @@ -51,6 +53,22 @@ public final class MySqlVehicleIdentityService implements VehicleIdentityResolve index.bind(binding); } + @Override + public void register(VehicleRegistrationBinding registration) { + if (registration == null) { + return; + } + upsertRegistration(registration); + if (registration.hasResolvedVin()) { + bind(new VehicleIdentityBinding( + registration.protocol(), + registration.vin(), + registration.phone(), + registration.deviceId(), + registration.plate())); + } + } + @Override public VehicleIdentity resolve(VehicleIdentityLookup lookup) { if (lookup == null) { @@ -83,9 +101,30 @@ public final class MySqlVehicleIdentityService implements VehicleIdentityResolve KEY idx_vehicle_identity_vin (vin) ) """.formatted(table); + String registrationSql = """ + CREATE TABLE IF NOT EXISTS %s ( + protocol VARCHAR(32) NOT NULL, + phone VARCHAR(64) NOT NULL, + vin VARCHAR(64) NOT NULL DEFAULT 'unknown', + device_id VARCHAR(128) NOT NULL DEFAULT '', + plate VARCHAR(128) NOT NULL DEFAULT '', + province INT NULL, + city INT NULL, + maker VARCHAR(64) NOT NULL DEFAULT '', + device_type VARCHAR(128) NOT NULL DEFAULT '', + plate_color INT NULL, + last_registered_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, + created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, + PRIMARY KEY (protocol, phone), + KEY idx_vehicle_registration_vin (vin), + KEY idx_vehicle_registration_device (protocol, device_id), + KEY idx_vehicle_registration_plate (protocol, plate) + ) + """.formatted(registrationTable); try (Connection connection = dataSource.getConnection(); Statement statement = connection.createStatement()) { statement.execute(sql); + statement.execute(registrationSql); } catch (SQLException e) { throw new IllegalStateException("vehicle identity mysql schema initialize failed: " + table, e); } @@ -105,12 +144,29 @@ public final class MySqlVehicleIdentityService implements VehicleIdentityResolve resultSet.getString("identifier_value"), resultSet.getString("vin")); } + loadRegistrationIndex(loaded, connection); return loaded; } catch (SQLException e) { throw new IllegalStateException("vehicle identity mysql index load failed: " + table, e); } } + private void loadRegistrationIndex(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); + 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"))); + } + } + } + private void bindLoadedRow(InMemoryVehicleIdentityService loaded, ProtocolId protocol, String identifierType, @@ -173,6 +229,46 @@ public final class MySqlVehicleIdentityService implements VehicleIdentityResolve } } + private void upsertRegistration(VehicleRegistrationBinding registration) { + String phone = normalize(registration.phone()); + if (phone.isBlank()) { + return; + } + String vin = registration.hasResolvedVin() ? registration.vin().trim() : "unknown"; + String sql = "INSERT INTO " + registrationTable + + " (protocol, phone, vin, device_id, plate, province, city, maker, device_type, plate_color) " + + "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) " + + "ON DUPLICATE KEY UPDATE " + + "vin = CASE WHEN VALUES(vin) <> 'unknown' THEN VALUES(vin) ELSE vin END, " + + "device_id = VALUES(device_id), plate = VALUES(plate), province = VALUES(province), " + + "city = VALUES(city), maker = VALUES(maker), device_type = VALUES(device_type), " + + "plate_color = VALUES(plate_color), last_registered_at = CURRENT_TIMESTAMP"; + try (Connection connection = dataSource.getConnection(); + PreparedStatement statement = connection.prepareStatement(sql)) { + statement.setString(1, protocolName(registration.protocol())); + statement.setString(2, phone); + statement.setString(3, vin); + statement.setString(4, normalize(registration.deviceId())); + statement.setString(5, normalize(registration.plate())); + setInteger(statement, 6, registration.province()); + setInteger(statement, 7, registration.city()); + statement.setString(8, normalize(registration.maker())); + statement.setString(9, normalize(registration.deviceType())); + setInteger(statement, 10, registration.plateColor()); + statement.executeUpdate(); + } catch (SQLException e) { + throw new IllegalStateException("vehicle identity mysql registration bind failed: " + registrationTable, e); + } + } + + private static void setInteger(PreparedStatement statement, int index, Integer value) throws SQLException { + if (value == null) { + statement.setObject(index, null); + } else { + statement.setInt(index, value); + } + } + private static String sanitizeTable(String table) { String value = table == null || table.isBlank() ? "vehicle_identity_binding" : table.trim(); if (!value.matches("[A-Za-z0-9_]+")) { diff --git a/modules/core/vehicle-identity/src/main/java/com/lingniu/ingest/identity/VehicleIdentityRegistry.java b/modules/core/vehicle-identity/src/main/java/com/lingniu/ingest/identity/VehicleIdentityRegistry.java index 08fb3bf8..d15808c5 100644 --- a/modules/core/vehicle-identity/src/main/java/com/lingniu/ingest/identity/VehicleIdentityRegistry.java +++ b/modules/core/vehicle-identity/src/main/java/com/lingniu/ingest/identity/VehicleIdentityRegistry.java @@ -3,4 +3,7 @@ package com.lingniu.ingest.identity; public interface VehicleIdentityRegistry { void bind(VehicleIdentityBinding binding); + + default void register(VehicleRegistrationBinding registration) { + } } diff --git a/modules/core/vehicle-identity/src/main/java/com/lingniu/ingest/identity/VehicleRegistrationBinding.java b/modules/core/vehicle-identity/src/main/java/com/lingniu/ingest/identity/VehicleRegistrationBinding.java new file mode 100644 index 00000000..4761c61b --- /dev/null +++ b/modules/core/vehicle-identity/src/main/java/com/lingniu/ingest/identity/VehicleRegistrationBinding.java @@ -0,0 +1,36 @@ +package com.lingniu.ingest.identity; + +import com.lingniu.ingest.api.ProtocolId; + +public record VehicleRegistrationBinding( + ProtocolId protocol, + String vin, + String phone, + String deviceId, + String plate, + Integer province, + Integer city, + String maker, + String deviceType, + Integer plateColor +) { + public VehicleRegistrationBinding { + vin = normalize(vin); + phone = normalize(phone); + deviceId = normalize(deviceId); + plate = normalize(plate); + maker = normalize(maker); + deviceType = normalize(deviceType); + if (phone.isBlank() && deviceId.isBlank() && plate.isBlank()) { + throw new IllegalArgumentException("at least one registration identifier must not be blank"); + } + } + + public boolean hasResolvedVin() { + return !vin.isBlank() && !"unknown".equalsIgnoreCase(vin); + } + + private static String normalize(String value) { + return value == null ? "" : value.trim(); + } +} 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 c4bc45ea..a007db99 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 @@ -39,8 +39,8 @@ class MySqlVehicleIdentityServiceTest { VehicleIdentity byPlate = service.resolve(new VehicleIdentityLookup( ProtocolId.JT808, "", "", "", "粤B00001")); - assertThat(jdbc.createdTables).singleElement() - .satisfies(sql -> assertThat(sql).contains("CREATE TABLE IF NOT EXISTS vehicle_identity_binding")); + assertThat(jdbc.createdTables) + .anySatisfy(sql -> assertThat(sql).contains("CREATE TABLE IF NOT EXISTS vehicle_identity_binding")); assertThat(jdbc.bindings).containsKey("JT808|PHONE|13079960001"); assertThat(byPhone.vin()).isEqualTo("LNVIN000000000123"); assertThat(byPhone.resolved()).isTrue(); @@ -105,9 +105,50 @@ class MySqlVehicleIdentityServiceTest { assertThat(jdbc.lookupSelects).isZero(); } + @Test + void storesRegistrationWithoutVinAndResolvesAfterVinIsWrittenBack() { + RecordingIdentityJdbc jdbc = new RecordingIdentityJdbc(); + MySqlVehicleIdentityService service = new MySqlVehicleIdentityService(jdbc.dataSource(), "vehicle_identity_binding"); + + service.register(new VehicleRegistrationBinding( + ProtocolId.JT808, + "unknown", + "13079960003", + "DEV003", + "粤B00003", + 44, + 4401, + "MAKER", + "TYPE-A", + 1)); + + VehicleIdentity before = service.resolve(new VehicleIdentityLookup( + ProtocolId.JT808, "", "13079960003", "", "")); + + jdbc.writeBackRegistrationVin("JT808", "13079960003", "LNVIN000000000789"); + service.refresh(); + VehicleIdentity byPhone = service.resolve(new VehicleIdentityLookup( + ProtocolId.JT808, "", "13079960003", "", "")); + VehicleIdentity byDevice = service.resolve(new VehicleIdentityLookup( + ProtocolId.JT808, "", "", "DEV003", "")); + VehicleIdentity byPlate = service.resolve(new VehicleIdentityLookup( + ProtocolId.JT808, "", "", "", "粤B00003")); + + 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(before.resolved()).isFalse(); + assertThat(byPhone.vin()).isEqualTo("LNVIN000000000789"); + assertThat(byPhone.source()).isEqualTo(VehicleIdentitySource.BOUND_PHONE); + assertThat(byDevice.vin()).isEqualTo("LNVIN000000000789"); + assertThat(byPlate.vin()).isEqualTo("LNVIN000000000789"); + } + private static final class RecordingIdentityJdbc { private final List createdTables = new ArrayList<>(); private final Map bindings = new HashMap<>(); + private final Map registrations = new HashMap<>(); private int loadAllSelects; private int lookupSelects; @@ -115,6 +156,14 @@ class MySqlVehicleIdentityServiceTest { bindings.put(protocol + "|" + identifierType + "|" + normalize(identifierValue), 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 DataSource dataSource() { return new DataSource() { @Override public Connection getConnection() { return connection(); } @@ -162,12 +211,16 @@ class MySqlVehicleIdentityServiceTest { List params = new ArrayList<>(); InvocationHandler handler = (proxy, method, args) -> { switch (method.getName()) { - case "setString" -> { + case "setString", "setInt", "setObject" -> { set(params, (Integer) args[0], args[1]); return null; } case "executeUpdate" -> { - upsert(params); + if (sql.startsWith("INSERT INTO vehicle_identity_binding_registration ")) { + upsertRegistration(params); + } else { + upsert(params); + } return 1; } case "executeQuery" -> { @@ -175,6 +228,9 @@ class MySqlVehicleIdentityServiceTest { loadAllSelects++; return allBindingsResultSet(); } + if (sql.startsWith("SELECT protocol, phone, device_id, plate, vin FROM ")) { + return registrationResultSet(); + } lookupSelects++; return resultSet(resolve(params)); } @@ -198,6 +254,21 @@ class MySqlVehicleIdentityServiceTest { bindings.put(protocol + "|" + identifierType + "|" + identifierValue, vin); } + private void upsertRegistration(List params) { + String protocol = (String) params.get(0); + String phone = normalize((String) params.get(1)); + String vin = (String) params.get(2); + String key = protocol + "|" + phone; + RegistrationRow existing = registrations.get(key); + String effectiveVin = existing != null && "unknown".equalsIgnoreCase(vin) ? existing.vin : vin; + registrations.put(key, new RegistrationRow( + protocol, + phone, + effectiveVin, + normalize((String) params.get(3)), + normalize((String) params.get(4)))); + } + private String resolve(List params) { String protocol = (String) params.get(0); String identifierType = (String) params.get(1); @@ -239,6 +310,41 @@ class MySqlVehicleIdentityServiceTest { }; } + private ResultSet registrationResultSet() { + List rows = registrations.values().stream() + .filter(row -> row.vin != null && !row.vin.isBlank() && !"unknown".equalsIgnoreCase(row.vin)) + .toList(); + 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 < rows.size(); + } + case "getString" -> registrationColumn(rows.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 registrationColumn(RegistrationRow row, Object column) { + return switch (column.toString()) { + case "protocol" -> row.protocol; + case "phone" -> row.phone; + case "device_id" -> row.deviceId; + case "plate" -> row.plate; + case "vin" -> row.vin; + default -> throw new IllegalArgumentException("unsupported column " + column); + }; + } + private static ResultSet resultSet(String vin) { InvocationHandler handler = new InvocationHandler() { private boolean read; @@ -283,5 +389,21 @@ class MySqlVehicleIdentityServiceTest { if (type == Byte.TYPE) return (byte) 0; return null; } + + private static final class RegistrationRow { + private final String protocol; + private final String phone; + private String vin; + private final String deviceId; + private final String plate; + + private RegistrationRow(String protocol, String phone, String vin, String deviceId, String plate) { + this.protocol = protocol; + this.phone = phone; + this.vin = vin; + this.deviceId = deviceId; + this.plate = plate; + } + } } } diff --git a/modules/protocols/protocol-jt808/src/main/java/com/lingniu/ingest/protocol/jt808/inbound/Jt808ChannelHandler.java b/modules/protocols/protocol-jt808/src/main/java/com/lingniu/ingest/protocol/jt808/inbound/Jt808ChannelHandler.java index da88dc8d..a94aa5d1 100644 --- a/modules/protocols/protocol-jt808/src/main/java/com/lingniu/ingest/protocol/jt808/inbound/Jt808ChannelHandler.java +++ b/modules/protocols/protocol-jt808/src/main/java/com/lingniu/ingest/protocol/jt808/inbound/Jt808ChannelHandler.java @@ -9,6 +9,7 @@ import com.lingniu.ingest.identity.VehicleIdentityLookup; import com.lingniu.ingest.identity.VehicleIdentityRegistry; import com.lingniu.ingest.identity.VehicleIdentityResolver; import com.lingniu.ingest.identity.VehicleIdentitySource; +import com.lingniu.ingest.identity.VehicleRegistrationBinding; import com.lingniu.ingest.protocol.jt808.codec.Jt808FrameEncoder; import com.lingniu.ingest.protocol.jt808.codec.Jt808MalformedFrame; import com.lingniu.ingest.protocol.jt808.codec.Jt808MessageDecoder; @@ -286,7 +287,18 @@ public class Jt808ChannelHandler extends SimpleChannelInboundHandler { Instant.now(), Instant.now(), Map.of("protocolVersion", msg.header().version().name()))); session = session.withVin(identity.vin()); - if (msg.body() instanceof Jt808Body.Register reg && reg.plate() != null) { + if (msg.body() instanceof Jt808Body.Register reg) { + identityRegistry.register(new VehicleRegistrationBinding( + ProtocolId.JT808, + session.vin(), + phone, + reg.deviceId(), + reg.plate(), + reg.province(), + reg.city(), + reg.maker(), + reg.deviceType(), + reg.plateColor())); if (identity.resolved() && session.vin() != null && !session.vin().isBlank() && !"unknown".equalsIgnoreCase(session.vin())) {