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-27 15:56:19 +08:00
parent db917698f9
commit 8029333613
7 changed files with 29038 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files
frontend/app.js
+1
View File
@@ -642,6 +642,7 @@ async function saveSettings() {
logLevel: document.getElementById('setLogLevel').value,
concurrency: parseInt(document.getElementById('setConcurrency').value) || 80,
attempts: parseInt(document.getElementById('setAttempts').value) || 1,
seckillMode: 1,
};
try {
let d = await App.SaveSettings(cfg);
frontend/src/components/modal/SettingsModal.vue
+2 -2
View File
@@ -13,7 +13,7 @@ const formData = ref({
concurrency: 150,
attempts: 1,
retryStages: [] as RetryStage[],
seckillMode: 0
seckillMode: 1
})
const open = async () => {
@@ -163,7 +163,7 @@ defineExpose({ open, close })
标准模式：每次兑换都执行完整校验流程（CSRF→限制检查→预兑换→兑换）
</template>
<template v-else>
激进模式：账号通过限制检查和预兑换后，后续轮次跳过限制检查和预兑换，直接调用兑换接口
激进模式：默认策略，优先复用预热结果直兑，必要时自动回退完整校验
</template>
</div>
</div>
frontend/src/composables/useSettings.ts
+3 -3
View File
@@ -20,7 +20,7 @@ export interface Settings {
const App = window.go.main.App
const defaultRetryStages: RetryStage[] = [
{ maxElapsed: 4, interval: 1000 },
{ maxElapsed: 4, interval: 120 },
{ maxElapsed: 12, interval: 200 },
{ maxElapsed: 30, interval: 500 },
{ maxElapsed: 75, interval: 1000 }
@@ -34,7 +34,7 @@ const settings = ref<Settings>({
concurrency: 150,
attempts: 1,
retryStages: [...defaultRetryStages],
seckillMode: 0
seckillMode: 1
})
export function useSettings() {
@@ -53,7 +53,7 @@ export function useSettings() {
retryStages: cfg.retryStages && cfg.retryStages.length > 0
? cfg.retryStages
: [...defaultRetryStages],
seckillMode: cfg.seckillMode || 0
seckillMode: cfg.seckillMode ?? 1
}
} catch (e: any) {
showToast('读取设置失败', e.message, 'error')
internal/config/config.go
+2 -2
View File
@@ -82,14 +82,14 @@ func defaultConfig() types.Config {
Concurrency: constants.DefaultConcurrency,
Attempts: constants.DefaultAttempts,
RetryStages: defaultRetryStages(),
SeckillMode: types.SeckillModeNormal,
SeckillMode: types.SeckillModeAggressive,
}
}
// defaultRetryStages 返回默认重试间隔阶段配置。
func defaultRetryStages() []types.RetryStage {
return []types.RetryStage{
{MaxElapsed: 4, Interval: 1000},
{MaxElapsed: 4, Interval: 120},
{MaxElapsed: 12, Interval: 200},
{MaxElapsed: 30, Interval: 500},
{MaxElapsed: 75, Interval: 1000},
internal/seckill/scheduler.go
+125 -34
View File
@@ -42,6 +42,13 @@ type roundOutcome struct {
attempts int
}
type runSettings struct {
concurrency int
attempts int
retryStages []types.RetryStage
seckillMode types.SeckillMode
}
// Scheduler 抢购调度器。
type Scheduler struct {
client *client.DouyuClient
@@ -50,6 +57,7 @@ type Scheduler struct {
successImages *image.SuccessImageSaver
settingsMu sync.RWMutex
concurrency int // 抢购并发数
attempts int // 单号重试次数
@@ -100,8 +108,8 @@ func New(cli *client.DouyuClient, paths types.AppPaths, font *truetype.Font, con
),
concurrency: concurrency,
attempts: attempts,
retryStages: retryStages,
seckillMode: seckillMode,
retryStages: cloneRetryStages(retryStages),
seckillMode: normalizeSeckillMode(seckillMode),
results: make(map[string]string),
attemptCache: make(map[string]*types.AttemptCache),
products: make(map[string]types.Product),
@@ -134,6 +142,8 @@ func (s *Scheduler) UpdateProducts(products []types.Product) {
// UpdateSettings 运行时更新并发数、重试次数、重试间隔和抢购模式。
func (s *Scheduler) UpdateSettings(concurrency, attempts int, retryStages []types.RetryStage, seckillMode types.SeckillMode) {
s.settingsMu.Lock()
defer s.settingsMu.Unlock()
if concurrency > 0 {
s.concurrency = concurrency
}
@@ -141,9 +151,45 @@ func (s *Scheduler) UpdateSettings(concurrency, attempts int, retryStages []type
s.attempts = attempts
}
if len(retryStages) > 0 {
s.retryStages = retryStages
s.retryStages = cloneRetryStages(retryStages)
}
s.seckillMode = normalizeSeckillMode(seckillMode)
}
func (s *Scheduler) settingsSnapshot() runSettings {
s.settingsMu.RLock()
settings := runSettings{
concurrency: s.concurrency,
attempts: s.attempts,
retryStages: cloneRetryStages(s.retryStages),
seckillMode: normalizeSeckillMode(s.seckillMode),
}
s.settingsMu.RUnlock()
if settings.concurrency <= 0 {
settings.concurrency = constants.DefaultConcurrency
}
if settings.attempts <= 0 {
settings.attempts = constants.DefaultAttempts
}
return settings
}
func cloneRetryStages(stages []types.RetryStage) []types.RetryStage {
if len(stages) == 0 {
return nil
}
out := make([]types.RetryStage, len(stages))
copy(out, stages)
return out
}
func normalizeSeckillMode(mode types.SeckillMode) types.SeckillMode {
switch mode {
case types.SeckillModeNormal, types.SeckillModeAggressive:
return mode
default:
return types.SeckillModeAggressive
}
s.seckillMode = seckillMode
}
// LoadAccounts 从账号文件加载账号。
@@ -326,7 +372,8 @@ func (s *Scheduler) RunNow(accountProducts map[string]string) {
defer s.finishRun()
s.ClearResults()
allAttempts := s.buildAttempts(accountProducts)
settings := s.settingsSnapshot()
allAttempts := s.buildAttempts(accountProducts, settings.attempts)
if len(allAttempts) == 0 {
logger.LogEvent("抢购", "warn", "没有可执行的抢购任务")
return
@@ -342,7 +389,7 @@ func (s *Scheduler) RunNow(accountProducts map[string]string) {
round := 0
s.clearAttemptCache()
if s.seckillMode == types.SeckillModeAggressive {
if settings.seckillMode == types.SeckillModeAggressive {
s.initCacheFromWarmup(allAttempts)
}
@@ -364,7 +411,7 @@ func (s *Scheduler) RunNow(accountProducts map[string]string) {
break
}
roundAttempts := s.pickRoundAttempts(pendingAttempts, round)
roundAttempts := pickRoundAttempts(pendingAttempts, round, settings.attempts)
if len(roundAttempts) == 0 {
logger.LogEvent("抢购", "info", "第%d轮无可执行账号，停止抢购", round)
break
@@ -375,7 +422,7 @@ func (s *Scheduler) RunNow(accountProducts map[string]string) {
if len(roundAttempts) == 0 {
sleepFor := nextReady.Sub(now)
if sleepFor <= 0 {
sleepFor = s.getRetryInterval(time.Since(startTime))
sleepFor = retryIntervalFor(settings.retryStages, time.Since(startTime))
}
if sleepFor > remaining {
sleepFor = remaining
@@ -389,9 +436,9 @@ func (s *Scheduler) RunNow(accountProducts map[string]string) {
logger.LogEvent("抢购", "info", "第%d轮: 待抢购 %d 个账号，本轮执行=%d，剩余时间=%v，模式=%s",
round, uniqueAttemptAccounts(pendingAttempts), len(roundAttempts), utils.RoundDuration(remaining),
s.getModeName())
modeName(settings.seckillMode))
outcome := s.executeRound(roundAttempts)
outcome := s.executeRound(roundAttempts, settings)
totalAttempts += outcome.attempts
for account := range outcome.successAccounts {
@@ -401,6 +448,7 @@ func (s *Scheduler) RunNow(accountProducts map[string]string) {
for account, reason := range outcome.stoppedAccounts {
if stoppedAccounts[account] == "" {
stoppedAccounts[account] = reason
logger.LogAccountEvent("抢购", "warn", account, "账号停止重试: 账号=%s 原因=%s", account, reason)
}
}
for product, reason := range outcome.stoppedProducts {
@@ -425,7 +473,7 @@ func (s *Scheduler) RunNow(accountProducts map[string]string) {
break
}
retryInterval := s.getRetryInterval(time.Since(startTime))
retryInterval := retryIntervalFor(settings.retryStages, time.Since(startTime))
if retryInterval > 0 && time.Since(startTime)+retryInterval < seckillMaxDuration {
time.Sleep(jitterDuration(retryInterval))
}
@@ -467,13 +515,13 @@ func filterCooldownAttempts(attempts []types.ExchangeAttempt, cooldownAccounts m
}
// pickRoundAttempts 每轮每个账号最多执行一次，避免同账号并发过猛触发频控。
func (s *Scheduler) pickRoundAttempts(attempts []types.ExchangeAttempt, round int) []types.ExchangeAttempt {
func pickRoundAttempts(attempts []types.ExchangeAttempt, round, attemptsPerAccount int) []types.ExchangeAttempt {
if len(attempts) == 0 {
return nil
}
preferredIndex := 1
if s.attempts > 0 {
preferredIndex = ((round - 1) % s.attempts) + 1
if attemptsPerAccount > 0 {
preferredIndex = ((round - 1) % attemptsPerAccount) + 1
}
byAccount := make(map[string]types.ExchangeAttempt)
@@ -502,7 +550,7 @@ func (s *Scheduler) pickRoundAttempts(attempts []types.ExchangeAttempt, round in
}
// executeRound 执行一轮抢购，并返回成功和终止重试的账号。
func (s *Scheduler) executeRound(attempts []types.ExchangeAttempt) roundOutcome {
func (s *Scheduler) executeRound(attempts []types.ExchangeAttempt, settings runSettings) roundOutcome {
outcome := roundOutcome{
successAccounts: make(map[string]bool),
stoppedAccounts: make(map[string]string),
@@ -511,7 +559,7 @@ func (s *Scheduler) executeRound(attempts []types.ExchangeAttempt) roundOutcome
}
var mu sync.Mutex
limit := make(chan struct{}, s.concurrency)
limit := make(chan struct{}, settings.concurrency)
var wg sync.WaitGroup
for _, attempt := range attempts {
@@ -523,7 +571,7 @@ func (s *Scheduler) executeRound(attempts []types.ExchangeAttempt) roundOutcome
// 根据模式选择执行方法
var result types.ExchangeResult
if s.seckillMode == types.SeckillModeAggressive {
if settings.seckillMode == types.SeckillModeAggressive {
result = s.runAttemptFast(attempt)
} else {
result = s.runAttempt(attempt)
@@ -569,20 +617,20 @@ func (s *Scheduler) executeRound(attempts []types.ExchangeAttempt) roundOutcome
return outcome
}
// getModeName 返回当前抢购模式的名称。
func (s *Scheduler) getModeName() string {
if s.seckillMode == types.SeckillModeAggressive {
// modeName 返回抢购模式名称。
func modeName(mode types.SeckillMode) string {
if mode == types.SeckillModeAggressive {
return "激进"
}
return "标准"
}
// getRetryInterval 根据配置的重试阶段计算当前应该使用的重试间隔。
func (s *Scheduler) getRetryInterval(elapsed time.Duration) time.Duration {
// retryIntervalFor 根据配置的重试阶段计算当前应该使用的重试间隔。
func retryIntervalFor(retryStages []types.RetryStage, elapsed time.Duration) time.Duration {
elapsedSec := int(elapsed.Seconds())
// 使用配置的阶段
for _, stage := range s.retryStages {
for _, stage := range retryStages {
if stage.MaxElapsed > 0 && elapsedSec < stage.MaxElapsed && stage.Interval > 0 {
return time.Duration(stage.Interval) * time.Millisecond
}
@@ -636,7 +684,10 @@ func uniqueAttemptAccounts(attempts []types.ExchangeAttempt) int {
return len(seen)
}
func (s *Scheduler) buildAttempts(accountProducts map[string]string) []types.ExchangeAttempt {
func (s *Scheduler) buildAttempts(accountProducts map[string]string, attemptsPerAccount int) []types.ExchangeAttempt {
if attemptsPerAccount <= 0 {
attemptsPerAccount = constants.DefaultAttempts
}
accounts := s.AccountSnapshot()
var attempts []types.ExchangeAttempt
for _, account := range accounts {
@@ -645,7 +696,7 @@ func (s *Scheduler) buildAttempts(accountProducts map[string]string) []types.Exc
continue
}
product := s.productForCommodityID(commodityID)
for i := 1; i <= s.attempts; i++ {
for i := 1; i <= attemptsPerAccount; i++ {
attempts = append(attempts, types.ExchangeAttempt{
Account: account,
Product: product,
@@ -677,7 +728,7 @@ func (s *Scheduler) enqueueSuccessImage(account types.Account, product types.Pro
}
func warmupKey(attempt types.ExchangeAttempt) string {
return fmt.Sprintf("%s|%s|%d", attempt.Account.Name, attempt.Product.CommodityID, attempt.Index)
return fmt.Sprintf("%s|%s", attempt.Account.Name, attempt.Product.CommodityID)
}
func (s *Scheduler) setWarmup(key string, data types.WarmupData) {
@@ -745,13 +796,15 @@ func (s *Scheduler) clearAttemptCache() {
}
func (s *Scheduler) warmupAttempts(attempts []types.ExchangeAttempt) {
attempts = uniqueWarmupAttempts(attempts)
if len(attempts) == 0 {
return
}
s.clearWarmups()
logger.LogEvent("预热", "info", "开始预热: %d 个请求", len(attempts))
limit := make(chan struct{}, s.concurrency)
settings := s.settingsSnapshot()
limit := make(chan struct{}, settings.concurrency)
var wg sync.WaitGroup
for _, attempt := range attempts {
wg.Add(1)
@@ -773,6 +826,7 @@ func (s *Scheduler) warmupAttempts(attempts []types.ExchangeAttempt) {
s.setWarmup(warmupKey(attempt), types.WarmupData{
CSRF: csrf,
RoleName: roleName,
Qualified: true,
CreatedAt: time.Now(),
})
logger.LogAccountEvent("预热", "success", attempt.Account.Name, "预热完成: 账号=%s 商品=%s 耗时=%s", attempt.Account.Name, attempt.Product.CommodityName, utils.RoundDuration(time.Since(start)))
@@ -782,6 +836,23 @@ func (s *Scheduler) warmupAttempts(attempts []types.ExchangeAttempt) {
logger.LogEvent("预热", "info", "预热任务结束")
}
func uniqueWarmupAttempts(attempts []types.ExchangeAttempt) []types.ExchangeAttempt {
if len(attempts) <= 1 {
return attempts
}
seen := make(map[string]struct{}, len(attempts))
out := make([]types.ExchangeAttempt, 0, len(attempts))
for _, attempt := range attempts {
key := warmupKey(attempt)
if _, ok := seen[key]; ok {
continue
}
seen[key] = struct{}{}
out = append(out, attempt)
}
return out
}
func (s *Scheduler) calibrateDouyuTime(label string) types.TimeCalibration {
calibration, err := s.client.BestServerTimeCalibration(timeCalibrateRounds, timeCalibrateWait)
if err != nil {
@@ -939,9 +1010,8 @@ func (s *Scheduler) runAttemptFast(attempt types.ExchangeAttempt) types.Exchange
})
result.RoleName = cache.RoleName
errMsg := err.Error()
// 检查是否是 CSRF 相关错误，如果是则回退到标准模式
if strings.Contains(errMsg, "csrf") || strings.Contains(errMsg, "CSRF") || strings.Contains(errMsg, "token") {
logger.LogAccountEvent("抢购", "warn", attempt.Account.Name, "激进模式回退标准: 账号=%s 原因=CSRF错误", attempt.Account.Name)
if shouldFallbackStandard(errMsg) {
logger.LogAccountEvent("抢购", "warn", attempt.Account.Name, "激进模式回退标准: 账号=%s 原因=%s", attempt.Account.Name, utils.TrimForLog(errMsg, 80))
s.setAttemptCache(attempt.Account.Name, nil) // 清空缓存
return s.runAttempt(attempt)
}
@@ -978,6 +1048,26 @@ func (s *Scheduler) runAttemptFast(attempt types.ExchangeAttempt) types.Exchange
return result
}
func shouldFallbackStandard(errMsg string) bool {
lower := strings.ToLower(errMsg)
switch {
case strings.Contains(lower, "csrf"):
return true
case strings.Contains(lower, "token"):
return true
case strings.Contains(lower, "preexchange"):
return true
case strings.Contains(lower, "limit"):
return true
case strings.Contains(errMsg, "预兑换"):
return true
case strings.Contains(errMsg, "限制检查"):
return true
default:
return false
}
}
// ScheduleRun 注册一次定时抢购。
func (s *Scheduler) ScheduleRun(req types.SeckillRequest) (map[string]any, error) {
if len(req.AccountProducts) == 0 {
@@ -1004,13 +1094,12 @@ func (s *Scheduler) ScheduleRun(req types.SeckillRequest) (map[string]any, error
}
ap := utils.CloneStringMap(req.AccountProducts)
attempts := s.buildAttempts(ap)
s.clearWarmups()
s.scheduleMu.Lock()
s.stopSchedulesLocked()
s.scheduleMu.Unlock()
s.scheduleTimers(ap, attempts, targetTime, triggerTime, calibration, false)
s.scheduleTimers(ap, targetTime, triggerTime, calibration, false)
resyncRemain := triggerTime.Sub(serverTime) - resyncBeforeTrigger
if resyncRemain < 0 {
@@ -1020,7 +1109,7 @@ 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, attempts, targetTime, triggerTime, newCalibration, true)
s.scheduleTimers(ap, targetTime, triggerTime, newCalibration, true)
})
s.scheduleMu.Unlock()
@@ -1054,7 +1143,7 @@ func (s *Scheduler) ScheduleRun(req types.SeckillRequest) (map[string]any, error
}, nil
}
func (s *Scheduler) scheduleTimers(accountProducts map[string]string, attempts []types.ExchangeAttempt, targetTime, triggerTime time.Time, calibration types.TimeCalibration, resync bool) {
func (s *Scheduler) scheduleTimers(accountProducts map[string]string, targetTime, triggerTime time.Time, calibration types.TimeCalibration, resync bool) {
serverTime := calibration.EstimatedServerNow()
remain := triggerTime.Sub(serverTime)
if remain < 0 {
@@ -1069,6 +1158,8 @@ func (s *Scheduler) scheduleTimers(accountProducts map[string]string, attempts [
defer s.scheduleMu.Unlock()
s.replaceTimerLocked(&s.warmupSchedule, warmupRemain, func() {
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() {
其他服务器日志/douyu_20260626_155547.log
+13539
View File
File diff suppressed because it is too large Load Diff
其他服务器日志/douyu_20260626_165111.log
+15366
View File
File diff suppressed because it is too large Load Diff