perf(go): cache identity binding lookups
This commit is contained in:
@@ -71,6 +71,7 @@
|
||||
- `jt808_registration`:808 注册、鉴权、最新活跃和 VIN 匹配状态。
|
||||
- 状态:`vehicle_identity_binding` 使用 VIN 主键;`jt808_registration` 使用 phone 主键;两张表都不保留代理自增主键和 `created_at`。
|
||||
- 查询:`vehicle_identity_binding` 的 phone/device_id/plate 都是唯一键,接入侧按唯一键直查 VIN,不做无意义排序。
|
||||
- 写入链路:gateway 对 binding 查 VIN 使用短 TTL 内存缓存,包含未命中缓存,避免无 VIN 高频位置帧每条都打 MySQL。
|
||||
- 生产:RDS 已在上线前删除旧 `vehicle_identity_binding_registration`、`vehicle_identity_bindings`,gateway 启动会自动创建最小 schema。
|
||||
|
||||
## 字段提升规则
|
||||
|
||||
@@ -146,6 +146,7 @@ func buildIdentityResolver(ctx context.Context, logger *slog.Logger) (identity.R
|
||||
table := env("VEHICLE_IDENTITY_TABLE", "vehicle_identity_binding")
|
||||
resolver := identity.NewMySQLResolverWithOptions(db, table, identity.MySQLResolverOptions{
|
||||
LocationTouchInterval: time.Duration(envInt("JT808_REGISTRATION_LOCATION_TOUCH_INTERVAL_SECONDS", 600)) * time.Second,
|
||||
LookupCacheTTL: time.Duration(envInt("IDENTITY_LOOKUP_CACHE_TTL_SECONDS", 600)) * time.Second,
|
||||
})
|
||||
if envBool("IDENTITY_MYSQL_ENSURE_SCHEMA", true) {
|
||||
if err := resolver.EnsureSchema(ctx); err != nil {
|
||||
@@ -153,7 +154,7 @@ func buildIdentityResolver(ctx context.Context, logger *slog.Logger) (identity.R
|
||||
return nil, nil, err
|
||||
}
|
||||
}
|
||||
logger.Info("identity mysql resolver enabled", "table", table)
|
||||
logger.Info("identity mysql resolver enabled", "table", table, "lookup_cache_ttl_seconds", envInt("IDENTITY_LOOKUP_CACHE_TTL_SECONDS", 600))
|
||||
return resolver, func() { _ = db.Close() }, nil
|
||||
}
|
||||
|
||||
|
||||
@@ -1,6 +1,8 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"os"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope"
|
||||
@@ -32,6 +34,18 @@ func TestNATSSinkConfigFromEnvUsesExplicitSubjects(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestGatewayConfiguresIdentityLookupCacheTTL(t *testing.T) {
|
||||
source, err := os.ReadFile("main.go")
|
||||
if err != nil {
|
||||
t.Fatalf("read main.go: %v", err)
|
||||
}
|
||||
for _, want := range []string{"LookupCacheTTL", "IDENTITY_LOOKUP_CACHE_TTL_SECONDS"} {
|
||||
if !strings.Contains(string(source), want) {
|
||||
t.Fatalf("gateway should expose identity lookup cache ttl, missing %s", want)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestNATSSinkConfigFromEnvDefaultsToGoSubjects(t *testing.T) {
|
||||
t.Setenv("NATS_URL", "nats://172.17.111.56:4222")
|
||||
|
||||
|
||||
@@ -26,12 +26,22 @@ type MySQLResolver struct {
|
||||
db *sql.DB
|
||||
table string
|
||||
locationTouchInterval time.Duration
|
||||
lookupCacheTTL time.Duration
|
||||
lookupMu sync.Mutex
|
||||
lookupCache map[string]lookupCacheEntry
|
||||
touchMu sync.Mutex
|
||||
locationTouches map[string]time.Time
|
||||
}
|
||||
|
||||
type MySQLResolverOptions struct {
|
||||
LocationTouchInterval time.Duration
|
||||
LookupCacheTTL time.Duration
|
||||
}
|
||||
|
||||
type lookupCacheEntry struct {
|
||||
vin string
|
||||
notFound bool
|
||||
expiresAt time.Time
|
||||
}
|
||||
|
||||
func NewMySQLResolver(db *sql.DB, table string) *MySQLResolver {
|
||||
@@ -50,10 +60,16 @@ func NewMySQLResolverWithOptions(db *sql.DB, table string, opts MySQLResolverOpt
|
||||
if interval <= 0 {
|
||||
interval = 10 * time.Minute
|
||||
}
|
||||
lookupTTL := opts.LookupCacheTTL
|
||||
if lookupTTL <= 0 {
|
||||
lookupTTL = 10 * time.Minute
|
||||
}
|
||||
return &MySQLResolver{
|
||||
db: db,
|
||||
table: table,
|
||||
locationTouchInterval: interval,
|
||||
lookupCacheTTL: lookupTTL,
|
||||
lookupCache: map[string]lookupCacheEntry{},
|
||||
locationTouches: map[string]time.Time{},
|
||||
}
|
||||
}
|
||||
@@ -161,9 +177,32 @@ func (r *MySQLResolver) Resolve(ctx context.Context, env envelope.FrameEnvelope)
|
||||
}
|
||||
|
||||
func (r *MySQLResolver) lookup(ctx context.Context, column string, value string) (string, error) {
|
||||
cacheKey := column + "\x00" + value
|
||||
now := time.Now()
|
||||
r.lookupMu.Lock()
|
||||
cached, ok := r.lookupCache[cacheKey]
|
||||
if ok && now.Before(cached.expiresAt) {
|
||||
r.lookupMu.Unlock()
|
||||
if cached.notFound {
|
||||
return "", sql.ErrNoRows
|
||||
}
|
||||
return cached.vin, nil
|
||||
}
|
||||
r.lookupMu.Unlock()
|
||||
|
||||
query := "SELECT vin FROM " + r.table + " WHERE " + column + " = ? AND vin IS NOT NULL AND vin <> ''"
|
||||
var vin string
|
||||
err := r.db.QueryRowContext(ctx, query, value).Scan(&vin)
|
||||
if err != nil && !errors.Is(err, sql.ErrNoRows) {
|
||||
return "", err
|
||||
}
|
||||
r.lookupMu.Lock()
|
||||
r.lookupCache[cacheKey] = lookupCacheEntry{
|
||||
vin: strings.TrimSpace(vin),
|
||||
notFound: errors.Is(err, sql.ErrNoRows),
|
||||
expiresAt: now.Add(r.lookupCacheTTL),
|
||||
}
|
||||
r.lookupMu.Unlock()
|
||||
return vin, err
|
||||
}
|
||||
|
||||
|
||||
@@ -5,6 +5,7 @@ import (
|
||||
"database/sql"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/DATA-DOG/go-sqlmock"
|
||||
|
||||
@@ -188,9 +189,6 @@ func TestMySQLResolverTracksFirstJT808LocationThenThrottles(t *testing.T) {
|
||||
WillReturnError(sql.ErrNoRows)
|
||||
mock.ExpectExec("INSERT INTO jt808_registration").
|
||||
WillReturnResult(sqlmock.NewResult(1, 1))
|
||||
mock.ExpectQuery("SELECT vin FROM vehicle_identity_binding WHERE phone = \\?").
|
||||
WithArgs("13307795425").
|
||||
WillReturnError(sql.ErrNoRows)
|
||||
|
||||
resolver := NewMySQLResolver(db, "vehicle_identity_binding")
|
||||
for i := 0; i < 2; i++ {
|
||||
@@ -210,6 +208,32 @@ func TestMySQLResolverTracksFirstJT808LocationThenThrottles(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestMySQLResolverCachesIdentityMissesForHighFrequencyFrames(t *testing.T) {
|
||||
db, mock := newMockDB(t)
|
||||
defer db.Close()
|
||||
mock.ExpectQuery("SELECT vin FROM vehicle_identity_binding WHERE phone = \\?").
|
||||
WithArgs("13307795425").
|
||||
WillReturnError(sql.ErrNoRows)
|
||||
|
||||
resolver := NewMySQLResolverWithOptions(db, "vehicle_identity_binding", MySQLResolverOptions{
|
||||
LocationTouchInterval: time.Hour,
|
||||
LookupCacheTTL: time.Hour,
|
||||
})
|
||||
for i := 0; i < 2; i++ {
|
||||
_, err := resolver.Resolve(context.Background(), envelope.FrameEnvelope{
|
||||
Protocol: envelope.ProtocolJT808,
|
||||
Phone: "013307795425",
|
||||
Parsed: map[string]any{},
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("Resolve() error = %v", err)
|
||||
}
|
||||
}
|
||||
if err := mock.ExpectationsWereMet(); err != nil {
|
||||
t.Fatalf("sql expectations: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestMySQLResolverEnsuresMinimalIdentitySchema(t *testing.T) {
|
||||
db, mock := newMockDB(t)
|
||||
defer db.Close()
|
||||
|
||||
Reference in New Issue
Block a user