Agent实现代码案例：从零构建高可用的任务调度系统

1次阅读

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

传统定时任务方案（如 CronJob）在分布式系统中暴露出明显短板：

单点故障 ：调度器宕机导致全线任务停滞
状态丢失 ：内存中的任务状态无法持久化
雪崩风险 ：任务堆积引发连锁反应

Agent 模式通过以下特性解决这些问题：

自治性 ：每个 Agent 独立管理生命周期
弹性伸缩 ：根据负载动态调整实例数
故障隔离 ：单个 Agent 失效不影响整体

指标	CronJob	消息队列	Agent 模式
吞吐量	低	中	高
容错性	无	部分	强
实现复杂度	简单	中等	较高

// 任务状态类型定义
type TaskState int

const (
    STARTING TaskState = iota // 启动中
    RUNNING                   // 运行中
    ERROR                     // 错误状态
)

// 状态转换方法
func (a *Agent) transitionState(from, to TaskState) error {a.lock.Lock()
    defer a.lock.Unlock()

    if a.currentState != from {return fmt.Errorf("invalid state transition")
    }
    a.currentState = to
    return nil
}

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

    // 创建租约
    _, err = a.etcdClient.Put(context.TODO(), 
        key, 
        a.instanceID, 
        clientv3.WithLease(resp.ID))

    // 保持租约存活
    keepAlive, err := a.etcdClient.KeepAlive(context.TODO(), resp.ID)
    go func() {
        for range keepAlive {// 空循环维持心跳}
    }()
    return err
}

// 心跳检测协程
func (a *Agent) startHeartbeat() {ticker := time.NewTicker(5 * time.Second)
    defer ticker.Stop()

    for {
        select {
        case <-ticker.C:
            if err := a.reportStatus(); err != nil {a.handleError(err)
            }
        case <-a.stopChan:
            return
        }
    }
}

批处理大小	QPS	平均延迟
1	1200	50ms
10	8500	65ms
100	42000	110ms

// 任务上下文对象池
var taskCtxPool = sync.Pool{New: func() interface{} {
        return &TaskContext{createdAt: time.Now(),
        }
    },
}

func getTaskContext() *TaskContext {return taskCtxPool.Get().(*TaskContext)
}

func releaseTaskContext(ctx *TaskContext) {ctx.Reset()
    taskCtxPool.Put(ctx)
}