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

perf(api): bound reverse geocode requests

Browse Source
This commit is contained in:
lingniu
2026-07-16 02:10:10 +08:00
parent 1e80278fee
commit 85e92e62da
3 changed files with 232 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

120
vehicle-data-platform/apps/api/internal/app/server.go
View File

@@ -2,6 +2,7 @@ package app
import (
"bytes"
"container/list"
"context"
"database/sql"
"encoding/json"
@@ -238,6 +239,112 @@ type mapReverseGeocodeResponse struct {
Adcode string `json:"adcode,omitempty"`
}
const (
reverseGeocodeCacheTTL = time.Hour
reverseGeocodeCacheMaxEntries = 4096
)
type reverseGeocodeCacheEntry struct {
key string
value mapReverseGeocodeResponse
expiresAt time.Time
}
type reverseGeocodeKeyLock struct {
mu sync.Mutex
refs int
}
type reverseGeocodeCache struct {
mu sync.Mutex
entries map[string]*list.Element
recency *list.List
keyLocks map[string]*reverseGeocodeKeyLock
maxEntries int
ttl time.Duration
now func() time.Time
}
func newReverseGeocodeCache(maxEntries int, ttl time.Duration) *reverseGeocodeCache {
return &reverseGeocodeCache{
entries: make(map[string]*list.Element),
recency: list.New(),
keyLocks: make(map[string]*reverseGeocodeKeyLock),
maxEntries: maxEntries,
ttl: ttl,
now: time.Now,
}
}
func reverseGeocodeCacheKey(longitude, latitude float64) string {
return fmt.Sprintf("%.5f,%.5f", longitude, latitude)
}
func (cache *reverseGeocodeCache) get(key string) (mapReverseGeocodeResponse, bool) {
cache.mu.Lock()
defer cache.mu.Unlock()
element, ok := cache.entries[key]
if !ok {
return mapReverseGeocodeResponse{}, false
}
entry := element.Value.(*reverseGeocodeCacheEntry)
if !entry.expiresAt.After(cache.now()) {
cache.remove(element)
return mapReverseGeocodeResponse{}, false
}
cache.recency.MoveToFront(element)
return entry.value, true
}
func (cache *reverseGeocodeCache) put(key string, value mapReverseGeocodeResponse) {
if cache.maxEntries <= 0 || cache.ttl <= 0 {
return
}
cache.mu.Lock()
defer cache.mu.Unlock()
if element, ok := cache.entries[key]; ok {
entry := element.Value.(*reverseGeocodeCacheEntry)
entry.value = value
entry.expiresAt = cache.now().Add(cache.ttl)
cache.recency.MoveToFront(element)
return
}
for len(cache.entries) >= cache.maxEntries {
cache.remove(cache.recency.Back())
}
entry := &reverseGeocodeCacheEntry{key: key, value: value, expiresAt: cache.now().Add(cache.ttl)}
cache.entries[key] = cache.recency.PushFront(entry)
}
func (cache *reverseGeocodeCache) remove(element *list.Element) {
if element == nil {
return
}
delete(cache.entries, element.Value.(*reverseGeocodeCacheEntry).key)
cache.recency.Remove(element)
}
func (cache *reverseGeocodeCache) lockKey(key string) func() {
cache.mu.Lock()
keyLock := cache.keyLocks[key]
if keyLock == nil {
keyLock = &reverseGeocodeKeyLock{}
cache.keyLocks[key] = keyLock
}
keyLock.refs++
cache.mu.Unlock()
keyLock.mu.Lock()
return func() {
keyLock.mu.Unlock()
cache.mu.Lock()
keyLock.refs--
if keyLock.refs == 0 {
delete(cache.keyLocks, key)
}
cache.mu.Unlock()
}
}
func withAMapReverseGeocodeAPI(next http.Handler, cfg config.Config, upstream string, client *http.Client) http.Handler {
apiKey := strings.TrimSpace(cfg.AMapAPIKey)
if client == nil {
@@ -247,6 +354,7 @@ func withAMapReverseGeocodeAPI(next http.Handler, cfg config.Config, upstream st
if upstream == "" {
upstream = "https://restapi.amap.com"
}
cache := newReverseGeocodeCache(reverseGeocodeCacheMaxEntries, reverseGeocodeCacheTTL)
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.URL.Path != "/api/map/reverse-geocode" {
next.ServeHTTP(w, r)
@@ -265,6 +373,17 @@ func withAMapReverseGeocodeAPI(next http.Handler, cfg config.Config, upstream st
httpx.WriteError(w, http.StatusBadRequest, "BAD_COORDINATE", "经纬度参数无效", err.Error(), requestTraceID(r))
return
}
cacheKey := reverseGeocodeCacheKey(longitude, latitude)
unlock := cache.lockKey(cacheKey)
defer unlock()
if cached, ok := cache.get(cacheKey); ok {
cached.Longitude = longitude
cached.Latitude = latitude
w.Header().Set("X-Reverse-Geocode-Cache", "HIT")
httpx.WriteOK(w, requestTraceID(r), cached)
return
}
w.Header().Set("X-Reverse-Geocode-Cache", "MISS")
target, err := url.Parse(upstream + "/v3/geocode/regeo")
if err != nil {
httpx.WriteError(w, http.StatusBadGateway, "AMAP_REVERSE_GEOCODE_URL_INVALID", "高德逆地理编码地址无效", err.Error(), requestTraceID(r))
@@ -331,6 +450,7 @@ func withAMapReverseGeocodeAPI(next http.Handler, cfg config.Config, upstream st
Township: amapJSONText(body.Regeocode.AddressComponent.Township),
Adcode: amapJSONText(body.Regeocode.AddressComponent.Adcode),
}
cache.put(cacheKey, result)
httpx.WriteOK(w, requestTraceID(r), result)
})
}

106
vehicle-data-platform/apps/api/internal/app/server_test.go
View File

@@ -6,6 +6,8 @@ import (
"net/http"
"net/http/httptest"
"strings"
"sync"
"sync/atomic"
"testing"
"time"
@@ -171,7 +173,9 @@ func TestAMapSecurityProxyRoutesKnownMapResources(t *testing.T) {
func TestAMapReverseGeocodeAPIUsesServerSideKey(t *testing.T) {
var gotKey string
var gotLocation string
var upstreamCalls int
upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
upstreamCalls++
gotKey = r.URL.Query().Get("key")
gotLocation = r.URL.Query().Get("location")
if r.URL.Path != "/v3/geocode/regeo" {
@@ -203,6 +207,9 @@ func TestAMapReverseGeocodeAPIUsesServerSideKey(t *testing.T) {
if rec.Code != http.StatusOK {
t.Fatalf("status = %d body=%s", rec.Code, rec.Body.String())
}
if rec.Header().Get("X-Reverse-Geocode-Cache") != "MISS" {
t.Fatalf("first response cache header = %q", rec.Header().Get("X-Reverse-Geocode-Cache"))
}
mapLongitude, mapLatitude := wgs84ToGCJ02(113.1234567, 23.7654321)
if gotKey != "server-api-key" || gotLocation != fmt.Sprintf("%.6f,%.6f", mapLongitude, mapLatitude) {
t.Fatalf("key=%q location=%q", gotKey, gotLocation)
@@ -227,6 +234,105 @@ func TestAMapReverseGeocodeAPIUsesServerSideKey(t *testing.T) {
if body.TraceID != "trace-map" || body.Data.Provider != "AMap" || body.Data.FormattedAddress != "广东省广州市天河区测试路" || body.Data.Adcode != "440106" {
t.Fatalf("unexpected reverse geocode body: %+v", body)
}
secondLongitude, secondLatitude := 113.1234571, 23.7654324
secondRecorder := httptest.NewRecorder()
secondRequest := httptest.NewRequest(http.MethodGet, fmt.Sprintf("/api/map/reverse-geocode?longitude=%.7f&latitude=%.7f", secondLongitude, secondLatitude), nil)
secondRequest.Header.Set("X-Trace-Id", "trace-map-cache")
handler.ServeHTTP(secondRecorder, secondRequest)
if secondRecorder.Code != http.StatusOK || secondRecorder.Header().Get("X-Reverse-Geocode-Cache") != "HIT" {
t.Fatalf("cached status=%d header=%q body=%s", secondRecorder.Code, secondRecorder.Header().Get("X-Reverse-Geocode-Cache"), secondRecorder.Body.String())
}
var cachedBody struct {
Data mapReverseGeocodeResponse `json:"data"`
TraceID string `json:"traceId"`
}
if err := json.Unmarshal(secondRecorder.Body.Bytes(), &cachedBody); err != nil {
t.Fatalf("cached response should be JSON: %v", err)
}
if upstreamCalls != 1 || cachedBody.TraceID != "trace-map-cache" || cachedBody.Data.Longitude != secondLongitude || cachedBody.Data.Latitude != secondLatitude {
t.Fatalf("upstreamCalls=%d cachedBody=%+v", upstreamCalls, cachedBody)
}
}
func TestReverseGeocodeCacheEvictsLeastRecentlyUsedAndExpires(t *testing.T) {
clock := time.Date(2026, 7, 16, 2, 0, 0, 0, time.UTC)
cache := newReverseGeocodeCache(2, time.Hour)
cache.now = func() time.Time { return clock }
cache.put("a", mapReverseGeocodeResponse{FormattedAddress: "A"})
cache.put("b", mapReverseGeocodeResponse{FormattedAddress: "B"})
if _, ok := cache.get("a"); !ok {
t.Fatal("recent entry a should exist")
}
cache.put("c", mapReverseGeocodeResponse{FormattedAddress: "C"})
if _, ok := cache.get("b"); ok {
t.Fatal("least recently used entry b should be evicted")
}
if _, ok := cache.get("a"); !ok {
t.Fatal("recent entry a should survive capacity eviction")
}
clock = clock.Add(time.Hour + time.Nanosecond)
if _, ok := cache.get("a"); ok {
t.Fatal("entry a should expire after the TTL")
}
}
func TestAMapReverseGeocodeAPICollapsesConcurrentCacheMisses(t *testing.T) {
var upstreamCalls atomic.Int32
upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
upstreamCalls.Add(1)
time.Sleep(20 * time.Millisecond)
_, _ = w.Write([]byte(`{"status":"1","regeocode":{"formatted_address":"并发测试地址","addressComponent":{}}}`))
}))
defer upstream.Close()
handler := withAMapReverseGeocodeAPI(http.NotFoundHandler(), config.Config{AMapAPIKey: "server-api-key"}, upstream.URL, http.DefaultClient)
const workers = 8
start := make(chan struct{})
results := make(chan int, workers)
var wait sync.WaitGroup
wait.Add(workers)
for index := 0; index < workers; index++ {
go func() {
defer wait.Done()
<-start
recorder := httptest.NewRecorder()
handler.ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, "/api/map/reverse-geocode?longitude=113.123456&latitude=23.123456", nil))
results <- recorder.Code
}()
}
close(start)
wait.Wait()
close(results)
for status := range results {
if status != http.StatusOK {
t.Fatalf("concurrent response status = %d", status)
}
}
if upstreamCalls.Load() != 1 {
t.Fatalf("upstream calls = %d, want 1", upstreamCalls.Load())
}
}
func TestAMapReverseGeocodeAPIDoesNotCacheFailures(t *testing.T) {
var upstreamCalls atomic.Int32
upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
if upstreamCalls.Add(1) == 1 {
http.Error(w, "temporary failure", http.StatusServiceUnavailable)
return
}
_, _ = w.Write([]byte(`{"status":"1","regeocode":{"formatted_address":"恢复地址","addressComponent":{}}}`))
}))
defer upstream.Close()
handler := withAMapReverseGeocodeAPI(http.NotFoundHandler(), config.Config{AMapAPIKey: "server-api-key"}, upstream.URL, http.DefaultClient)
first := httptest.NewRecorder()
handler.ServeHTTP(first, httptest.NewRequest(http.MethodGet, "/api/map/reverse-geocode?longitude=113.12&latitude=23.12", nil))
second := httptest.NewRecorder()
handler.ServeHTTP(second, httptest.NewRequest(http.MethodGet, "/api/map/reverse-geocode?longitude=113.12&latitude=23.12", nil))
if first.Code != http.StatusBadGateway || second.Code != http.StatusOK || upstreamCalls.Load() != 2 {
t.Fatalf("first=%d second=%d upstreamCalls=%d", first.Code, second.Code, upstreamCalls.Load())
}
}
func TestAMapReverseGeocodeAPIRequiresServerSideKey(t *testing.T) {