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

simplify vehicle identity binding to plate vin
Some checks failed
ci/woodpecker/push/woodpecker Pipeline was canceled
Details

Browse Source
This commit is contained in:
lingniu
2026-06-29 23:07:25 +08:00
parent 6a17134942
commit 1682b3f83f
5 changed files with 135 additions and 83 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
deploy/local/launchctl/README.md
View File

@@ -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` 仅适合一次性联调，不建议用于生产验证。
## 启停

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

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

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

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

17
modules/core/vehicle-identity/src/test/java/com/lingniu/ingest/identity/MySqlVehicleIdentityServiceJdbcTest.java
View File

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

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

@@ -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<Object> 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<Object> 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<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" -> 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<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 "plate" -> entry.getKey();
case "vin" -> entry.getValue();
default -> throw new IllegalArgumentException("unsupported column " + name);
};
}
private static String legacyColumn(Map.Entry<String, String> 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<RegistrationRow> 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<RegistrationRow> rows = registrations.values().stream()
.filter(row -> row.vin != null && !row.vin.isBlank() && !"unknown".equalsIgnoreCase(row.vin))
.toList();
return registrationRowsResultSet(rows);
}
private ResultSet registrationRowsResultSet(List<RegistrationRow> rows) {
InvocationHandler handler = new InvocationHandler() {
private int index = -1;