refactor(go): avoid duplicate realtime protocol payloads
This commit is contained in:
@@ -52,7 +52,7 @@ flowchart LR
|
|||||||
| MySQL | `vehicle_daily_mileage` | Queryable daily mileage | vin, date, protocol, daily mileage, source mileage | temporary vehicle keys, generic metric key/value rows, per-frame raw details |
|
| MySQL | `vehicle_daily_mileage` | Queryable daily mileage | vin, date, protocol, daily mileage, source mileage | temporary vehicle keys, generic metric key/value rows, per-frame raw details |
|
||||||
| MySQL | `vehicle_identity_binding` | Manual identity mapping | vin, plate, phone, device id | registration history |
|
| MySQL | `vehicle_identity_binding` | Manual identity mapping | vin, plate, phone, device id | registration history |
|
||||||
| MySQL | `jt808_registration` | JT808 registration and auth trace | phone, device id, plate, auth code, vin match state, first/latest seen | GB32960 or MQTT records |
|
| MySQL | `jt808_registration` | JT808 registration and auth trace | phone, device id, plate, auth code, vin match state, first/latest seen | GB32960 or MQTT records |
|
||||||
| Redis | `vehicle:latest:{vehicleKey}` | Latest merged realtime state | cross-protocol latest fields | historical data |
|
| Redis | `vehicle:latest:{vehicleKey}` | Latest merged realtime state | cross-protocol latest fields only | full parsed payloads and historical data |
|
||||||
| Redis | `vehicle:realtime-raw:{protocol}:{vehicleKey}` | Latest full realtime protocol state | latest protocol parsed payload | historical data |
|
| Redis | `vehicle:realtime-raw:{protocol}:{vehicleKey}` | Latest full realtime protocol state | latest protocol parsed payload | historical data |
|
||||||
|
|
||||||
## Event Envelope Rules
|
## Event Envelope Rules
|
||||||
|
|||||||
@@ -81,9 +81,9 @@ Redis 使用 DB 50,定位为实时缓存,不作为历史事实来源。
|
|||||||
|
|
||||||
| Key 族 | 用途 |
|
| Key 族 | 用途 |
|
||||||
| --- | --- |
|
| --- | --- |
|
||||||
| `vehicle:latest:{vehicleKey}` | 跨协议合并后的最新实时快照 |
|
| `vehicle:latest:{vehicleKey}` | 跨协议合并后的最新核心字段快照,不重复保存完整 parsed |
|
||||||
| `vehicle:latest:{vehicleKey}:{protocol}` | 单协议最新快照 |
|
| `vehicle:latest:{vehicleKey}:{protocol}` | 单协议最新快照 |
|
||||||
| `vehicle:realtime-raw:{protocol}:{vehicleKey}` | 单协议最新完整 parsed 状态 |
|
| `vehicle:realtime-raw:{protocol}:{vehicleKey}` | 单协议最新完整 parsed 状态,是实时完整协议字段的唯一 Redis 副本 |
|
||||||
| `vehicle:online:{vehicleKey}` | 在线状态和 TTL |
|
| `vehicle:online:{vehicleKey}` | 在线状态和 TTL |
|
||||||
| `vehicle:protocols:{vehicleKey}` | 当前车辆最近出现过的协议集合 |
|
| `vehicle:protocols:{vehicleKey}` | 当前车辆最近出现过的协议集合 |
|
||||||
| `vehicle:last_seen` | 最近活跃车辆排序集合 |
|
| `vehicle:last_seen` | 最近活跃车辆排序集合 |
|
||||||
|
|||||||
@@ -638,7 +638,7 @@
|
|||||||
<thead><tr><th>接口类型</th><th>数据源</th><th>说明</th></tr></thead>
|
<thead><tr><th>接口类型</th><th>数据源</th><th>说明</th></tr></thead>
|
||||||
<tbody>
|
<tbody>
|
||||||
<tr><td>VIN 是否在线</td><td>Redis online key</td><td>TTL 内有数据即在线,默认 TTL 600 秒。</td></tr>
|
<tr><td>VIN 是否在线</td><td>Redis online key</td><td>TTL 内有数据即在线,默认 TTL 600 秒。</td></tr>
|
||||||
<tr><td>VIN 实时数据</td><td>Redis merged snapshot</td><td>跨协议合并 fields,并保留各协议 ProtocolData。</td></tr>
|
<tr><td>VIN 实时数据</td><td>Redis latest snapshot</td><td>跨协议合并核心 fields;完整协议字段通过 realtime-raw 查询。</td></tr>
|
||||||
<tr><td>单协议实时 RAW</td><td>Redis realtime-raw</td><td>查看某 VIN/phone 在某协议下最新 parsed 全量字段。</td></tr>
|
<tr><td>单协议实时 RAW</td><td>Redis realtime-raw</td><td>查看某 VIN/phone 在某协议下最新 parsed 全量字段。</td></tr>
|
||||||
</tbody>
|
</tbody>
|
||||||
</table>
|
</table>
|
||||||
@@ -722,7 +722,7 @@
|
|||||||
</div>
|
</div>
|
||||||
<div class="ok">
|
<div class="ok">
|
||||||
<strong>实时多协议合并</strong><br>
|
<strong>实时多协议合并</strong><br>
|
||||||
Redis 同时保留 latest snapshot、protocol snapshot、realtime-raw,既能看统一实时,也能追单协议原始字段。
|
Redis latest snapshot 只保留统一实时核心字段;完整协议 parsed 只放 realtime-raw,避免重复缓存大 JSON。
|
||||||
</div>
|
</div>
|
||||||
<div class="ok">
|
<div class="ok">
|
||||||
<strong>身份解析降级</strong><br>
|
<strong>身份解析降级</strong><br>
|
||||||
|
|||||||
@@ -18,7 +18,6 @@ type Snapshot struct {
|
|||||||
Fields map[string]any `json:"fields,omitempty"`
|
Fields map[string]any `json:"fields,omitempty"`
|
||||||
FieldTimesMS map[string]int64 `json:"field_times_ms,omitempty"`
|
FieldTimesMS map[string]int64 `json:"field_times_ms,omitempty"`
|
||||||
Parsed map[string]any `json:"parsed,omitempty"`
|
Parsed map[string]any `json:"parsed,omitempty"`
|
||||||
ProtocolData map[envelope.Protocol]map[string]any `json:"protocol_data,omitempty"`
|
|
||||||
UpdatedAtMS int64 `json:"updated_at_ms"`
|
UpdatedAtMS int64 `json:"updated_at_ms"`
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -92,7 +92,6 @@ func (r *Repository) Update(ctx context.Context, env envelope.FrameEnvelope) err
|
|||||||
VIN: vin,
|
VIN: vin,
|
||||||
Fields: map[string]any{},
|
Fields: map[string]any{},
|
||||||
FieldTimesMS: map[string]int64{},
|
FieldTimesMS: map[string]int64{},
|
||||||
ProtocolData: map[envelope.Protocol]map[string]any{},
|
|
||||||
}
|
}
|
||||||
} else if merged.VIN == "" && vin != "" {
|
} else if merged.VIN == "" && vin != "" {
|
||||||
merged.VIN = vin
|
merged.VIN = vin
|
||||||
@@ -105,10 +104,6 @@ func (r *Repository) Update(ctx context.Context, env envelope.FrameEnvelope) err
|
|||||||
merged.SourceEndpoint = env.SourceEndpoint
|
merged.SourceEndpoint = env.SourceEndpoint
|
||||||
}
|
}
|
||||||
merged.Protocols = protocols
|
merged.Protocols = protocols
|
||||||
if merged.ProtocolData == nil {
|
|
||||||
merged.ProtocolData = map[envelope.Protocol]map[string]any{}
|
|
||||||
}
|
|
||||||
merged.ProtocolData[env.Protocol] = cloneMap(protocolSnapshot.Parsed)
|
|
||||||
merged.UpdatedAtMS = nowMS
|
merged.UpdatedAtMS = nowMS
|
||||||
if err := r.setJSON(ctx, mergedKey(vehicleKey), merged, r.cfg.ttl()); err != nil {
|
if err := r.setJSON(ctx, mergedKey(vehicleKey), merged, r.cfg.ttl()); err != nil {
|
||||||
return err
|
return err
|
||||||
|
|||||||
@@ -2,6 +2,7 @@ package realtime
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
|
"encoding/json"
|
||||||
"net/http"
|
"net/http"
|
||||||
"net/http/httptest"
|
"net/http/httptest"
|
||||||
"strings"
|
"strings"
|
||||||
@@ -120,12 +121,12 @@ func TestRepositoryStoresFullParsedProtocolSnapshotAndMergesGB32960Units(t *test
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatalf("GetMerged() error = %v", err)
|
t.Fatalf("GetMerged() error = %v", err)
|
||||||
}
|
}
|
||||||
gb, ok := merged.ProtocolData[envelope.ProtocolGB32960]
|
mergedJSON, err := json.Marshal(merged)
|
||||||
if !ok {
|
if err != nil {
|
||||||
t.Fatalf("merged protocol data missing: %#v", merged.ProtocolData)
|
t.Fatal(err)
|
||||||
}
|
}
|
||||||
if len(gb["data_units"].([]any)) != 2 {
|
if strings.Contains(string(mergedJSON), "protocol_data") {
|
||||||
t.Fatalf("merged protocol data did not keep full parsed fields: %#v", gb)
|
t.Fatalf("merged snapshot should not duplicate full protocol parsed data: %s", string(mergedJSON))
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user