package seckill import ( "errors" "fmt" "math" "sync/atomic" "time" "douyu_go/internal/constants" "douyu_go/internal/logger" "douyu_go/internal/types" "douyu_go/internal/utils" ) // ScheduleRun 注册一次定时抢购。 func (s *Scheduler) ScheduleRun(req types.SeckillRequest) (types.ScheduleResult, error) { if len(req.AccountProducts) == 0 { return types.ScheduleResult{}, errors.New("请选择账号和商品") } if req.TargetTime == "" { return types.ScheduleResult{}, errors.New("请设置目标时间") } targetLocal, err := time.ParseInLocation("2006-01-02 15:04:05", req.TargetTime, constants.CST) if err != nil { return types.ScheduleResult{}, fmt.Errorf("时间格式错误: %w", err) } targetTime := targetLocal.UTC() calibration := s.calibrateDouyuTime("定时") serverTime := calibration.EstimatedServerNow() advance := dynamicAdvance(calibration) triggerTime := targetTime.Add(-advance) remain := triggerTime.Sub(serverTime) if remain <= 0 { return types.ScheduleResult{}, errors.New("目标时间已过") } ap := utils.CloneStringMap(req.AccountProducts) s.scheduleMu.Lock() s.stopSchedulesLocked() s.clearWarmups() s.scheduleMu.Unlock() s.scheduleTimers(ap, targetTime, triggerTime, calibration, req.Attempts, req.DelayMs, false) resyncRemain := triggerTime.Sub(serverTime) - resyncBeforeTrigger if resyncRemain < 0 { resyncRemain = 0 } s.scheduleMu.Lock() s.replaceTimerLocked(&s.resyncSchedule, resyncRemain, func() { newCalibration := s.calibrateDouyuTime("临近") newTriggerTime := targetTime.Add(-dynamicAdvance(newCalibration)) s.scheduleTimers(ap, targetTime, newTriggerTime, newCalibration, req.Attempts, req.DelayMs, true) }) s.scheduleMu.Unlock() warmupRemain := remain - warmupBeforeTrigger if warmupRemain < 0 { warmupRemain = 0 } logger.LogEvent("定时", "success", "定时已设置: 目标=%s 触发=%s 提前=%dms 斗鱼估算=%s 本地偏移=%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"), advance.Milliseconds(), serverTime.In(constants.CST).Format("2006-01-02 15:04:05.000"), calibration.Offset.Milliseconds(), calibration.ErrorBound.Milliseconds(), calibration.RTT.Milliseconds(), time.Now().Add(warmupRemain).In(constants.CST).Format("15:04:05.000"), time.Now().Add(resyncRemain).In(constants.CST).Format("15:04:05.000"), ) remainSecond := int(math.Ceil(triggerTime.Sub(serverTime).Seconds())) if remainSecond < 0 { remainSecond = 0 } return types.ScheduleResult{ Success: true, TargetTime: targetTime.In(constants.CST).Format("2006-01-02 15:04:05"), TriggerTime: triggerTime.In(constants.CST).Format("2006-01-02 15:04:05"), RemainSecond: remainSecond, OffsetMs: calibration.Offset.Milliseconds(), ErrorBoundMs: calibration.ErrorBound.Milliseconds(), RttMs: calibration.RTT.Milliseconds(), }, nil } func (s *Scheduler) scheduleTimers(accountProducts map[string]string, targetTime, triggerTime time.Time, calibration types.TimeCalibration, attempts, delayMs int, resync bool) { serverTime := calibration.EstimatedServerNow() remain := triggerTime.Sub(serverTime) if remain < 0 { remain = 0 } s.scheduleMu.Lock() defer s.scheduleMu.Unlock() if !resync { warmupRemain := remain - warmupBeforeTrigger if warmupRemain < 0 { warmupRemain = 0 } gen := atomic.AddInt64(&s.warmupGen, 1) s.replaceTimerLocked(&s.warmupSchedule, warmupRemain, func() { if atomic.LoadInt64(&s.warmupGen) != gen { return } if s.clientSnapshot().Activity.IsXiaodian { return } logger.LogEvent("预热", "success", "预热触发: 本地=%s", time.Now().In(constants.CST).Format("15:04:05.000")) settings := s.settingsSnapshot() attempts := s.buildAttempts(accountProducts, settings.attempts) s.warmupAttempts(attempts) }) } s.replaceTimerLocked(&s.schedule, remain, func() { now := time.Now() estimatedServer := now.Add(calibration.Offset) logger.LogEvent("定时", "success", "定时触发: 本地=%s 斗鱼估算=%s 偏移=%dms 误差<=%dms", now.In(constants.CST).Format("15:04:05.000"), estimatedServer.In(constants.CST).Format("15:04:05.000"), calibration.Offset.Milliseconds(), calibration.ErrorBound.Milliseconds()) if !s.TryStartRun() { logger.LogEvent("抢购", "warn", "已有抢购任务正在运行,本次定时触发已跳过") return } if s.clientSnapshot().Activity.IsXiaodian { s.RunXiaodian(accountProducts, attempts, delayMs) return } s.RunNow(accountProducts) }) if resync { logger.LogEvent("定时", "success", "临近复校已应用: 目标=%s 触发=%s 斗鱼估算=%s 偏移=%dms 误差<=%dms RTT=%dms", targetTime.In(constants.CST).Format("2006-01-02 15:04:05.000"), triggerTime.In(constants.CST).Format("2006-01-02 15:04:05.000"), serverTime.In(constants.CST).Format("2006-01-02 15:04:05.000"), calibration.Offset.Milliseconds(), calibration.ErrorBound.Milliseconds(), calibration.RTT.Milliseconds()) } } // CancelSchedule 停止尚未触发的定时任务。 func (s *Scheduler) CancelSchedule() bool { s.scheduleMu.Lock() defer s.scheduleMu.Unlock() stopped := false if s.schedule != nil { stopped = s.schedule.Stop() } s.stopSchedulesLocked() s.clearWarmups() return stopped } func (s *Scheduler) calibrateDouyuTime(label string) types.TimeCalibration { cli := s.clientSnapshot() calibration, err := cli.BestServerTimeCalibration(timeCalibrateRounds, timeCalibrateWait) if err != nil { logger.LogEvent("时间", "warn", "%s校准降级: %v", label, err) } if !calibration.Fallback { s.recordCalibrationOffset(calibration.Offset) } smoothed := s.smoothedOffset() if smoothed == 0 && calibration.Fallback { smoothed = calibration.Offset } calibration.Offset = smoothed logger.SetDouyuOffset(smoothed) logger.LogEvent("时间", "success", "%s校准完成: 斗鱼估算=%s 平滑偏移=%dms 误差<=%dms RTT=%dms 采样=%d 来源=%s", label, calibration.EstimatedServerNow().In(constants.CST).Format("2006-01-02 15:04:05.000"), smoothed.Milliseconds(), calibration.ErrorBound.Milliseconds(), calibration.RTT.Milliseconds(), calibration.Samples, calibration.Source) return calibration } func (s *Scheduler) replaceTimerLocked(timer **time.Timer, d time.Duration, fn func()) { if *timer != nil { (*timer).Stop() } *timer = time.AfterFunc(d, fn) } func (s *Scheduler) 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 } }