修复兑换响应JSON解析Bug并优化时间校准
- 修复Exchange/BatchExchangeLimit响应data为空字符串时JSON解析失败,导致"点击太快"等错误无法被冷却机制捕获 - 复校时间从触发前10秒改为3秒,减少偏移漂移 - 新增校准偏移中位数平滑,保留最近3次取中位数过滤跳变异常 - 提前量从固定50ms改为动态值 max(50ms, errorBound + RTT/2) - 复校时重新计算触发时间,适应校准偏移变化
This commit is contained in:
+19
-17
@@ -313,12 +313,7 @@ func (c *DouyuClient) BatchExchangeLimit(cookie, commodityID string, csrf *strin
|
||||
if csrf != nil {
|
||||
currentCSRF = *csrf
|
||||
}
|
||||
var result struct {
|
||||
types.APIResponse
|
||||
Data struct {
|
||||
Limit string `json:"limit"`
|
||||
} `json:"data"`
|
||||
}
|
||||
var result types.APIResponse
|
||||
if err := c.postMultipart(constants.BatchLimitURL, cookie, csrf, "批量限制", map[string]string{
|
||||
"manualID": constants.ManualID,
|
||||
"commodityID": commodityID,
|
||||
@@ -330,9 +325,15 @@ func (c *DouyuClient) BatchExchangeLimit(cookie, commodityID string, csrf *strin
|
||||
if result.Error != 0 {
|
||||
return 0, fmt.Errorf("批量限制接口失败: %s", result.Msg)
|
||||
}
|
||||
limit, err := strconv.Atoi(result.Data.Limit)
|
||||
var data struct {
|
||||
Limit string `json:"limit"`
|
||||
}
|
||||
if err := json.Unmarshal(result.Data, &data); err != nil {
|
||||
return 0, fmt.Errorf("批量限制数据解析失败: %w", err)
|
||||
}
|
||||
limit, err := strconv.Atoi(data.Limit)
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("批量限制值无效: %q", result.Data.Limit)
|
||||
return 0, fmt.Errorf("批量限制值无效: %q", data.Limit)
|
||||
}
|
||||
return limit, nil
|
||||
}
|
||||
@@ -365,14 +366,7 @@ func (c *DouyuClient) Exchange(cookie, commodityID string, csrf *string) (string
|
||||
if csrf != nil {
|
||||
currentCSRF = *csrf
|
||||
}
|
||||
var result struct {
|
||||
types.APIResponse
|
||||
Data struct {
|
||||
ExchangeID int `json:"exchangeId"`
|
||||
ExchangeNum string `json:"exchangeNum"`
|
||||
CommodityTyp int `json:"commodityType"`
|
||||
} `json:"data"`
|
||||
}
|
||||
var result types.APIResponse
|
||||
if err := c.postMultipart(constants.ExchangeURL, cookie, csrf, "正式兑换", map[string]string{
|
||||
"manualID": constants.ManualID,
|
||||
"commodityID": commodityID,
|
||||
@@ -388,7 +382,15 @@ func (c *DouyuClient) Exchange(cookie, commodityID string, csrf *string) (string
|
||||
}
|
||||
return "", fmt.Errorf("兑换失败: %s", result.Msg)
|
||||
}
|
||||
return fmt.Sprintf("exchangeId=%d,num=%s", result.Data.ExchangeID, result.Data.ExchangeNum), nil
|
||||
var data struct {
|
||||
ExchangeID int `json:"exchangeId"`
|
||||
ExchangeNum string `json:"exchangeNum"`
|
||||
CommodityTyp int `json:"commodityType"`
|
||||
}
|
||||
if err := json.Unmarshal(result.Data, &data); err != nil {
|
||||
return "", fmt.Errorf("兑换数据解析失败: %w", err)
|
||||
}
|
||||
return fmt.Sprintf("exchangeId=%d,num=%s", data.ExchangeID, data.ExchangeNum), nil
|
||||
}
|
||||
|
||||
// RoleName 查询绑定角色名。
|
||||
|
||||
@@ -8,6 +8,7 @@ import (
|
||||
"math/rand"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"sort"
|
||||
"strings"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
@@ -24,17 +25,18 @@ import (
|
||||
)
|
||||
|
||||
const (
|
||||
warmupBeforeTrigger = 2 * time.Second
|
||||
warmupMaxAge = 30 * time.Second
|
||||
qualifiedMaxAge = 30 * time.Second
|
||||
timeCalibrateRounds = 3
|
||||
timeCalibrateWait = 1200 * time.Millisecond
|
||||
resyncBeforeTrigger = 10 * time.Second
|
||||
seckillMaxDuration = 75 * time.Second
|
||||
behaviorCooldown = 5 * time.Second
|
||||
clickFastCooldown = 3 * time.Second
|
||||
internalAdvance = 50 * time.Millisecond
|
||||
maxCooldownCount = 5
|
||||
warmupBeforeTrigger = 2 * time.Second
|
||||
warmupMaxAge = 30 * time.Second
|
||||
qualifiedMaxAge = 30 * time.Second
|
||||
timeCalibrateRounds = 3
|
||||
timeCalibrateWait = 1200 * time.Millisecond
|
||||
resyncBeforeTrigger = 3 * time.Second
|
||||
calibrationHistorySize = 3
|
||||
seckillMaxDuration = 75 * time.Second
|
||||
behaviorCooldown = 5 * time.Second
|
||||
clickFastCooldown = 3 * time.Second
|
||||
minInternalAdvance = 50 * time.Millisecond
|
||||
maxCooldownCount = 5
|
||||
)
|
||||
|
||||
type roundOutcome struct {
|
||||
@@ -91,6 +93,9 @@ type Scheduler struct {
|
||||
|
||||
productsMu sync.RWMutex
|
||||
products map[string]types.Product
|
||||
|
||||
calibrationMu sync.Mutex
|
||||
calibrationOffsets []time.Duration // 最近N次校准偏移,用于中位数平滑
|
||||
}
|
||||
|
||||
// New 创建抢购调度器。
|
||||
@@ -111,13 +116,14 @@ func New(cli *client.DouyuClient, paths types.AppPaths, font *truetype.Font, con
|
||||
1,
|
||||
256,
|
||||
),
|
||||
concurrency: concurrency,
|
||||
attempts: attempts,
|
||||
retryStages: cloneRetryStages(retryStages),
|
||||
seckillMode: normalizeSeckillMode(seckillMode),
|
||||
results: make(map[string]string),
|
||||
attemptCache: make(map[string]*types.AttemptCache),
|
||||
products: make(map[string]types.Product),
|
||||
concurrency: concurrency,
|
||||
attempts: attempts,
|
||||
retryStages: cloneRetryStages(retryStages),
|
||||
seckillMode: normalizeSeckillMode(seckillMode),
|
||||
results: make(map[string]string),
|
||||
attemptCache: make(map[string]*types.AttemptCache),
|
||||
products: make(map[string]types.Product),
|
||||
calibrationOffsets: make([]time.Duration, 0, calibrationHistorySize),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -866,16 +872,54 @@ func uniqueWarmupAttempts(attempts []types.ExchangeAttempt) []types.ExchangeAtte
|
||||
return out
|
||||
}
|
||||
|
||||
// recordCalibrationOffset 记录校准偏移,保留最近N次用于中位数平滑。
|
||||
func (s *Scheduler) recordCalibrationOffset(offset time.Duration) {
|
||||
s.calibrationMu.Lock()
|
||||
defer s.calibrationMu.Unlock()
|
||||
s.calibrationOffsets = append(s.calibrationOffsets, offset)
|
||||
if len(s.calibrationOffsets) > calibrationHistorySize {
|
||||
s.calibrationOffsets = s.calibrationOffsets[len(s.calibrationOffsets)-calibrationHistorySize:]
|
||||
}
|
||||
}
|
||||
|
||||
// smoothedOffset 返回最近校准偏移的中位数,过滤跳变异常。
|
||||
func (s *Scheduler) smoothedOffset() time.Duration {
|
||||
s.calibrationMu.Lock()
|
||||
defer s.calibrationMu.Unlock()
|
||||
n := len(s.calibrationOffsets)
|
||||
if n == 0 {
|
||||
return 0
|
||||
}
|
||||
sorted := make([]time.Duration, n)
|
||||
copy(sorted, s.calibrationOffsets)
|
||||
sort.Slice(sorted, func(i, j int) bool { return sorted[i] < sorted[j] })
|
||||
return sorted[n/2]
|
||||
}
|
||||
|
||||
// dynamicAdvance 根据校准误差边界和 RTT 计算动态提前量,最少 50ms。
|
||||
func dynamicAdvance(calibration types.TimeCalibration) time.Duration {
|
||||
adv := calibration.ErrorBound + calibration.RTT/2
|
||||
if adv < minInternalAdvance {
|
||||
adv = minInternalAdvance
|
||||
}
|
||||
return adv
|
||||
}
|
||||
|
||||
func (s *Scheduler) calibrateDouyuTime(label string) types.TimeCalibration {
|
||||
calibration, err := s.client.BestServerTimeCalibration(timeCalibrateRounds, timeCalibrateWait)
|
||||
if err != nil {
|
||||
logger.LogEvent("时间", "warn", "%s校准降级: %v", label, err)
|
||||
}
|
||||
logger.SetDouyuOffset(calibration.Offset)
|
||||
logger.LogEvent("时间", "success", "%s校准完成: 斗鱼估算=%s 偏移=%dms 误差<=%dms RTT=%dms 采样=%d 来源=%s",
|
||||
if !calibration.Fallback {
|
||||
s.recordCalibrationOffset(calibration.Offset)
|
||||
}
|
||||
smoothed := s.smoothedOffset()
|
||||
logger.SetDouyuOffset(smoothed)
|
||||
logger.LogEvent("时间", "success", "%s校准完成: 斗鱼估算=%s 偏移=%dms 平滑偏移=%dms 误差<=%dms RTT=%dms 采样=%d 来源=%s",
|
||||
label,
|
||||
calibration.EstimatedServerNow().In(constants.CST).Format("2006-01-02 15:04:05.000"),
|
||||
calibration.Offset.Milliseconds(),
|
||||
smoothed.Milliseconds(),
|
||||
calibration.ErrorBound.Milliseconds(),
|
||||
calibration.RTT.Milliseconds(),
|
||||
calibration.Samples,
|
||||
@@ -1100,7 +1144,8 @@ func (s *Scheduler) ScheduleRun(req types.SeckillRequest) (map[string]any, error
|
||||
calibration := s.calibrateDouyuTime("定时")
|
||||
serverTime := calibration.EstimatedServerNow()
|
||||
|
||||
triggerTime := targetTime.Add(-internalAdvance)
|
||||
advance := dynamicAdvance(calibration)
|
||||
triggerTime := targetTime.Add(-advance)
|
||||
remain := triggerTime.Sub(serverTime)
|
||||
if remain <= 0 {
|
||||
return nil, errors.New("目标时间已过")
|
||||
@@ -1122,7 +1167,9 @@ func (s *Scheduler) ScheduleRun(req types.SeckillRequest) (map[string]any, error
|
||||
s.scheduleMu.Lock()
|
||||
s.replaceTimerLocked(&s.resyncSchedule, resyncRemain, func() {
|
||||
newCalibration := s.calibrateDouyuTime("临近")
|
||||
s.scheduleTimers(ap, targetTime, triggerTime, newCalibration, true)
|
||||
newAdvance := dynamicAdvance(newCalibration)
|
||||
newTriggerTime := targetTime.Add(-newAdvance)
|
||||
s.scheduleTimers(ap, targetTime, newTriggerTime, newCalibration, true)
|
||||
})
|
||||
s.scheduleMu.Unlock()
|
||||
|
||||
@@ -1131,12 +1178,11 @@ func (s *Scheduler) ScheduleRun(req types.SeckillRequest) (map[string]any, error
|
||||
warmupRemain = 0
|
||||
}
|
||||
|
||||
logger.LogEvent("定时", "success", "定时已设置: 目标=%s 触发=%s 提前=%dms 斗鱼估算=%s 本地偏移=%dms 误差<=%dms RTT=%dms 预热=%s 复校=%s",
|
||||
logger.LogEvent("定时", "success", "定时已设置: 目标=%s 触发=%s 提前=%dms 平滑偏移=%dms 误差<=%dms RTT=%dms 预热=%s 复校=%s",
|
||||
targetTime.In(constants.CST).Format("2006-01-02 15:04:05.000"),
|
||||
triggerTime.In(constants.CST).Format("2006-01-02 15:04:05.000"),
|
||||
internalAdvance.Milliseconds(),
|
||||
serverTime.In(constants.CST).Format("2006-01-02 15:04:05.000"),
|
||||
calibration.Offset.Milliseconds(),
|
||||
advance.Milliseconds(),
|
||||
s.smoothedOffset().Milliseconds(),
|
||||
calibration.ErrorBound.Milliseconds(),
|
||||
calibration.RTT.Milliseconds(),
|
||||
time.Now().Add(warmupRemain).In(constants.CST).Format("15:04:05.000"),
|
||||
|
||||
Reference in New Issue
Block a user