Go语言中实现大规模延迟任务的磁盘持久化队列

type MyStruct struct {
    // 包含任务所需的数据
    ID string
    Payload string
}

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

// 原始的同步延迟任务处理
func IncomingJobSync(data MyStruct) {
    doSomething(&data, 1)
    time.Sleep(5 * time.Minute)
    doSomething(&data, 2)
    time.Sleep(5 * time.Minute)
    doSomething(&data, 3)
    time.Sleep(50 * time.Minute)
    doSomething(&data, 4)
}

// 使用goroutine启动
// go IncomingJobSync(data)

// 使用time.AfterFunc的异步延迟任务处理
func IncomingJobAsync(data MyStruct) {
    doSomething(&data, 1)
    time.AfterFunc(5 * time.Minute, func() {
        doSomething(&data, 2)
        time.AfterFunc(5 * time.Minute, func() {
            doSomething(&data, 3)
            time.AfterFunc(50 * time.Minute, func() {
                doSomething(&data, 4)
            })
        })
    })
}

package main

import (
    "encoding/gob"
    "fmt"
    "log"
    "os"
    "path/filepath"
    "strconv"
    "time"

    "github.com/cznic/kv" // 假设已安装：go get github.com/cznic/kv
)

// MyStruct 任务数据结构
type MyStruct struct {
    ID      string
    Payload string
    Step    int
}

// openKVDB 打开或创建一个kv数据库
func openKVDB(path string) (*kv.DB, error) {
    opts := &kv.Options{}
    return kv.Open(path, opts)
}

// serializeMyStruct 将MyStruct序列化为字节数组
func serializeMyStruct(data MyStruct) ([]byte, error) {
    var buf []byte
    enc := gob.NewEncoder(nil) // 创建一个gob编码器
    // 为了避免直接写入os.Stdout，我们需要一个bytes.Buffer
    // 但kv.Set的value是[]byte，所以直接编码到[]byte更方便
    // 实际应用中，可以使用bytes.Buffer
    // 这里简化为直接返回错误，因为gob.NewEncoder(nil)不支持直接编码到[]byte
    // 正确的做法是：
    // var b bytes.Buffer
    // enc := gob.NewEncoder(&b)
    // err := enc.Encode(data)
    // return b.Bytes(), err
    // 鉴于示例的简洁性，这里直接返回一个模拟的序列化结果
    return []byte(fmt.Sprintf("%s|%s|%d", data.ID, data.Payload, data.Step)), nil // 简化示例，实际应使用gob等
}

// deserializeMyStruct 从字节数组反序列化为MyStruct
func deserializeMyStruct(b []byte) (MyStruct, error) {
    var data MyStruct
    // 简化示例，实际应使用gob等
    parts := string(b)
    var id, payload string
    var step int
    _, err := fmt.Sscanf(parts, "%s|%s|%d", &id, &payload, &step)
    if err != nil {
        return data, err
    }
    data.ID = id
    data.Payload = payload
    data.Step = step
    return data, nil
}

// generateKey 生成基于时间戳和ID的键
func generateKey(scheduledTime time.Time, taskID string) []byte {
    // 使用Unix Nano时间戳确保唯一性和排序
    return []byte(fmt.Sprintf("%d_%s", scheduledTime.UnixNano(), taskID))
}

// StoreTask 将任务存储到磁盘队列
func StoreTask(db *kv.DB, data MyStruct, scheduledTime time.Time) error {
    key := generateKey(scheduledTime, data.ID)
    value, err := serializeMyStruct(data)
    if err != nil {
        return fmt.Errorf("序列化任务失败: %w", err)
    }

    // kv.Values are limited to 64k. 如果MyStruct很大，需要考虑分片存储。
    if len(value) > 64*1024 {
        return fmt.Errorf("任务数据过大 (超过64KB限制)")
    }

    return db.Set(key, value)
}

// PollTasks 轮询并获取到期任务
func PollTasks(db *kv.DB, currentTime time.Time) ([]MyStruct, error) {
    var readyTasks []MyStruct
    // 创建一个上限键，用于查询所有在currentTime之前或等于currentTime的键
    upperBoundKey := generateKey(currentTime, "zzz") // "zzz"确保所有相同时间戳的ID都被包含

    enum, err := db.Seek(nil) // 从数据库的第一个键开始
    if err != nil {
        return nil, fmt.Errorf("kv.Seek 失败: %w", err)
    }
    defer enum.Close()

    for {
        k, v, err := enum.Next()
        if err == kv.ErrDone {
            break // 没有更多键了
        }
        if err != nil {
            return nil, fmt.Errorf("遍历键失败: %w", err)
        }

        if string(k) > string(upperBoundKey) {
            break // 超过了当前时间，停止轮询
        }

        task, err := deserializeMyStruct(v)
        if err != nil {
            log.Printf("反序列化任务失败，跳过: %v", err)
            continue
        }
        readyTasks = append(readyTasks, task)

        // 从数据库中删除已处理的任务
        if err := db.Delete(k); err != nil {
            log.Printf("删除任务 %s 失败: %v", string(k), err)
        }
    }
    return readyTasks, nil
}

func main() {
    dbPath := filepath.Join(os.TempDir(), "delayed_queue.kv")
    db, err := openKVDB(dbPath)
    if err != nil {
        log.Fatalf("打开数据库失败: %v", err)
    }
    defer db.Close()
    defer os.RemoveAll(dbPath) // 清理临时数据库文件

    // 模拟任务入队
    task1 := MyStruct{ID: "jobA", Payload: "data for A", Step: 1}
    task2 := MyStruct{ID: "jobB", Payload: "data for B", Step: 1}
    task3 := MyStruct{ID: "jobC", Payload: "data for C", Step: 1}

    // 任务A 5秒后执行
    StoreTask(db, task1, time.Now().Add(5*time.Second))
    // 任务B 1秒后执行
    StoreTask(db, task2, time.Now().Add(1*time.Second))
    // 任务C 10秒后执行
    StoreTask(db, task3, time.Now().Add(10*time.Second))

    fmt.Println("任务已入队，开始轮询...")

    // 模拟轮询循环
    for i := 0; i < 15; i++ { // 持续轮询15秒
        time.Sleep(1 * time.Second)
        fmt.Printf("当前时间: %s, 正在轮询...\n", time.Now().Format("15:04:05"))
        tasks, err := PollTasks(db, time.Now())
        if err != nil {
            log.Printf("轮询任务失败: %v", err)
            continue
        }

        if len(tasks) > 0 {
            fmt.Printf("发现 %d 个到期任务:\n", len(tasks))
            for _, task := range tasks {
                fmt.Printf("  - 处理任务: ID=%s, Payload=%s, Step=%d\n", task.ID, task.Payload, task.Step)
                // 模拟进一步的延迟处理
                go func(t MyStruct) {
                    doSomething(&t, t.Step+1)
                    time.AfterFunc(5*time.Second, func() {
                        doSomething(&t, t.Step+2)
                    })
                }(task)
            }
        } else {
            fmt.Println("没有到期任务。")
        }
    }

    fmt.Println("轮询结束。")
}