Go语言中基于磁盘的延迟队列实现与内存优化

func IncomingJob(data MyStruct) {
    // 立即执行
    dosomething(&data, 1)
    time.Sleep(5 * time.Minute)
    // 5分钟后执行
    dosomething(&data, 2)
    time.Sleep(5 * time.Minute)
    // 10分钟后执行
    dosomething(&data, 3)
    time.Sleep(50 * time.Minute)
    // 60分钟后执行
    dosomething(&data, 4)
}

func IncomingJob(data MyStruct) {
    // 立即执行
    dosomething(&data, 1)
    time.AfterFunc(5*time.Minute, func() {
        // 5分钟后执行
        dosomething(&data, 2)
        time.AfterFunc(5*time.Minute, func() {
            // 10分钟后执行
            dosomething(&data, 3)
            time.AfterFunc(50*time.Minute, func() {
                // 60分钟后执行
                dosomething(&data, 4)
            })
        })
    })
}

// 示例：使用cznic/kv库的伪代码
package main

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

    "github.com/cznic/kv" // 假设已安装此库
)

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

// openDB 打开或创建KV数据库
func openDB(path string) (*kv.DB, error) {
    createOpen := kv.Open
    if _, err := os.Stat(path); os.IsNotExist(err) {
        createOpen = kv.Create
    }
    return createOpen(path, &kv.Options{})
}

// serializeMyStruct 序列化MyStruct
func serializeMyStruct(data MyStruct) ([]byte, error) {
    var buf []byte
    enc := gob.NewEncoder(nil) // 创建一个编码器
    // Gob编码需要一个Writer，这里我们用一个临时的buf
    // 更实际的用法是使用bytes.Buffer
    // 这里简化为直接编码到[]byte，实际需要bytes.Buffer
    // 修正：直接使用bytes.Buffer
    var b bytes.Buffer
    enc = gob.NewEncoder(&b)
    if err := enc.Encode(data); err != nil {
        return nil, err
    }
    return b.Bytes(), nil
}

// deserializeMyStruct 反序列化MyStruct
func deserializeMyStruct(b []byte) (MyStruct, error) {
    var data MyStruct
    dec := gob.NewDecoder(bytes.NewReader(b))
    if err := dec.Decode(&data); err != nil {
        return data, err
    }
    return data, nil
}

// EnqueueTask 将任务入队
func EnqueueTask(db *kv.DB, task MyStruct, scheduledTime time.Time) error {
    serializedData, err := serializeMyStruct(task)
    if err != nil {
        return fmt.Errorf("serialize task failed: %w", err)
    }

    // 键设计: "unix_nano_timestamp" + "_" + "task_id"
    // 这样可以按时间戳排序，且同一时间戳下的任务有唯一键
    key := []byte(fmt.Sprintf("%d_%s", scheduledTime.UnixNano(), task.ID))

    return db.Set(key, serializedData)
}

// PollAndExecuteDueTasks 轮询并执行到期任务
func PollAndExecuteDueTasks(db *kv.DB, dosomething func(*MyStruct, int)) {
    t := db.NewTransaction()
    defer t.Rollback() // 确保事务回滚或提交

    // 迭代所有键，按键（时间戳）升序
    enum, err := t.SeekFirst()
    if err != nil {
        log.Printf("Error seeking first: %v", err)
        return
    }

    for {
        k, v, err := enum.Next()
        if err == kv.ErrNotFound {
            break // 没有更多任务
        }
        if err != nil {
            log.Printf("Error enumerating: %v", err)
            break
        }

        keyStr := string(k)
        parts := strings.SplitN(keyStr, "_", 2)
        if len(parts) != 2 {
            log.Printf("Invalid key format: %s", keyStr)
            continue
        }

        scheduledUnixNano, err := strconv.ParseInt(parts[0], 10, 64)
        if err != nil {
            log.Printf("Invalid timestamp in key: %s, err: %v", keyStr, err)
            continue
        }

        scheduledTime := time.Unix(0, scheduledUnixNano)
        if scheduledTime.After(time.Now()) {
            // 当前任务未到期，由于键是排序的，后续任务也未到期
            break
        }

        // 任务已到期，执行
        task, err := deserializeMyStruct(v)
        if err != nil {
            log.Printf("Error deserializing task %s: %v", keyStr, err)
            // 考虑是否删除此损坏任务或记录错误
            continue
        }

        log.Printf("Executing task ID: %s, Step: %d at %s", task.ID, task.Step, time.Now())
        dosomething(&task, task.Step)

        // 任务执行完毕，从数据库中删除
        if err := t.Delete(k); err != nil {
            log.Printf("Error deleting task %s: %v", keyStr, err)
            // 错误处理，可能需要重试或记录
        }
    }

    if err := t.Commit(); err != nil {
        log.Printf("Error committing transaction: %v", err)
    }
}