timewheel

⏱️ Adaptive Hybrid Timer Scheduler (AHTS)

自适应混合定时调度器 —— 一种面向高并发在线服务的高性能定时任务调度方案。

传统时间轮的核心矛盾是"一种数据结构试图满足所有时间尺度的需求"。AHTS 通过分层异构架构，让每种数据结构只做它最擅长的事情。

✨ 核心特性

• O(1) 插入与取消 — 近期任务通过稀疏时间轮 + 反向索引实现常数时间操作
• 真正的 Tickless — 无任务时零 CPU 空转，精确睡眠到下一个到期时间点
• 漏 tick 补偿 — GC 停顿或系统调度延迟不会导致任务丢失
• 分层异构架构 — 近期任务使用稀疏时间轮，远期任务使用跳表，各取最优
• 惰性迁移 — 远期任务在接近到期时才迁移到近期轮，避免无效计算
• 并发安全 — 通过 race detector 验证，所有操作线程安全
• 周期任务支持 — 内置周期任务自动重新调度，每次执行创建独立实例避免并发问题
• 零依赖 — 纯标准库实现，无第三方依赖

🏗️ 架构

┌──────────────────────────────────────────────────────────────┐
│               Adaptive Hybrid Timer Scheduler                │
│                                                              │
│   ┌──────────────────────┐    ┌──────────────────────┐       │
│   │   Near-term Region   │    │   Far-term Region    │       │
│   │   [0, threshold)     │    │   [threshold, +∞)    │       │
│   │                      │    │                      │       │
│   │  Sparse Timing Wheel │◄───│     Skip List        │       │
│   │  • HashMap<tick,      │惰性│  • Sorted by time    │       │
│   │    HashMap<id,task>> │迁移│  • O(log n) ops      │       │
│   │  • Reverse index     │    │  • taskMap for O(1)  │       │
│   │    taskID → tick     │    │    lookup by ID      │       │
│   │  • Min-heap for O(1) │    │                      │       │
│   │    next-tick lookup  │    │                      │       │
│   └──────────┬───────────┘    └──────────────────────┘       │
│              │                                               │
│              ▼                                               │
│   ┌───────────────────────────┐                              │
│   │   Tickless Wake Engine    │                              │
│   │   • Precise sleep         │                              │
│   │   • Zero idle CPU         │                              │
│   │   • Interruptible by new  │                              │
│   │     tasks via CAS + chan  │                              │
│   └───────────────────────────┘                              │
└──────────────────────────────────────────────────────────────┘

运作流程

                    ┌─────────────┐
                    │  插入任务     │
                    └──────┬──────┘
                           │
                    delay < threshold?
                     ┌─────┴─────┐
                     │ Yes       │ No
                     ▼           ▼
              ┌────────────┐ ┌──────────┐
              │ 稀疏时间轮   │ │  跳表     │
              │ O(1) 插入   │ │ O(logn)  │
              └─────┬──────┘ └────┬─────┘
                    │             │
                    │    惰性迁移   │
                    │◄────────────┘
                    │   (接近到期时)
                    ▼
              ┌────────────┐
              │ Tickless   │
              │ 精确唤醒    │──→ 执行到期任务
              └────────────┘

📦 安装

go get github.com/atlaschan0010/timewheel

要求： Go 1.22+

🚀 快速开始

package main

import (
    "fmt"
    "time"
    "github.com/atlaschan0010/timewheel"
)

func main() {
    // 创建调度器（使用默认配置）
    scheduler := timewheel.New(timewheel.DefaultConfig())

    // 启动调度器
    scheduler.Start()
    defer scheduler.Stop()

    // 调度一个延迟任务
    scheduler.Schedule(5*time.Second, func() {
        fmt.Println("Task executed!")
    })

    // 调度可取消的任务
    taskID := scheduler.Schedule(10*time.Second, func() {
        fmt.Println("This won't print")
    })
    scheduler.Cancel(taskID) // O(1) 取消

    // 调度周期任务
    tickerID := scheduler.SchedulePeriodic(1*time.Second, func() {
        fmt.Println("Periodic task!")
    })

    time.Sleep(5 * time.Second)
    scheduler.Cancel(tickerID) // 停止周期任务

    time.Sleep(1 * time.Second)
}

⚙️ 配置

cfg := timewheel.Config{
    TickInterval:       1 * time.Millisecond,  // 近期轮精度
    NearTermThreshold:  60 * time.Second,       // 近期/远期分界线
    MigrationBatchSize: 128,                    // 每次迁移的最大任务数
    LazyDeleteEnabled:  true,                   // 启用惰性删除
}
scheduler := timewheel.New(cfg)

调优建议

📖 API

调度任务

// 延迟任务（自动生成 ID）
taskID := scheduler.Schedule(delay, func() { ... })

// 带指定 ID 的延迟任务（相同 ID 会替换旧任务）
taskID := scheduler.ScheduleWithID("my-task", delay, func() { ... })

// 周期任务
tickerID := scheduler.SchedulePeriodic(interval, func() { ... })

// 带指定 ID 的周期任务
tickerID := scheduler.SchedulePeriodicWithID("my-ticker", interval, func() { ... })

取消任务

// 取消任务（近期任务 O(1)，远期任务 O(log n)）
cancelled := scheduler.Cancel(taskID)

任务替换

// 相同 ID 调度会自动取消旧任务
scheduler.ScheduleWithID("heartbeat", 30*time.Second, handler1)
// ... 稍后 ...
scheduler.ScheduleWithID("heartbeat", 30*time.Second, handler2) // handler1 被取消

统计信息

stats := scheduler.Stats()
fmt.Printf("Executed:   %d\n", stats.TasksExecuted)
fmt.Printf("Cancelled:  %d\n", stats.TasksCancelled)
fmt.Printf("Migrated:   %d\n", stats.TasksMigrated)
fmt.Printf("Near-term:  %d\n", stats.NearTermTasks)
fmt.Printf("Far-term:   %d\n", stats.FarTermTasks)

生命周期

scheduler.Start() // 启动调度引擎（幂等，多次调用安全）
scheduler.Stop()  // 停止调度引擎，等待当前执行中的回调完成

📊 性能

测试环境： Apple M3 Max, Go 1.25, macOS

BenchmarkSchedule-16              2,184,120       552.2 ns/op     253 B/op    4 allocs/op
BenchmarkCancel-16                6,472,408       238.6 ns/op       0 B/op    0 allocs/op
BenchmarkScheduleAndCancel-16     2,344,303       520.6 ns/op     388 B/op    7 allocs/op

对比传统方案

🔧 内部实现细节

稀疏时间轮 (SparseWheel)

存储结构:
  buckets:   map[tick]map[taskID]*Task    // 正向：tick → 任务集合
  taskIndex: map[taskID]tick              // 反向：任务 → 所在 tick
  tickHeap:  MinHeap<tick>                // 最小堆：O(1) 获取最早 tick

Insert(tick, task):   O(1) — HashMap 插入 + 堆 push
Delete(taskID):       O(1) — 反向索引定位 tick → HashMap 删除
PopTicksUpTo(max):    O(k) — 从堆顶连续弹出，k 为到期任务数
PeekNextTick():       O(1) — 堆顶元素

跳表 (SkipList)

存储结构:
  sorted nodes:  按 ExpireAt 排序的跳表节点
  taskMap:       map[taskID]*SkipListNode   // O(1) 按 ID 定位节点

Insert(task):        O(log n)
DeleteByID(taskID):  O(log n) — 先 O(1) 定位节点，再 O(log n) 更新指针
PeekMin():           O(1)    — 头节点的下一个
PopMin():            O(1) 摊还

Tickless 引擎 (TicklessEngine)

主循环:
  1. nextWake = min(nearTerm.PeekNextTick, farTerm.PeekMin)
  2. if nextWake == 0: 阻塞等待新任务（WaitForTask）
  3. SleepUntil(nextWake) — 可被新任务 CAS + channel 打断
  4. advance() — PopTicksUpTo(currentTick) + 惰性迁移

CAS 唤醒机制:
  NotifyIfEarlier(expireAt):
    for {
      current = atomicLoad()
      if expireAt < current:
        if CAS(current, expireAt): Wake(); return
        continue  // CAS 失败，重试（解决 ABA 问题）
      return
    }

🎯 适用场景

✅ 最佳匹配

⚠️ 非最佳场景

🧪 测试

# 运行所有测试
go test -v ./...

# 运行 race detector
go test -race ./...

# 运行 benchmark
go test -bench=. -benchmem

📄 许可证

Apache-2.0 license