企业官网建设流程全解析

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 集群的基础设施。实现路径建议：先上分布式队列解决多实例消费问题，再加心跳和孤儿回收解决故障转移，最后引入一致性哈希解决会话粘滞。每一步独立验证，别一下子上全。