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

多次校对时间等等

Browse Source
This commit is contained in:
yml2213
2026-06-19 23:22:20 +08:00
parent dd96bc0402
commit f04ba4ed64
3 changed files with 392 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files
douyu_client.go
+125 -7
View File
@@ -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 拉取商城商品列表。
seckill.go
+213 -34
View File
@@ -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
}
types.go
+54 -2
View File
@@ -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