Go语言中基于磁盘的延迟任务队列实现

type MyStruct struct {
    ID    string
    Value int
    // ... 其他数据字段
}

func dosomething(data *MyStruct, stage int) {
    // 模拟对数据执行操作
    // fmt.Printf("Processing %s at stage %d\n", data.ID, stage)
}

func IncomingJob(data MyStruct) {
    // 立即执行
    dosomething(&data, 1)

    // 5分钟后执行
    time.AfterFunc(5*time.Minute, func() {
        dosomething(&data, 2)

        // 10分钟后执行
        time.AfterFunc(5*time.Minute, func() {
            dosomething(&data, 3)
        })

        // 60分钟后执行
        time.AfterFunc(50*time.Minute, func() {
            dosomething(&data, 4)
        })
    })
}

import (
    "encoding/json"
    "time"
)

type DelayedJob struct {
    ExecuteAt time.Time // 任务计划执行时间
    Data      MyStruct  // 实际的任务数据
    Stage     int       // 任务执行阶段
}

// 序列化任务数据
func (dj *DelayedJob) MarshalBinary() ([]byte, error) {
    return json.Marshal(dj)
}

// 反序列化任务数据
func (dj *DelayedJob) UnmarshalBinary(data []byte) error {
    return json.Unmarshal(data, dj)
}

import (
    "fmt"
    "strconv"
    "time"
    "github.com/cznic/kv" // 假设使用cznic/kv作为示例
)

// SaveJobToDisk 将延迟任务保存到磁盘
func SaveJobToDisk(db *kv.DB, job DelayedJob) error {
    // 使用时间戳和纳秒作为键，确保唯一性和顺序性
    key := []byte(fmt.Sprintf("%d_%d", job.ExecuteAt.UnixNano(), time.Now().Nanosecond()))
    value, err := job.MarshalBinary()
    if err != nil {
        return fmt.Errorf("failed to marshal job: %w", err)
    }
    return db.Set(key, value)
}

// PollAndExecuteJobs 轮询数据库并执行到期的任务
func PollAndExecuteJobs(db *kv.DB, interval time.Duration) {
    ticker := time.NewTicker(interval)
    defer ticker.Stop()

    for range ticker.C {
        now := time.Now()
        // 构建一个上限键，用于查询所有当前或之前到期的任务
        maxKey := []byte(fmt.Sprintf("%d_", now.UnixNano()))

        enum, _, err := db.Seek(nil) // 从头开始枚举
        if err != nil {
            fmt.Printf("Error seeking DB: %v\n", err)
            continue
        }

        var keysToDelete [][]byte
        for {
            k, v, err := enum.Next()
            if err == kv.ErrDone {
                break
            }
            if err != nil {
                fmt.Printf("Error getting next item: %v\n", err)
                break
            }

            // 解析键中的时间戳
            keyStr := string(k)
            parts := splitKey(keyStr) // 假设有一个函数可以安全地分割键
            if len(parts) < 1 {
                continue
            }
            jobTimeNano, err := strconv.ParseInt(parts[0], 10, 64)
            if err != nil {
                fmt.Printf("Error parsing timestamp from key %s: %v\n", keyStr, err)
                continue
            }

            if time.Unix(0, jobTimeNano).Before(now) || time.Unix(0, jobTimeNano).Equal(now) {
                var job DelayedJob
                if err := job.UnmarshalBinary(v); err != nil {
                    fmt.Printf("Error unmarshaling job: %v\n", err)
                    // 即使反序列化失败，也可能需要删除，以免阻塞队列
                    keysToDelete = append(keysToDelete, k)
                    continue
                }

                // 执行任务
                fmt.Printf("Executing job ID: %s, Stage: %d at %s\n", job.Data.ID, job.Stage, now.Format(time.RFC3339))
                dosomething(&job.Data, job.Stage)

                // 标记为待删除
                keysToDelete = append(keysToDelete, k)
            } else {
                // 任务未到期，由于键是按时间排序的，后续任务也未到期
                break
            }
        }

        // 批量删除已处理的任务
        for _, k := range keysToDelete {
            if err := db.Delete(k); err != nil {
                fmt.Printf("Error deleting key %s: %v\n", string(k), err)
            }
        }
    }
}

// 辅助函数：安全地分割键
func splitKey(key string) []string {
    // 假设键格式为 "timestamp_sequence"
    for i := 0; i < len(key); i++ {
        if key[i] == '_' {
            return []string{key[:i], key[i+1:]}
        }
    }
    return []string{key}
}

// 示例：模拟原始 IncomingJob 逻辑，但将任务持久化
func ScheduleIncomingJob(db *kv.DB, data MyStruct) {
    // 立即执行第一阶段
    dosomething(&data, 1)

    // 调度后续阶段
    now := time.Now()
    _ = SaveJobToDisk(db, DelayedJob{ExecuteAt: now.Add(5 * time.Minute), Data: data, Stage: 2})
    _ = SaveJobToDisk(db, DelayedJob{ExecuteAt: now.Add(10 * time.Minute), Data: data, Stage: 3})
    _ = SaveJobToDisk(db, DelayedJob{ExecuteAt: now.Add(60 * time.Minute), Data: data, Stage: 4})
}

func main() {
    // 初始化 kv 数据库
    // 注意：cznic/kv 可能需要特定的文件路径和配置
    // 这是一个概念性示例，实际使用请参考 cznic/kv 文档
    // db, err := kv.Open("my_disk_queue.kv", &kv.Options{})
    // if err != nil {
    //  log.Fatalf("Failed to open kv DB: %v", err)
    // }
    // defer db.Close()

    // 模拟一个简单的内存 map 作为 kv.DB 的替代，仅用于演示逻辑
    // 实际生产环境请使用真正的磁盘数据库
    type mockDB struct {
        data map[string][]byte
    }
    // ... (mockDB 的实现和 kv.DB 接口对齐，这里省略具体细节)
    // 假设我们有一个 db 实例
    var db *kv.DB // 实际应为初始化的 kv.DB 实例

    // 启动轮询器
    go PollAndExecuteJobs(db, 1*time.Second)

    // 模拟接收新任务
    for i := 0; i < 1000; i++ {
        data := MyStruct{ID: fmt.Sprintf("job-%d", i), Value: i}
        ScheduleIncomingJob(db, data)
    }

    // 保持主 goroutine 运行，以便后台任务继续
    select {}
}