diff --git a/douyu_client.go b/douyu_client.go index 5a11c67..8ff060c 100644 --- a/douyu_client.go +++ b/douyu_client.go @@ -33,20 +33,125 @@ func NewDouyuClient(timeAPIURL string, trace *traceRecorder) *DouyuClient { // ServerTime 从商品详情接口读取 data.preExchange.currentTime。 func (c *DouyuClient) ServerTime() (time.Time, error) { + sample, err := c.ServerTimeSample() + return sample.EstimatedServerNow(), err +} + +// ServerTimeSample 从斗鱼秒级 currentTime 估算毫秒级服务端时间。 +func (c *DouyuClient) ServerTimeSample() (TimeSample, error) { var lastErr error for i := 0; i < 3; i++ { - t, err := c.fetchServerTimeOnce() + sample, err := c.fetchServerTimeOnce() if err == nil { - return t, nil + sample.Samples = 1 + sample.Source = "single" + return sample, nil } lastErr = err time.Sleep(100 * time.Millisecond) } - return time.Now().UTC(), fmt.Errorf("服务端时间获取失败,已降级本地时间: %w", lastErr) + now := time.Now() + return TimeSample{ + ServerTime: now.UTC(), + LocalTime: now, + Offset: 0, + Source: "localFallback", + Fallback: true, + }, fmt.Errorf("服务端时间获取失败,已降级本地时间: %w", lastErr) +} + +// CalibratedServerTimeSample 通过等待 currentTime 跳秒估算本地与斗鱼秒边界的偏移。 +func (c *DouyuClient) CalibratedServerTimeSample(maxWait time.Duration) (TimeSample, error) { + first, err := c.ServerTimeSample() + if err != nil || maxWait <= 0 { + return first, err + } + + deadline := time.Now().Add(maxWait) + prev := first + for time.Now().Before(deadline) { + time.Sleep(40 * time.Millisecond) + current, err := c.fetchServerTimeOnce() + if err != nil { + continue + } + if current.RawCurrentTime <= prev.RawCurrentTime { + prev = current + continue + } + + sampleGap := current.LocalTime.Sub(prev.LocalTime) + localBoundary := prev.LocalTime.Add(sampleGap / 2) + serverBoundary := time.Unix(current.RawCurrentTime, 0).UTC() + offset := serverBoundary.Sub(localBoundary.UTC()) + current.Offset = offset + current.ServerTime = current.EstimatedServerNow() + current.LocalTime = time.Now() + current.RTT += first.RTT + current.ErrorBound = sampleGap / 2 + current.Samples = 2 + current.Source = "boundary" + return current, nil + } + + first.ServerTime = first.EstimatedServerNow() + return first, nil +} + +// BestServerTimeCalibration 多次校准,选择误差范围最小、RTT更低的一次。 +func (c *DouyuClient) BestServerTimeCalibration(rounds int, maxWait time.Duration) (TimeCalibration, error) { + if rounds <= 0 { + rounds = 1 + } + var best TimeSample + var lastErr error + for i := 0; i < rounds; i++ { + sample, err := c.CalibratedServerTimeSample(maxWait) + if err != nil { + lastErr = err + } + if sample.LocalTime.IsZero() { + continue + } + sample.Samples = max(sample.Samples, 1) + if best.LocalTime.IsZero() || betterTimeSample(sample, best) { + best = sample + } + } + if best.LocalTime.IsZero() { + now := time.Now() + return TimeCalibration{ + CreatedAt: now, + Source: "localFallback", + Fallback: true, + }, fmt.Errorf("斗鱼时间校准失败: %w", lastErr) + } + return TimeCalibration{ + Offset: best.Offset, + ErrorBound: best.ErrorBound, + RTT: best.RTT, + Samples: rounds, + CreatedAt: time.Now(), + Source: best.Source, + Fallback: best.Fallback, + }, lastErr +} + +func betterTimeSample(candidate, current TimeSample) bool { + if candidate.ErrorBound != 0 && current.ErrorBound == 0 { + return true + } + if candidate.ErrorBound == 0 && current.ErrorBound != 0 { + return false + } + if candidate.ErrorBound != current.ErrorBound { + return candidate.ErrorBound < current.ErrorBound + } + return candidate.RTT < current.RTT } // fetchServerTimeOnce 执行一次服务端时间请求。 -func (c *DouyuClient) fetchServerTimeOnce() (time.Time, error) { +func (c *DouyuClient) fetchServerTimeOnce() (TimeSample, error) { url := appendCacheBuster(c.timeAPIURL) var result struct { APIResponse @@ -56,14 +161,27 @@ func (c *DouyuClient) fetchServerTimeOnce() (time.Time, error) { } `json:"preExchange"` } `json:"data"` } + start := time.Now() if err := c.getJSON(url, "", "服务时间", &result); err != nil { - return time.Time{}, err + return TimeSample{}, err } + end := time.Now() ts, err := strconv.ParseInt(result.Data.PreExchange.CurrentTime, 10, 64) if err != nil || ts <= 0 { - return time.Time{}, fmt.Errorf("currentTime 无效: %q", result.Data.PreExchange.CurrentTime) + return TimeSample{}, fmt.Errorf("currentTime 无效: %q", result.Data.PreExchange.CurrentTime) } - return time.Unix(ts, 0).UTC(), nil + rtt := end.Sub(start) + + // 斗鱼 currentTime 只有秒级;用响应到达时本地时钟的毫秒部分补齐,避免定时提前量被秒级截断吃掉。 + estimatedLocal := start.Add(rtt / 2) + serverTime := time.Unix(ts, 0).Add(time.Duration(estimatedLocal.Nanosecond())).UTC() + return TimeSample{ + ServerTime: serverTime, + LocalTime: estimatedLocal, + RTT: rtt, + Offset: serverTime.Sub(estimatedLocal.UTC()), + RawCurrentTime: ts, + }, nil } // Products 拉取商城商品列表。 diff --git a/seckill.go b/seckill.go index a09377b..cbfff32 100644 --- a/seckill.go +++ b/seckill.go @@ -8,6 +8,14 @@ import ( "time" ) +const ( + warmupBeforeTrigger = 2 * time.Second + warmupMaxAge = 30 * time.Second + timeCalibrateRounds = 3 + timeCalibrateWait = 1200 * time.Millisecond + resyncBeforeTrigger = 10 * time.Second +) + // runNow 立即执行一批抢购任务。 func (s *appState) runNow(accountProducts map[string]string) { s.clearResults() @@ -72,7 +80,113 @@ func (s *appState) buildAttempts(accountProducts map[string]string) []ExchangeAt return attempts } +func warmupKey(attempt ExchangeAttempt) string { + return fmt.Sprintf("%s|%s|%d", attempt.Account.Name, attempt.Product.CommodityID, attempt.Index) +} + +func (s *appState) setWarmup(key string, data WarmupData) { + s.warmupsMu.Lock() + defer s.warmupsMu.Unlock() + if s.warmups == nil { + s.warmups = make(map[string]WarmupData) + } + s.warmups[key] = data +} + +func (s *appState) warmupFor(attempt ExchangeAttempt) (WarmupData, bool) { + s.warmupsMu.RLock() + data, ok := s.warmups[warmupKey(attempt)] + s.warmupsMu.RUnlock() + if !ok || data.CreatedAt.IsZero() || time.Since(data.CreatedAt) > warmupMaxAge { + return WarmupData{}, false + } + return data, true +} + +func (s *appState) clearWarmups() { + s.warmupsMu.Lock() + defer s.warmupsMu.Unlock() + s.warmups = make(map[string]WarmupData) +} + +func (s *appState) warmupAttempts(attempts []ExchangeAttempt) { + if len(attempts) == 0 { + return + } + s.clearWarmups() + logEvent("预热", "info", "开始预热: %d 个请求", len(attempts)) + + limit := make(chan struct{}, defaultConcurrency) + var wg sync.WaitGroup + for _, attempt := range attempts { + wg.Add(1) + go func(attempt ExchangeAttempt) { + defer wg.Done() + limit <- struct{}{} + defer func() { <-limit }() + + start := time.Now() + csrf, err := s.client.GenerateCSRF(attempt.Account.Cookie) + if err != nil { + logEvent("预热", "warn", "CSRF预热失败: 账号=%s 商品=%s err=%v", attempt.Account.Name, attempt.Product.CommodityName, err) + return + } + roleName, err := s.client.RoleName(attempt.Account.Cookie) + if err != nil { + logEvent("预热", "warn", "角色预热失败: 账号=%s err=%v", attempt.Account.Name, err) + } + s.setWarmup(warmupKey(attempt), WarmupData{ + CSRF: csrf, + RoleName: roleName, + CreatedAt: time.Now(), + }) + logEvent("预热", "success", "预热完成: 账号=%s 商品=%s 耗时=%s", attempt.Account.Name, attempt.Product.CommodityName, roundDuration(time.Since(start))) + }(attempt) + } + wg.Wait() + logEvent("预热", "info", "预热任务结束") +} + +func (s *appState) calibrateDouyuTime(label string) TimeCalibration { + calibration, err := s.client.BestServerTimeCalibration(timeCalibrateRounds, timeCalibrateWait) + if err != nil { + logEvent("时间", "warn", "%s校准降级: %v", label, err) + } + logEvent("时间", "success", "%s校准完成: 斗鱼估算=%s 偏移=%dms 误差<=%dms RTT=%dms 采样=%d 来源=%s", + label, + calibration.EstimatedServerNow().In(cst).Format("2006-01-02 15:04:05.000"), + calibration.Offset.Milliseconds(), + calibration.ErrorBound.Milliseconds(), + calibration.RTT.Milliseconds(), + calibration.Samples, + calibration.Source) + return calibration +} + +func (s *appState) replaceTimerLocked(timer **time.Timer, d time.Duration, fn func()) { + if *timer != nil { + (*timer).Stop() + } + *timer = time.AfterFunc(d, fn) +} + +func (s *appState) stopSchedulesLocked() { + if s.schedule != nil { + s.schedule.Stop() + s.schedule = nil + } + if s.warmupSchedule != nil { + s.warmupSchedule.Stop() + s.warmupSchedule = nil + } + if s.resyncSchedule != nil { + s.resyncSchedule.Stop() + s.resyncSchedule = nil + } +} + // runAttempt 按抓包链路执行一次兑换:CSRF -> 限制 -> 预兑换 -> 正式兑换。 +// 同一次抢购尝试只生成一个 CSRF,后续接口共用,减少临门请求耗时。 func (s *appState) runAttempt(attempt ExchangeAttempt) ExchangeResult { result := ExchangeResult{ Account: attempt.Account.Name, @@ -81,13 +195,24 @@ func (s *appState) runAttempt(attempt ExchangeAttempt) ExchangeResult { } logEvent("抢购", "info", "执行兑换链路: 账号=%s 商品=%s 第%d包", attempt.Account.Name, attempt.Product.CommodityName, attempt.Index) - limitCSRF, err := s.client.GenerateCSRF(attempt.Account.Cookie) - if err != nil { - result.Message = "失败|生成CSRF失败: " + err.Error() - return result + var roleName string + csrf := "" + if warmup, ok := s.warmupFor(attempt); ok { + csrf = warmup.CSRF + roleName = warmup.RoleName + logEvent("抢购", "info", "使用预热CSRF: 账号=%s 商品=%s 预热距今=%s", attempt.Account.Name, attempt.Product.CommodityName, roundDuration(time.Since(warmup.CreatedAt))) + } + if csrf == "" { + var err error + csrf, err = s.client.GenerateCSRF(attempt.Account.Cookie) + if err != nil { + result.Message = "失败|生成CSRF失败: " + err.Error() + return result + } + logEvent("抢购", "info", "CSRF已生成,本次兑换链路复用同一个 token: 账号=%s 商品=%s", attempt.Account.Name, attempt.Product.CommodityName) } - limit, err := s.client.BatchExchangeLimit(attempt.Account.Cookie, attempt.Product.CommodityID, limitCSRF) + limit, err := s.client.BatchExchangeLimit(attempt.Account.Cookie, attempt.Product.CommodityID, csrf) if err != nil { result.Message = "失败|批量限制检查失败: " + err.Error() return result @@ -98,30 +223,22 @@ func (s *appState) runAttempt(attempt ExchangeAttempt) ExchangeResult { } logEvent("抢购", "info", "批量限制通过: 账号=%s 商品=%s limit=%d", attempt.Account.Name, attempt.Product.CommodityName, limit) - preCSRF, err := s.client.GenerateCSRF(attempt.Account.Cookie) - if err != nil { - result.Message = "失败|生成预兑换CSRF失败: " + err.Error() - return result - } - if err := s.client.PreExchange(attempt.Account.Cookie, attempt.Product.CommodityID, preCSRF); err != nil { + if err := s.client.PreExchange(attempt.Account.Cookie, attempt.Product.CommodityID, csrf); err != nil { result.Message = "失败|预兑换失败: " + err.Error() return result } - exchangeCSRF, err := s.client.GenerateCSRF(attempt.Account.Cookie) - if err != nil { - result.Message = "失败|生成兑换CSRF失败: " + err.Error() - return result - } - exchangeInfo, err := s.client.Exchange(attempt.Account.Cookie, attempt.Product.CommodityID, exchangeCSRF) + exchangeInfo, err := s.client.Exchange(attempt.Account.Cookie, attempt.Product.CommodityID, csrf) if err != nil { result.Message = "失败|" + err.Error() return result } - roleName, err := s.client.RoleName(attempt.Account.Cookie) - if err != nil { - logEvent("抢购", "warn", "%s 角色名查询失败: %v", attempt.Account.Name, err) + if roleName == "" { + roleName, err = s.client.RoleName(attempt.Account.Cookie) + if err != nil { + logEvent("抢购", "warn", "%s 角色名查询失败: %v", attempt.Account.Name, err) + } } result.OK = true @@ -149,10 +266,8 @@ func (s *appState) scheduleRun(req SeckillRequest) (map[string]any, error) { } targetTime := targetLocal.UTC() - serverTime, err := s.client.ServerTime() - if err != nil { - logEvent("定时", "warn", "定时使用降级时间: %v", err) - } + calibration := s.calibrateDouyuTime("定时") + serverTime := calibration.EstimatedServerNow() advance := time.Duration(req.Advance) * time.Millisecond triggerTime := targetTime.Add(-advance) @@ -162,20 +277,41 @@ func (s *appState) scheduleRun(req SeckillRequest) (map[string]any, error) { } ap := cloneStringMap(req.AccountProducts) + attempts := s.buildAttempts(ap) + s.clearWarmups() s.scheduleMu.Lock() - if s.schedule != nil { - s.schedule.Stop() + s.stopSchedulesLocked() + s.scheduleMu.Unlock() + + s.scheduleTimers(ap, attempts, targetTime, triggerTime, calibration, false) + + resyncRemain := triggerTime.Sub(serverTime) - resyncBeforeTrigger + if resyncRemain < 0 { + resyncRemain = 0 } - s.schedule = time.AfterFunc(remain, func() { - logEvent("定时", "success", "定时触发: 本地=%s", time.Now().In(cst).Format("15:04:05.000")) - s.runNow(ap) + + s.scheduleMu.Lock() + s.replaceTimerLocked(&s.resyncSchedule, resyncRemain, func() { + newCalibration := s.calibrateDouyuTime("临近") + s.scheduleTimers(ap, attempts, targetTime, triggerTime, newCalibration, true) }) s.scheduleMu.Unlock() - logEvent("定时", "success", "定时已设置: 目标=%s 触发=%s 提前=%dms", + warmupRemain := remain - warmupBeforeTrigger + if warmupRemain < 0 { + warmupRemain = 0 + } + + logEvent("定时", "success", "定时已设置: 目标=%s 触发=%s 提前=%dms 斗鱼估算=%s 本地偏移=%dms 误差<=%dms RTT=%dms 预热=%s 复校=%s", targetTime.In(cst).Format("2006-01-02 15:04:05.000"), triggerTime.In(cst).Format("2006-01-02 15:04:05.000"), req.Advance, + serverTime.In(cst).Format("2006-01-02 15:04:05.000"), + calibration.Offset.Milliseconds(), + calibration.ErrorBound.Milliseconds(), + calibration.RTT.Milliseconds(), + time.Now().Add(warmupRemain).In(cst).Format("15:04:05.000"), + time.Now().Add(resyncRemain).In(cst).Format("15:04:05.000"), ) return map[string]any{ @@ -185,17 +321,60 @@ func (s *appState) scheduleRun(req SeckillRequest) (map[string]any, error) { "advance": req.Advance, "remainSecond": int(remain.Seconds()), "serverTime": serverTime.In(cst).Format("2006-01-02 15:04:05.000"), + "offsetMs": calibration.Offset.Milliseconds(), + "errorBoundMs": calibration.ErrorBound.Milliseconds(), + "rttMs": calibration.RTT.Milliseconds(), }, nil } +func (s *appState) scheduleTimers(accountProducts map[string]string, attempts []ExchangeAttempt, targetTime, triggerTime time.Time, calibration TimeCalibration, resync bool) { + serverTime := calibration.EstimatedServerNow() + remain := triggerTime.Sub(serverTime) + if remain < 0 { + remain = 0 + } + warmupRemain := remain - warmupBeforeTrigger + if warmupRemain < 0 { + warmupRemain = 0 + } + + s.scheduleMu.Lock() + defer s.scheduleMu.Unlock() + s.replaceTimerLocked(&s.warmupSchedule, warmupRemain, func() { + logEvent("预热", "success", "预热触发: 本地=%s", time.Now().In(cst).Format("15:04:05.000")) + s.warmupAttempts(attempts) + }) + s.replaceTimerLocked(&s.schedule, remain, func() { + now := time.Now() + estimatedServer := now.Add(calibration.Offset) + logEvent("定时", "success", "定时触发: 本地=%s 斗鱼估算=%s 偏移=%dms 误差<=%dms", + now.In(cst).Format("15:04:05.000"), + estimatedServer.In(cst).Format("15:04:05.000"), + calibration.Offset.Milliseconds(), + calibration.ErrorBound.Milliseconds()) + s.runNow(accountProducts) + }) + if resync { + logEvent("定时", "success", "临近复校已应用: 目标=%s 触发=%s 斗鱼估算=%s 本地偏移=%dms 误差<=%dms RTT=%dms 预热=%s", + targetTime.In(cst).Format("2006-01-02 15:04:05.000"), + triggerTime.In(cst).Format("2006-01-02 15:04:05.000"), + serverTime.In(cst).Format("2006-01-02 15:04:05.000"), + calibration.Offset.Milliseconds(), + calibration.ErrorBound.Milliseconds(), + calibration.RTT.Milliseconds(), + time.Now().Add(warmupRemain).In(cst).Format("15:04:05.000")) + } +} + // cancelSchedule 停止尚未触发的定时任务。 func (s *appState) cancelSchedule() bool { s.scheduleMu.Lock() defer s.scheduleMu.Unlock() - if s.schedule == nil { - return false + stopped := false + if s.schedule != nil { + stopped = s.schedule.Stop() } - stopped := s.schedule.Stop() - s.schedule = nil + s.stopSchedulesLocked() + s.clearWarmups() return stopped } diff --git a/types.go b/types.go index a44c80b..6bbb54b 100644 --- a/types.go +++ b/types.go @@ -100,8 +100,13 @@ type appState struct { resultsMu sync.RWMutex results map[string]string - scheduleMu sync.Mutex - schedule *time.Timer + scheduleMu sync.Mutex + schedule *time.Timer + warmupSchedule *time.Timer + resyncSchedule *time.Timer + + warmupsMu sync.RWMutex + warmups map[string]WarmupData fontMu sync.Mutex font *truetype.Font @@ -144,6 +149,53 @@ type ExchangeResult struct { Message string } +type TimeSample struct { + ServerTime time.Time + LocalTime time.Time + RTT time.Duration + Offset time.Duration + ErrorBound time.Duration + RawCurrentTime int64 + Samples int + Source string + Fallback bool +} + +type TimeCalibration struct { + Offset time.Duration + ErrorBound time.Duration + RTT time.Duration + Samples int + CreatedAt time.Time + Source string + Fallback bool +} + +func (c TimeCalibration) EstimatedServerNow() time.Time { + if c.CreatedAt.IsZero() { + return time.Now().UTC() + } + return time.Now().Add(c.Offset).UTC() +} + +func (c TimeCalibration) IsZero() bool { + return c.CreatedAt.IsZero() +} + +type WarmupData struct { + CSRF string + RoleName string + CreatedAt time.Time +} + +// EstimatedServerNow 根据采样偏移估算当前斗鱼时间。 +func (t TimeSample) EstimatedServerNow() time.Time { + if t.LocalTime.IsZero() { + return t.ServerTime + } + return time.Now().Add(t.Offset).UTC() +} + type memoryLogStore struct { mu sync.RWMutex nextID int64