54fe363257
- P0-1: subscription_maintenance_queue 使用 RWMutex 防止 channel close/send 竞态 - P0-2: billing_service CalculateCostWithLongContext 修复被吞没的 out-range 错误 - P1-1: timing_wheel_service Schedule/ScheduleRecurring 添加 SetTimer 错误日志 - P1-2: sora_gateway_service StoreFromURLs 失败时降级使用原始 URL - P1-3: concurrency_cache 用 Pipeline 替代 Lua 脚本兼容 Redis Cluster - P1-6: sora_media_cleanup_service runCleanup 添加 nil cfg/storage 防护 Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
73 lines
1.9 KiB
Go
73 lines
1.9 KiB
Go
package service
|
|
|
|
import (
|
|
"fmt"
|
|
"log"
|
|
"sync"
|
|
"time"
|
|
|
|
"github.com/zeromicro/go-zero/core/collection"
|
|
)
|
|
|
|
var newTimingWheel = collection.NewTimingWheel
|
|
|
|
// TimingWheelService wraps go-zero's TimingWheel for task scheduling
|
|
type TimingWheelService struct {
|
|
tw *collection.TimingWheel
|
|
stopOnce sync.Once
|
|
}
|
|
|
|
// NewTimingWheelService creates a new TimingWheelService instance
|
|
func NewTimingWheelService() (*TimingWheelService, error) {
|
|
// 1 second tick, 3600 slots = supports up to 1 hour delay
|
|
// execute function: runs func() type tasks
|
|
tw, err := newTimingWheel(1*time.Second, 3600, func(key, value any) {
|
|
if fn, ok := value.(func()); ok {
|
|
fn()
|
|
}
|
|
})
|
|
if err != nil {
|
|
return nil, fmt.Errorf("创建 timing wheel 失败: %w", err)
|
|
}
|
|
return &TimingWheelService{tw: tw}, nil
|
|
}
|
|
|
|
// Start starts the timing wheel
|
|
func (s *TimingWheelService) Start() {
|
|
log.Println("[TimingWheel] Started (auto-start by go-zero)")
|
|
}
|
|
|
|
// Stop stops the timing wheel
|
|
func (s *TimingWheelService) Stop() {
|
|
s.stopOnce.Do(func() {
|
|
s.tw.Stop()
|
|
log.Println("[TimingWheel] Stopped")
|
|
})
|
|
}
|
|
|
|
// Schedule schedules a one-time task
|
|
func (s *TimingWheelService) Schedule(name string, delay time.Duration, fn func()) {
|
|
if err := s.tw.SetTimer(name, fn, delay); err != nil {
|
|
log.Printf("[TimingWheel] SetTimer failed for %q: %v", name, err)
|
|
}
|
|
}
|
|
|
|
// ScheduleRecurring schedules a recurring task
|
|
func (s *TimingWheelService) ScheduleRecurring(name string, interval time.Duration, fn func()) {
|
|
var schedule func()
|
|
schedule = func() {
|
|
fn()
|
|
if err := s.tw.SetTimer(name, schedule, interval); err != nil {
|
|
log.Printf("[TimingWheel] recurring SetTimer failed for %q: %v", name, err)
|
|
}
|
|
}
|
|
if err := s.tw.SetTimer(name, schedule, interval); err != nil {
|
|
log.Printf("[TimingWheel] initial SetTimer failed for %q: %v", name, err)
|
|
}
|
|
}
|
|
|
|
// Cancel cancels a scheduled task
|
|
func (s *TimingWheelService) Cancel(name string) {
|
|
_ = s.tw.RemoveTimer(name)
|
|
}
|