Agent开发实例：从零构建高可靠任务调度系统的实战指南

1次阅读

没有评论

共计 3294 个字符，预计需要花费 9 分钟才能阅读完成。

在分布式系统中，任务调度是一个常见但复杂的问题。传统的解决方案如 Crontab 或 Kubernetes Job 在面对高并发、高可靠需求时，往往显得力不从心。

任务堆积 ：当任务执行速度跟不上产生速度时，会导致队列积压，最终可能引发系统崩溃。
节点故障 ：单个节点宕机可能导致任务丢失或重复执行。
状态一致性 ：在分布式环境下，如何保证任务状态的一致性是个难题。

传统方案的局限性：

Crontab 缺乏分布式支持，无法处理节点故障转移
Kubernetes Job 虽然支持分布式，但缺乏细粒度的任务控制和状态管理

我们采用多级队列 + 状态机的核心设计模式：

多级队列 ：
高优先级队列：处理实时性要求高的任务
普通队列：处理常规任务
延迟队列：处理定时任务
状态机设计 ：
任务状态包括：Pending、Running、Success、Failed、Timeout
状态转换严格遵循预设规则

// 获取分布式锁
func (a *Agent) acquireLock(key string, ttl int) (bool, error) {resp, err := a.etcdClient.Grant(context.Background(), int64(ttl))
    if err != nil {return false, err}

    txn := a.etcdClient.Txn(context.Background())
    txn.If(clientv3.Compare(clientv3.CreateRevision(key), "=", 0)).
        Then(clientv3.OpPut(key, "locked", clientv3.WithLease(resp.ID))).
        Else(clientv3.OpGet(key))

    txnResp, err := txn.Commit()
    if err != nil {return false, err}

    return txnResp.Succeeded, nil
}

基于一致性哈希算法分配任务
动态调整分片大小以适应节点性能差异

每个 Agent 定期向 etcd 写入心跳信息
主节点监控从节点心跳
超时未收到心跳的节点会被标记为故障
自动重新分配故障节点上的任务

// PriorityQueue 实现带优先级的任务队列
type PriorityQueue struct {tasks []*Task
    mu    sync.Mutex
}

// Push 添加任务到队列
func (pq *PriorityQueue) Push(task *Task) {pq.mu.Lock()
    defer pq.mu.Unlock()

    heap.Push(pq, task)
    log.Printf("Task %s added to queue with priority %d", task.ID, task.Priority)
}

// Pop 从队列获取最高优先级任务
func (pq *PriorityQueue) Pop() *Task {pq.mu.Lock()
    defer pq.mu.Unlock()

    if len(pq.tasks) == 0 {return nil}

    task := heap.Pop(pq).(*Task)
    log.Printf("Task %s popped from queue", task.ID)
    return task
}

// StateMachine 处理任务状态转换
type StateMachine struct {
    currentState TaskState
    transitions  map[TaskState][]TaskState}

// Transition 执行状态转换
func (sm *StateMachine) Transition(newState TaskState) error {validTransitions, ok := sm.transitions[sm.currentState]
    if !ok {return fmt.Errorf("invalid current state: %s", sm.currentState)
    }

    for _, validState := range validTransitions {
        if validState == newState {
            sm.currentState = newState
            log.Printf("State transition: %s -> %s", sm.currentState, newState)
            return nil
        }
    }

    return fmt.Errorf("invalid transition: %s -> %s", sm.currentState, newState)
}

// 定义 Prometheus 指标
var (
    tasksProcessed = prometheus.NewCounterVec(
        prometheus.CounterOpts{
            Name: "agent_tasks_processed_total",
            Help: "Total number of processed tasks",
        },
        []string{"status"},
    )

    taskDuration = prometheus.NewHistogram(
        prometheus.HistogramOpts{
            Name:    "agent_task_duration_seconds",
            Help:    "Task processing duration in seconds",
            Buckets: prometheus.DefBuckets,
        },
    )
)

// 在任务处理函数中埋点
func processTask(task *Task) {startTime := time.Now()
    defer func() {duration := time.Since(startTime).Seconds()
        taskDuration.Observe(duration)
    }()

    // 任务处理逻辑...

    tasksProcessed.WithLabelValues("success").Inc()}

我们在不同压力下测试了系统性能：

并发任务数	平均延迟 (ms)	吞吐量 (task/s)
100	23	4200
1000	56	17800
10000	132	75600

使用 gVisor 实现任务执行沙箱隔离
限制任务可访问的系统资源

// 设置任务资源限制
func setResourceLimits() {
    // CPU 限制
    cpuQuota := int64(100000) // 0.1 CPU 核心
    if err := cgroups.WriteCgroupProc("/sys/fs/cgroup/cpu/agent/tasks", "cpu.cfs_quota_us", strconv.FormatInt(cpuQuota, 10)); err != nil {log.Printf("Failed to set CPU limit: %v", err)
    }

    // 内存限制
    memLimit := "100M"
    if err := cgroups.WriteCgroupProc("/sys/fs/cgroup/memory/agent/tasks", "memory.limit_in_bytes", memLimit); err != nil {log.Printf("Failed to set memory limit: %v", err)
    }
}