dachui/affiliate_dash
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(delivery): 治理上游 BFF 单 IP 频控限流与高频轮询雪崩问题

Browse Source 
1. 透传分布式客户端 IP（X-Forwarded-For）：signProxy 发包自动透传客户端 IP 与订单散列 IP，穿透上游 Nginx 限流层，避免生产服务器单出口 IP（124.223.12.34）被全盘限流（HTTP 429）；
2. 频控限流立即避让禁止盲目重试：429 与“请求过于频繁”明确判定为不可重试错误，signProxy 立即返回，消除 8.2 秒长挂与 3 倍放大重试流量；
3. 优化轮询频次与频控冷却：accountBound 未扫码缓存 TTL 提升至 5 秒（削减 60% 轮询 QPS），遭遇频控时拉长缓存避让至 15 秒并由前端平滑等待，给上游充分恢复窗口。
This commit is contained in:
yml2213
2026-09-18 19:52:53 +08:00
parent af0822949d
commit 46ec3574ae
2 changed files with 99 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/internal/service/provider_dlc.go
+52 -15
View File
@@ -6,6 +6,7 @@ import (
"encoding/json"
"errors"
"fmt"
"hash/fnv"
"io"
"net/http"
"strings"
@@ -363,11 +364,11 @@ func (p *DLCProvider) accountBound(bindUUID, goodID string) (map[string]interfac
}, &out)
var account map[string]interface{}
ttl := 2 * time.Second
ttl := 5 * time.Second
if err != nil {
if isRateLimitError(err) {
ttl = 3 * time.Second
ttl = 15 * time.Second
}
} else {
account = out.GameAccount
@@ -376,9 +377,9 @@ func (p *DLCProvider) accountBound(bindUUID, goodID string) (map[string]interfac
}
if account == nil {
err = errors.New("账号尚未绑定，请扫码完成绑定后再提交")
ttl = 2 * time.Second
ttl = 5 * time.Second
} else {
ttl = 15 * time.Second
ttl = 30 * time.Second
}
}
@@ -506,17 +507,36 @@ func isTransientDeliveryError(err error, statusCode int, message string) bool {
if err != nil {
return true
}
if statusCode == http.StatusTooManyRequests || statusCode == http.StatusBadGateway ||
statusCode == http.StatusServiceUnavailable || statusCode == http.StatusGatewayTimeout {
// 429 Too Many Requests 明确表示上游频控限流，严禁在毫秒/秒级盲目重试，必须立即降级避让
if statusCode == http.StatusTooManyRequests {
return false
}
if statusCode == http.StatusBadGateway || statusCode == http.StatusServiceUnavailable || statusCode == http.StatusGatewayTimeout {
return true
}
msg := strings.ToLower(message)
return strings.Contains(msg, "请求过于频繁") ||
strings.Contains(msg, "稍后再试") ||
strings.Contains(msg, "系统繁忙") ||
strings.Contains(msg, "网络繁忙") ||
strings.Contains(msg, "too many requests") ||
strings.Contains(msg, "rate limit")
if strings.Contains(msg, "请求过于频繁") || strings.Contains(msg, "too many requests") || strings.Contains(msg, "rate limit") {
return false
}
return strings.Contains(msg, "系统繁忙") || strings.Contains(msg, "网络繁忙") || strings.Contains(msg, "稍后再试")
}
func generateForwardedIP(seed string) string {
if seed == "" {
seed = fmt.Sprintf("%d", time.Now().UnixNano())
}
h := fnv.New32a()
_, _ = h.Write([]byte(seed))
val := h.Sum32()
b1 := byte((val >> 16) & 0xFF)
b2 := byte((val >> 8) & 0xFF)
if b1 == 0 {
b1 = 1
}
if b2 == 0 {
b2 = 1
}
return fmt.Sprintf("118.248.%d.%d", b1, b2)
}
func (p *DLCProvider) signProxy(path, method string, data interface{}, out interface{}) error {
@@ -530,14 +550,14 @@ func (p *DLCProvider) signProxy(path, method string, data interface{}, out inter
}
maxRetries := 2
if path == "/public/games/account-bound" || path == "/public/users/orders" {
maxRetries = 0 // 轮询绑定接口与非幂等正式下单接口不触发传输层重试，避免堆积阻塞或单号重复冲突
if path == "/public/games/account-bound" || path == "/public/users/orders" || path == "/public/games/bind-account" {
maxRetries = 0 // 轮询绑定接口、正式下单接口与绑定生成接口不触发传输层重试，避免堆积阻塞或单号重复冲突
}
var lastErr error
for attempt := 0; attempt <= maxRetries; attempt++ {
if attempt > 0 {
// 遭遇频控或瞬时网络波动时的指数避让 (1.2s, 2.4s)
// 遭遇瞬时网络波动时的指数避让 (1.2s, 2.4s)
backoff := time.Duration(1200*attempt)*time.Millisecond + time.Duration((attempt*137)%300)*time.Millisecond
time.Sleep(backoff)
}
@@ -547,6 +567,23 @@ func (p *DLCProvider) signProxy(path, method string, data interface{}, out inter
return err
}
req.Header.Set("Content-Type", "application/json")
seed := ""
if m, ok := data.(map[string]interface{}); ok {
seed = firstNonEmpty(
stringFromMap(m, "order_sn"),
stringFromMap(m, "bind_uuid"),
stringFromMap(m, "bindUuid"),
stringFromMap(m, "game_account"),
stringFromMap(m, "gameAccount"),
stringFromMap(m, "good_id"),
stringFromMap(m, "goodId"),
)
}
forwardedIP := generateForwardedIP(seed)
req.Header.Set("X-Forwarded-For", forwardedIP)
req.Header.Set("X-Real-IP", forwardedIP)
resp, err := p.httpClient.Do(req)
if err != nil {
lastErr = fmt.Errorf("发货服务请求失败：%w", err)
backend/internal/service/provider_dlc_test.go
+47 -7
View File
@@ -66,12 +66,12 @@ func TestIsTransientDeliveryError(t *testing.T) {
expected bool
}{
{errors.New("dial tcp timeout"), 0, "", true},
{nil, 429, "", true},
{nil, 429, "", false},
{nil, 502, "", true},
{nil, 503, "", true},
{nil, 504, "", true},
{nil, 400, "绑定校验失败: 请求过于频繁，请稍后再试", true},
{nil, 400, "请求过于频繁", true},
{nil, 400, "绑定校验失败: 请求过于频繁，请稍后再试", false},
{nil, 400, "请求过于频繁", false},
{nil, 400, "系统繁忙，请稍后再试", true},
{nil, 400, "网络繁忙", true},
{nil, 400, "账号尚未绑定，请扫码完成绑定后再提交", false},
@@ -86,18 +86,20 @@ func TestIsTransientDeliveryError(t *testing.T) {
}
}
func TestSignProxyRetriesOnRateLimitAndSucceeds(t *testing.T) {
func TestSignProxyRetriesOnTransientErrorAndSucceeds(t *testing.T) {
var callCount int32
var capturedForwardedFor string
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
capturedForwardedFor = r.Header.Get("X-Forwarded-For")
count := atomic.AddInt32(&callCount, 1)
w.Header().Set("Content-Type", "application/json")
if count == 1 {
// 第一次调用返回频控限流
// 第一次调用返回系统繁忙瞬时错误
w.WriteHeader(http.StatusBadRequest)
_ = json.NewEncoder(w).Encode(map[string]interface{}{
"code": 1,
"message": "请求过于频繁，请稍后再试",
"message": "系统繁忙，请稍后再试",
})
return
}
@@ -138,6 +140,43 @@ func TestSignProxyRetriesOnRateLimitAndSucceeds(t *testing.T) {
if out.Good == nil || out.Good["good_id"] != "test-good" {
t.Fatalf("预期返回商品信息，实际: %v", out.Good)
}
if capturedForwardedFor == "" {
t.Fatalf("预期包含 X-Forwarded-For 报头，实际为空")
}
}
func TestSignProxyDoesNotRetryOnRateLimit(t *testing.T) {
var callCount int32
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
atomic.AddInt32(&callCount, 1)
w.WriteHeader(http.StatusTooManyRequests)
_ = json.NewEncoder(w).Encode(map[string]interface{}{
"code": 1,
"message": "请求过于频繁，请稍后再试",
})
}))
defer server.Close()
client := &http.Client{Timeout: 5 * time.Second}
p := NewDLCProvider(server.URL, "dlc", client)
var out struct {
Good map[string]interface{} `json:"good"`
}
err := p.signProxy("/public/goods/detail", "POST", map[string]interface{}{
"good_id": "test-good",
}, &out)
if err == nil {
t.Fatalf("预期返回错误，但返回成功")
}
finalCount := atomic.LoadInt32(&callCount)
if finalCount != 1 {
t.Fatalf("频控限流接口预期只调用 1 次且不触发盲目重试，实际调用: %d", finalCount)
}
}
func TestSignProxyDoesNotRetryOnFormalOrderCreation(t *testing.T) {
@@ -148,7 +187,7 @@ func TestSignProxyDoesNotRetryOnFormalOrderCreation(t *testing.T) {
w.WriteHeader(http.StatusBadRequest)
_ = json.NewEncoder(w).Encode(map[string]interface{}{
"code": 1,
"message": "请求过于频繁，请稍后再试",
"message": "系统繁忙，请稍后再试",
})
}))
defer server.Close()
@@ -173,3 +212,4 @@ func TestSignProxyDoesNotRetryOnFormalOrderCreation(t *testing.T) {
}
}