Agent 集群化部署架构多实例下的任务分配与状态同步一、单实例 Agent 的单点瓶颈与状态孤岛生产环境中第一个 Agent 上线后一切顺利——日处理 500 个任务P99 延迟 3 秒。第二周用户量翻倍同一台机器上的 Agent 开始排队任务超时率从 0.1% 飙升到 15%。决定横向扩容时才发现Agent 的任务状态全在内存里扩容后新实例不知道老实例有哪些进行中的任务。多实例部署不是简单地把一份代码部署到多台机器。它需要解决三个核心问题任务如何分配、状态如何共享、故障如何转移。二、集群化架构设计任务分发、状态同步、故障恢复Agent 集群化的架构分为三层任务分发层接收外部请求根据负载均衡策略分配给不同 Agent 实例。可以采用一致性哈希保证同一会话的任务路由到同一实例也可以使用分布式队列如 Redis Stream、Kafka解耦。状态同步层Agent 运行过程中产生的中间状态需要持久化。轻量方案用 Redis重量方案用 etcd 或数据库。关键设计是状态的读写粒度——太粗导致冲突太细导致延迟。故障恢复层当某个 Agent 实例挂了它手里的任务需要被其他实例接管。心跳检测 任务超时重分配是标准方案。flowchart TD subgraph Ingress[入口层] API[API Gateway] end subgraph Dispatch[任务分发层] DS[分布式调度器] Q[Redis Stream 任务队列] end subgraph Agents[Agent 实例集群] A1[Agent-1: running] A2[Agent-2: running] A3[Agent-3: running] end subgraph State[状态存储层] Redis[(Redis: 任务锁会话状态)] DB[(MySQL: 任务审计日志)] end subgraph Recovery[故障恢复] HB[心跳检测] FR[孤儿任务回收] end API --|创建任务| DS DS --|PUSH| Q Q --|消费| A1 Q --|消费| A2 Q --|消费| A3 A1 -.-|心跳 每5秒| HB A2 -.-|心跳 每5秒| HB A3 -.-|心跳 每5秒| HB HB --|超时未心跳| FR FR --|重新入队| Q A1 --|写状态| Redis A2 --|写状态| Redis A3 --|写状态| Redis A1 --|审计日志| DB A2 --|审计日志| DB A3 --|审计日志| DB三、Go 实现基于 Redis 的 Agent 集群管理器package agentcluster import ( context encoding/json fmt log os sync time github.com/go-redis/redis/v8 github.com/google/uuid ) // TaskStatus 任务状态枚举 type TaskStatus string const ( TaskPending TaskStatus pending TaskProcessing TaskStatus processing TaskCompleted TaskStatus completed TaskFailed TaskStatus failed ) // AgentTask Agent 任务定义 type AgentTask struct { ID string json:id UserID string json:user_id Input string json:input Status TaskStatus json:status WorkerID string json:worker_id Result string json:result,omitempty CreatedAt time.Time json:created_at UpdatedAt time.Time json:updated_at } // AgentInstance 集群中的 Agent 实例 type AgentInstance struct { ID string json:id Hostname string json:hostname Status string json:status // active, draining, dead MaxConcurrent int json:max_concurrent CurrentLoad int json:current_load LastHeartbeat time.Time json:last_heartbeat } // ClusterManager Agent 集群管理器 type ClusterManager struct { redis *redis.Client instance *AgentInstance taskQueue string // Redis List key cancelFunc context.CancelFunc mu sync.Mutex // 配置 heartbeatInterval time.Duration orphanTimeout time.Duration // 孤儿任务回收超时 } // NewClusterManager 创建集群管理器 func NewClusterManager(rdb *redis.Client, maxConcurrent int) (*ClusterManager, error) { hostname, _ : os.Hostname() instance : AgentInstance{ ID: uuid.New().String(), Hostname: hostname, Status: active, MaxConcurrent: maxConcurrent, } cm : ClusterManager{ redis: rdb, instance: instance, taskQueue: agent:task_queue, heartbeatInterval: 5 * time.Second, orphanTimeout: 30 * time.Second, } // 启动心跳 go cm.startHeartbeat() // 启动孤儿任务回收 go cm.startOrphanReaper() return cm, nil } // EnqueueTask 将任务加入分布式队列 func (cm *ClusterManager) EnqueueTask(ctx context.Context, task *AgentTask) error { task.ID uuid.New().String() task.Status TaskPending task.CreatedAt time.Now() task.UpdatedAt time.Now() data, err : json.Marshal(task) if err ! nil { return fmt.Errorf(序列化任务失败: %w, err) } // 使用 Redis Stream 实现可靠的消息队列 if err : cm.redis.XAdd(ctx, redis.XAddArgs{ Stream: cm.taskQueue, MaxLen: 10000, // 限制队列长度防止内存爆炸 Approx: true, Values: map[string]interface{}{ task: string(data), }, }).Err(); err ! nil { return fmt.Errorf(任务入队失败: %w, err) } // 同时持久化到数据库用于审计 cm.redis.Set(ctx, agent:task:task.ID, string(data), 24*time.Hour) log.Printf([CLUSTER] 任务入队: %s, task.ID) return nil } // ConsumeTasks 消费任务循环每个 Agent 实例的主循环 func (cm *ClusterManager) ConsumeTasks(ctx context.Context, handler func(*AgentTask) error) error { // 创建消费者组幂等操作 groupName : agent-workers consumerName : cm.instance.ID cm.redis.XGroupCreateMkStream(ctx, cm.taskQueue, groupName, 0) for { select { case -ctx.Done(): return ctx.Err() default: } // 负载控制当前任务数已达上限等待一段时间再消费 cm.mu.Lock() currentLoad : cm.instance.CurrentLoad cm.mu.Unlock() if currentLoad cm.instance.MaxConcurrent { time.Sleep(500 * time.Millisecond) continue } // 从消费者组读取消息 streams, err : cm.redis.XReadGroup(ctx, redis.XReadGroupArgs{ Group: groupName, Consumer: consumerName, Streams: []string{cm.taskQueue, }, Count: 1, Block: 2 * time.Second, }).Result() if err ! nil { if err redis.Nil { continue // 无消息等待 } log.Printf([CLUSTER] 读取任务失败: %v, err) continue } for _, stream : range streams { for _, msg : range stream.Messages { taskData, ok : msg.Values[task].(string) if !ok { cm.redis.XAck(ctx, cm.taskQueue, groupName, msg.ID) continue } var task AgentTask if err : json.Unmarshal([]byte(taskData), task); err ! nil { cm.redis.XAck(ctx, cm.taskQueue, groupName, msg.ID) continue } // 标记任务为处理中 task.Status TaskProcessing task.WorkerID cm.instance.ID task.UpdatedAt time.Now() cm.mu.Lock() cm.instance.CurrentLoad cm.mu.Unlock() // 异步处理任务 go func(t AgentTask, msgID string) { defer func() { cm.mu.Lock() cm.instance.CurrentLoad-- cm.mu.Unlock() cm.redis.XAck(ctx, cm.taskQueue, groupName, msgID) }() // 设置任务超时 taskCtx, cancel : context.WithTimeout(ctx, 60*time.Second) defer cancel() // 获取分布式锁防止重复处理 lockKey : fmt.Sprintf(agent:lock:%s, t.ID) locked, err : cm.redis.SetNX(taskCtx, lockKey, cm.instance.ID, 60*time.Second).Result() if err ! nil || !locked { return // 已被其他实例处理 } defer cm.redis.Del(context.Background(), lockKey) if err : handler(t); err ! nil { t.Status TaskFailed t.Result err.Error() log.Printf([CLUSTER] 任务失败 %s: %v, t.ID, err) } else { t.Status TaskCompleted log.Printf([CLUSTER] 任务完成 %s, t.ID) } t.UpdatedAt time.Now() data, _ : json.Marshal(t) cm.redis.Set(context.Background(), agent:task:t.ID, string(data), 24*time.Hour) }(task, msg.ID) } } } } // startHeartbeat 心跳维护向 Redis 报告存活状态 func (cm *ClusterManager) startHeartbeat() { ticker : time.NewTicker(cm.heartbeatInterval) defer ticker.Stop() for range ticker.C { cm.mu.Lock() cm.instance.LastHeartbeat time.Now() cm.mu.Unlock() data, _ : json.Marshal(cm.instance) cm.redis.Set(context.Background(), agent:instance:cm.instance.ID, string(data), cm.heartbeatInterval*3, // TTL 为心跳间隔的 3 倍 ) } } // startOrphanReaper 孤儿任务回收回收死实例未完成的任务 func (cm *ClusterManager) startOrphanReaper() { ticker : time.NewTicker(10 * time.Second) defer ticker.Stop() for range ticker.C { ctx : context.Background() // 扫描所有实例 keys, _ : cm.redis.Keys(ctx, agent:instance:*).Result() for _, key : range keys { data, _ : cm.redis.Get(ctx, key).Bytes() var inst AgentInstance if err : json.Unmarshal(data, inst); err ! nil { continue } // 心跳超时 → 标记为 dead if time.Since(inst.LastHeartbeat) cm.orphanTimeout { log.Printf([CLUSTER] 实例 %s 无响应开始回收任务, inst.ID) cm.redis.Del(ctx, key) // 回收该实例的进行中任务 cm.reclaimOrphanTasks(ctx, inst.ID) } } } } // reclaimOrphanTasks 回收孤儿任务重新入队 func (cm *ClusterManager) reclaimOrphanTasks(ctx context.Context, instanceID string) { pattern : agent:task:* keys, _ : cm.redis.Keys(ctx, pattern).Result() for _, key : range keys { data, _ : cm.redis.Get(ctx, key).Bytes() var task AgentTask if err : json.Unmarshal(data, task); err ! nil { continue } if task.WorkerID instanceID task.Status TaskProcessing { task.Status TaskPending task.WorkerID task.UpdatedAt time.Now() data, _ : json.Marshal(task) cm.redis.Set(ctx, key, string(data), 24*time.Hour) // 重新入队 cm.redis.XAdd(ctx, redis.XAddArgs{ Stream: cm.taskQueue, Values: map[string]interface{}{task: string(data)}, }) log.Printf([CLUSTER] 孤儿任务 %s 已重新入队, task.ID) } } }四、集群化的边界与权衡有状态 vs 无状态 Agent。如果 Agent 不需要跨请求共享上下文如简单的问答可以设计成无状态用轮询负载均衡即可。如果需要维持对话上下文如多轮对话 Agent则要用一致性哈希或会话粘滞将同一用户路由到同一实例。任务队列的可靠性选型。Redis Stream 轻量但消息持久化有限Kafka 更可靠但运维成本高。对于大多数中小规模的 Agent 集群10 实例以内Redis Stream 配合数据库双重写入已经足够。分布式锁的 TTL 设置。锁的 TTL 应该大于任务最长执行时间。如果任务可能在模型调用上阻塞 30 秒锁的 TTL 至少设为 60 秒。太短会导致锁提前释放引发重复处理太长则导致死锁后等待过久。心跳超时不能设得太敏感。网络抖动可能导致短暂心跳丢失。如果心跳超时设为 5 秒就触发回收一次 GC 暂停可能让正常实例被误判为挂掉。建议心跳间隔 5 秒超时阈值 15-30 秒。五、总结Agent 集群化的核心问题不是怎么写代码而是怎么管理状态。分布式任务队列解耦调度、Redis 存储共享状态、心跳机制保障故障恢复——这三个组件构成了 Agent 集群的基础设施。实现路径建议先上分布式队列解决多实例消费问题再加心跳和孤儿回收解决故障转移最后引入一致性哈希解决会话粘滞。每一步独立验证别一下子上全。