yml/dygou_go
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

合并远程main并保留和平小店功能

Browse Source
This commit is contained in:
yml2213
2026-08-09 10:52:56 +08:00
parent cd1c3db691 c61be33d77
commit 1b697ced09
3 changed files with 113 additions and 63 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/seckill/schedule_timer.go
+38 -27
View File
@@ -3,6 +3,7 @@ package seckill
import (
"errors"
"fmt"
"math"
"sync/atomic"
"time"
@@ -30,7 +31,8 @@ func (s *Scheduler) ScheduleRun(req types.SeckillRequest) (types.ScheduleResult,
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 types.ScheduleResult{}, errors.New("目标时间已过")
@@ -52,7 +54,8 @@ func (s *Scheduler) ScheduleRun(req types.SeckillRequest) (types.ScheduleResult,
s.scheduleMu.Lock()
s.replaceTimerLocked(&s.resyncSchedule, resyncRemain, func() {
newCalibration := s.calibrateDouyuTime("临近")
s.scheduleTimers(ap, targetTime, triggerTime, newCalibration, req.Attempts, req.DelayMs, true)
newTriggerTime := targetTime.Add(-dynamicAdvance(newCalibration))
s.scheduleTimers(ap, targetTime, newTriggerTime, newCalibration, req.Attempts, req.DelayMs, true)
})
s.scheduleMu.Unlock()
@@ -64,7 +67,7 @@ func (s *Scheduler) ScheduleRun(req types.SeckillRequest) (types.ScheduleResult,
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"),
internalAdvance.Milliseconds(),
advance.Milliseconds(),
serverTime.In(constants.CST).Format("2006-01-02 15:04:05.000"),
calibration.Offset.Milliseconds(),
calibration.ErrorBound.Milliseconds(),
@@ -73,7 +76,7 @@ func (s *Scheduler) ScheduleRun(req types.SeckillRequest) (types.ScheduleResult,
time.Now().Add(resyncRemain).In(constants.CST).Format("15:04:05.000"),
)
remainSecond := int(triggerTime.Sub(serverTime).Seconds())
remainSecond := int(math.Ceil(triggerTime.Sub(serverTime).Seconds()))
if remainSecond < 0 {
remainSecond = 0
}
@@ -95,26 +98,27 @@ func (s *Scheduler) scheduleTimers(accountProducts map[string]string, targetTime
if remain < 0 {
remain = 0
}
warmupRemain := remain - warmupBeforeTrigger
if warmupRemain < 0 {
warmupRemain = 0
}
s.scheduleMu.Lock()
defer s.scheduleMu.Unlock()
gen := atomic.AddInt64(&s.warmupGen, 1)
s.replaceTimerLocked(&s.warmupSchedule, warmupRemain, func() {
if atomic.LoadInt64(&s.warmupGen) != gen {
return
if !resync {
warmupRemain := remain - warmupBeforeTrigger
if warmupRemain < 0 {
warmupRemain = 0
}
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)
})
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)
@@ -134,14 +138,13 @@ func (s *Scheduler) scheduleTimers(accountProducts map[string]string, targetTime
s.RunNow(accountProducts)
})
if resync {
logger.LogEvent("定时", "success", "临近复校已应用: 目标=%s 触发=%s 斗鱼估算=%s 本地偏移=%dms 误差<=%dms RTT=%dms 预热=%s",
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(),
time.Now().Add(warmupRemain).In(constants.CST).Format("15:04:05.000"))
calibration.RTT.Milliseconds())
}
}
@@ -164,11 +167,19 @@ func (s *Scheduler) calibrateDouyuTime(label string) types.TimeCalibration {
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()
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"),
calibration.Offset.Milliseconds(),
smoothed.Milliseconds(),
calibration.ErrorBound.Milliseconds(),
calibration.RTT.Milliseconds(),
calibration.Samples,
internal/seckill/scheduler.go
+56 -19
View File
@@ -3,6 +3,7 @@ package seckill
import (
"fmt"
"math/rand"
"sort"
"strings"
"sync"
"time"
@@ -18,18 +19,19 @@ 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
xiaodianDefaultDelay = 500 * time.Millisecond
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
xiaodianDefaultDelay = 500 * time.Millisecond
)
type roundOutcome struct {
@@ -94,6 +96,9 @@ type Scheduler struct {
productsMu sync.RWMutex
products map[string]types.Product
calibrationMu sync.Mutex
calibrationOffsets []time.Duration // 最近N次校准偏移，用于中位数平滑
}
// SetStateChangeCallback 设置运行状态变化回调。
@@ -131,13 +136,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),
}
}
@@ -939,6 +945,37 @@ func classifyXiaodianRetryStop(message string) (account bool, product bool, reas
}
}
// 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()
if len(s.calibrationOffsets) == 0 {
return 0
}
sorted := append([]time.Duration(nil), s.calibrationOffsets...)
sort.Slice(sorted, func(i, j int) bool { return sorted[i] < sorted[j] })
return sorted[len(sorted)/2]
}
// dynamicAdvance 根据校准误差边界和 RTT 计算动态提前量，最少 50ms。
func dynamicAdvance(calibration types.TimeCalibration) time.Duration {
advance := calibration.ErrorBound + calibration.RTT/2
if advance < minInternalAdvance {
advance = minInternalAdvance
}
return advance
}
func uniqueAttemptAccounts(attempts []types.ExchangeAttempt) int {
seen := make(map[string]struct{}, len(attempts))
for _, attempt := range attempts {