Agent第二天：如何解决长周期任务的状态管理与恢复难题

1次阅读

没有评论

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

分布式系统中的 Agent 常需处理 ETL 流水线、模型训练等长周期任务。当遭遇节点重启或网络分区时，传统方案存在两类典型问题：

状态同步缺口：内存中的执行进度未持久化，导致任务需从头开始
数据一致性风险：重复执行可能引发数据重复插入或部分更新

某电商公司的订单分析 Agent 曾因未处理状态恢复，在每日凌晨的集群维护时段损失 3.7 小时计算资源。

优点：实现简单，直接利用现有数据库事务
缺点：高频写入压力大，如 MySQL 在每秒千级状态更新时出现连接池耗尽

优点：完整记录状态变更历史，便于审计
缺点：事件回放消耗计算资源，如 Apache Kafka 事件日志需额外处理压缩

优点：平衡资源消耗与恢复效率，Kubernetes 的 Pod 重启策略即采用此方案
缺点：需设计合理的检查点间隔，过密影响性能，过疏增加恢复成本

class TaskStateMachine:
    STATES = ['PENDING', 'RUNNING', 'PAUSED', 'COMPLETED', 'FAILED']

    def __init__(self):
        self.current_state = 'PENDING'
        self.checkpoint_file = '/tmp/task_checkpoint.json'

    def transition(self, new_state):
        if new_state not in self.STATES:
            raise ValueError(f"Invalid state: {new_state}")
        # 写入检查点前先序列化
        checkpoint = {
            "state": new_state,
            "timestamp": int(time.time())
        }
        with open(self.checkpoint_file, 'w') as f:
            json.dump(checkpoint, f)
        self.current_state = new_state

通过分片 ID+ 版本号确保操作可重复执行：

func ProcessShard(shardID string, version int64) error {
    // 检查是否已处理过该版本
    if store.GetVersion(shardID) >= version {return nil // 幂等返回}

    // 实际处理逻辑
    err := realBusinessLogic(shardID)
    if err != nil {return fmt.Errorf("process failed: %v", err)
    }

    // 更新版本号
    store.SetVersion(shardID, version)
    return nil
}

Go 语言实现带错误重试的检查点存储：

func SaveCheckpoint(cp *Checkpoint) error {
    maxRetries := 3
    for i := 0; i < maxRetries; i++ {data, err := json.Marshal(cp)
        if err != nil {continue}

        // 使用临时文件避免写入中途崩溃
        tmpPath := fmt.Sprintf("%s.tmp", cp.FilePath)
        if err := os.WriteFile(tmpPath, data, 0644); err != nil {time.Sleep(time.Second * time.Duration(i+1))
            continue
        }

        // 原子性重命名
        if err := os.Rename(tmpPath, cp.FilePath); err == nil {return nil}
    }
    return fmt.Errorf("failed after %d retries", maxRetries)
}

压缩策略：对 JSON 检查点使用 zstd 压缩，实测可减少 70% 存储空间
加密方案：采用 AWS KMS 信封加密，密钥轮换时不需重新加密全部数据

为每个分片设置独立的锁路径
使用 Consul 的分布式锁实现跨节点互斥
锁超时时间设置为平均处理时间的 3 倍

错误做法：

# 整个集群共用同一个 Redis 锁
lock = redis_lock.Lock(redis_client, "global_task_lock")

正确做法：

# 按任务维度加锁
lock_key = f"task_lock:{task_id}"
lock = redis_lock.Lock(redis_client, lock_key)

建议采用 Protobuf 定义状态结构，通过 reserved 字段保留旧版标识：

message TaskState {
    reserved 2, 5 to 10;
    string task_id = 1;
    int64 processed_items = 3;
    // 新增字段时使用新编号
    map<string, string> metadata = 4;
}

在 AWS Lambda 等无服务器环境中需注意：