From 0e1a95a3f89141a613fd53b967c1a54a0f231b56 Mon Sep 17 00:00:00 2001 From: duorameng <2997944583@qq.com> Date: Tue, 19 May 2026 20:23:05 +0800 Subject: [PATCH] feat: implement agent custom scheduler config and queue model #101 --- agent/agent.go | 45 +++++--- internal/controllers/agent_controller.go | 89 +++++++++++----- internal/executor/scheduler.go | 18 +++- internal/models/agent.go | 74 +++++++++++--- internal/models/vo/agent_vo.go | 39 +++++-- internal/services/agent_service.go | 9 +- internal/services/tasks/executor_service.go | 24 ++++- web/src/api/index.ts | 11 +- web/src/views/agents/Agents.vue | 108 ++++++++++++++++++-- 9 files changed, 332 insertions(+), 85 deletions(-) diff --git a/agent/agent.go b/agent/agent.go index 72480a1..171c543 100644 --- a/agent/agent.go +++ b/agent/agent.go @@ -152,6 +152,7 @@ type Agent struct { wsStopCh chan struct{} // 用于停止当前 WebSocket 相关的 goroutine taskLogs map[string][]string // 记录最近的日志行，用于失败显示 logMu sync.Mutex // taskLogs 的锁 + schedulerStarted bool // 调度器是否已经启动 } func NewAgent(config *Config, configFile string) *Agent { @@ -267,9 +268,7 @@ func (a *Agent) Start() error { } logger.Infof("机器识别码: %s", a.machineID[:16]+"...") - a.scheduler.Start() - a.cronManager.Start() - + // 调度器暂不在此启动，等待 WebSocket 连接成功并获取到调度配置后再启动 go a.wsLoop() logger.Info("Agent 已启动 (时区: Asia/Shanghai, 模式: WebSocket)") @@ -279,8 +278,16 @@ func (a *Agent) Start() error { func (a *Agent) Stop() { close(a.stopCh) a.closeWS() - a.cronManager.Stop() - a.scheduler.Stop() + + a.mu.Lock() + started := a.schedulerStarted + a.schedulerStarted = false + a.mu.Unlock() + + if started { + a.cronManager.Stop() + a.scheduler.Stop() + } logger.Info("Agent 已停止") } @@ -460,16 +467,30 @@ func (a *Agent) updateSchedulerConfig(config map[string]interface{}) { newCfg.RateInterval = time.Duration(v) * time.Millisecond } } + if val, ok := config["strict_queue"]; ok { + if v, ok := val.(bool); ok { + newCfg.StrictQueue = v + } + } - // 只有当配置发生变化时才重新加载 - // 只有当配置发生变化时才重新加载 - if newCfg != currentCfg { - logger.Infof("收到调度配置更新: workers=%d, queue=%d, rate=%v", - newCfg.WorkerCount, newCfg.QueueSize, newCfg.RateInterval) + a.mu.Lock() + started := a.schedulerStarted + a.schedulerStarted = true + a.mu.Unlock() + + if !started { + logger.Infof("首次连接成功，启动调度器配置: workers=%d, queue=%d, rate=%v, strict=%t", + newCfg.WorkerCount, newCfg.QueueSize, newCfg.RateInterval, newCfg.StrictQueue) + // 用下发的最新配置加载并启动调度器与计划任务管理器 + a.scheduler.Reload(newCfg) + a.cronManager.Start() + } else if newCfg != currentCfg { + logger.Infof("收到调度配置更新: workers=%d, queue=%d, rate=%v, strict=%t", + newCfg.WorkerCount, newCfg.QueueSize, newCfg.RateInterval, newCfg.StrictQueue) a.scheduler.Reload(newCfg) } else { - logger.Infof("当前调度配置: workers=%d, queue=%d, rate=%v", - newCfg.WorkerCount, newCfg.QueueSize, newCfg.RateInterval) + logger.Infof("当前调度配置未改变: workers=%d, queue=%d, rate=%v, strict=%t", + newCfg.WorkerCount, newCfg.QueueSize, newCfg.RateInterval, newCfg.StrictQueue) } } diff --git a/internal/controllers/agent_controller.go b/internal/controllers/agent_controller.go index 951afc0..5ae3d03 100644 --- a/internal/controllers/agent_controller.go +++ b/internal/controllers/agent_controller.go @@ -45,6 +45,29 @@ func (c *AgentController) List(ctx *gin.Context) { utils.Success(ctx, vo.ToAgentVOListFromModels(agents)) } +// getActiveSchedulerConfig 获取 Agent 的实际调度配置（若为空或零值，则使用系统默认的 settings） +func (c *AgentController) getActiveSchedulerConfig(agent *models.Agent) map[string]interface{} { + workerCount := agent.SchedulerConfig.WorkerCount + queueSize := agent.SchedulerConfig.QueueSize + rateInterval := int(agent.SchedulerConfig.RateInterval / time.Millisecond) + strictQueue := agent.SchedulerConfig.StrictQueue + + // 如果未配置（WorkerCount <= 0），则使用全局系统设置 + if workerCount <= 0 { + workerCount = getIntSetting(c.settingsService, constant.SectionScheduler, constant.KeyWorkerCount, 4) + queueSize = getIntSetting(c.settingsService, constant.SectionScheduler, constant.KeyQueueSize, 100) + rateInterval = getIntSetting(c.settingsService, constant.SectionScheduler, constant.KeyRateInterval, 200) + strictQueue = false + } + + return map[string]interface{}{ + "worker_count": workerCount, + "queue_size": queueSize, + "rate_interval": rateInterval, + "strict_queue": strictQueue, + } +} + // Update 更新 Agent func (c *AgentController) Update(ctx *gin.Context) { id := ctx.Param("id") @@ -54,16 +77,17 @@ func (c *AgentController) Update(ctx *gin.Context) { } var req struct { - Name string `json:"name" binding:"required"` - Description string `json:"description"` - Enabled bool `json:"enabled"` + Name string `json:"name" binding:"required"` + Description string `json:"description"` + Enabled bool `json:"enabled"` + SchedulerConfig *vo.AgentSchedulerConfigVO `json:"scheduler_config"` } - + if err := ctx.ShouldBindJSON(&req); err != nil { utils.BadRequest(ctx, "参数错误") return } - + // 获取旧状态 oldAgent := c.agentService.GetByID(id) if oldAgent == nil { @@ -71,12 +95,21 @@ func (c *AgentController) Update(ctx *gin.Context) { return } wasEnabled := utils.DerefBool(oldAgent.Enabled, true) + + var schedulerConfig models.AgentSchedulerConfig + if req.SchedulerConfig != nil { + schedulerConfig.WorkerCount = req.SchedulerConfig.WorkerCount + schedulerConfig.QueueSize = req.SchedulerConfig.QueueSize + schedulerConfig.RateInterval = time.Duration(req.SchedulerConfig.RateInterval) * time.Millisecond + schedulerConfig.Verbose = req.SchedulerConfig.Verbose + schedulerConfig.StrictQueue = req.SchedulerConfig.StrictQueue + } - if err := c.agentService.Update(id, req.Name, req.Description, req.Enabled); err != nil { + if err := c.agentService.Update(id, req.Name, req.Description, req.Enabled, schedulerConfig); err != nil { utils.ServerError(ctx, err.Error()) return } - + // 如果启用状态发生变化，通知 Agent if wasEnabled != req.Enabled { if req.Enabled { @@ -93,7 +126,20 @@ func (c *AgentController) Update(ctx *gin.Context) { }) } } - + + // 推送最新的调度配置给 Agent (如果 Agent 在线) + if req.Enabled { + // 重新加载已更新的 Agent 信息以获取正确的 SchedulerConfig + updatedAgent := c.agentService.GetByID(id) + if updatedAgent != nil { + c.wsManager.SendToAgent(id, services.WSTypeConnected, map[string]interface{}{ + "agent_id": id, + "name": req.Name, + "scheduler_config": c.getActiveSchedulerConfig(updatedAgent), + }) + } + } + utils.SuccessMsg(ctx, "更新成功") } @@ -415,26 +461,17 @@ func (c *AgentController) WSConnect(ctx *gin.Context) { // 更新 Agent 状态 c.agentService.Heartbeat(token, ip, "", "", "", "", "") - // 获取调度配置 - workerCount := getIntSetting(c.settingsService, constant.SectionScheduler, constant.KeyWorkerCount, 4) - queueSize := getIntSetting(c.settingsService, constant.SectionScheduler, constant.KeyQueueSize, 100) - rateInterval := getIntSetting(c.settingsService, constant.SectionScheduler, constant.KeyRateInterval, 200) - - // 发送连接成功消息（包含注册状态和调度配置） + // 获取调度配置并发送连接成功消息（包含注册状态和调度配置） + schedCfg := c.getActiveSchedulerConfig(agent) c.wsManager.SendToAgent(agent.ID, services.WSTypeConnected, map[string]interface{}{ - "agent_id": agent.ID, - "name": agent.Name, - "is_new_agent": isNewAgent, - "machine_id": machineID, - "scheduler_config": map[string]interface{}{ - "worker_count": workerCount, - "queue_size": queueSize, - "rate_interval": rateInterval, - }, + "agent_id": agent.ID, + "name": agent.Name, + "is_new_agent": isNewAgent, + "machine_id": machineID, + "scheduler_config": schedCfg, }) - - logger.Infof("[AgentWS] Agent #%s 连接成功 (配置: workers=%d, queue=%d, rate=%d)", - agent.ID, workerCount, queueSize, rateInterval) + + logger.Infof("[AgentWS] Agent #%s 连接成功 (配置: %v)", agent.ID, schedCfg) // 启动读写协程 go c.wsWritePump(ac) diff --git a/internal/executor/scheduler.go b/internal/executor/scheduler.go index ceeec80..e9a72a7 100644 --- a/internal/executor/scheduler.go +++ b/internal/executor/scheduler.go @@ -37,6 +37,7 @@ type SchedulerConfig struct { QueueSize int // 队列大小 RateInterval time.Duration // 速率限制间隔 Verbose bool // 是否开启详细日志 + StrictQueue bool // 是否开启严格排队（满时拒绝执行，不降级直接执行） } // TaskType 任务类型 @@ -281,9 +282,16 @@ func (s *Scheduler) EnqueueOrExecute(req *ExecutionRequest) { s.handler.OnTaskScheduled(req) } default: - // 队列满，直接执行（降级处理） - s.logger.Warnf("[Scheduler] 任务队列已满，直接执行任务 %s", req.TaskID) - go s.executeTask(req) + if s.config.StrictQueue { + s.logger.Errorf("[Scheduler] 任务队列已满，拒绝执行任务 %s", req.TaskID) + if s.handler != nil { + s.handler.OnTaskFailed(req, fmt.Errorf("任务队列已满，拒绝执行")) + } + } else { + // 队列满，直接执行（降级处理） + s.logger.Warnf("[Scheduler] 任务队列已满，直接执行任务 %s", req.TaskID) + go s.executeTask(req) + } } } @@ -592,8 +600,8 @@ func (s *Scheduler) Reload(config SchedulerConfig) { // 重启 workers s.Start() - s.logger.Infof("[Scheduler] 配置已重载: workers=%d, queue=%d, rate=%v", - config.WorkerCount, config.QueueSize, config.RateInterval) + s.logger.Infof("[Scheduler] 配置已重载: workers=%d, queue=%d, rate=%v, strict=%t", + config.WorkerCount, config.QueueSize, config.RateInterval, config.StrictQueue) } // GetQueueSize 获取当前队列大小 diff --git a/internal/models/agent.go b/internal/models/agent.go index b20150c..c7dba6c 100644 --- a/internal/models/agent.go +++ b/internal/models/agent.go @@ -1,28 +1,68 @@ package models import ( + "database/sql/driver" + "encoding/json" + "errors" + "time" + "github.com/engigu/baihu-panel/internal/constant" ) +// AgentSchedulerConfig Agent 调度器配置 +type AgentSchedulerConfig struct { + WorkerCount int `json:"worker_count"` + QueueSize int `json:"queue_size"` + RateInterval time.Duration `json:"rate_interval"` + Verbose bool `json:"verbose"` + StrictQueue bool `json:"strict_queue"` +} + +// Value 序列化为数据库字符串 +func (c AgentSchedulerConfig) Value() (driver.Value, error) { + bytes, err := json.Marshal(c) + if err != nil { + return nil, err + } + return string(bytes), nil +} + +// Scan 反序列化数据库字符串为结构体 +func (c *AgentSchedulerConfig) Scan(value interface{}) error { + if value == nil { + return nil + } + bytes, ok := value.([]byte) + if !ok { + str, ok := value.(string) + if !ok { + return errors.New("invalid type for AgentSchedulerConfig") + } + bytes = []byte(str) + } + return json.Unmarshal(bytes, c) +} + // Agent 远程执行代理 type Agent struct { - ID string `json:"id" gorm:"primaryKey;size:20"` - Name string `json:"name" gorm:"size:100;not null"` // Agent 名称 - Token string `json:"token" gorm:"size:64;index"` // 认证 Token（可重复使用） - MachineID string `json:"machine_id" gorm:"size:64;uniqueIndex"` // 机器识别码（唯一） - Description string `json:"description" gorm:"size:255"` // 描述 - Status string `json:"status" gorm:"size:20;default:'pending';index"` // 状态: constant.AgentStatusOnline, constant.AgentStatusOffline - LastSeen *LocalTime `json:"last_seen"` // 最后心跳时间 - IP string `json:"ip" gorm:"size:45"` // Agent IP 地址 - Version string `json:"version" gorm:"size:50"` // Agent 版本 - BuildTime string `json:"build_time" gorm:"size:30"` // Agent 构建时间 - Hostname string `json:"hostname" gorm:"size:100"` // Agent 主机名 - OS string `json:"os" gorm:"size:20"` // 操作系统 - Arch string `json:"arch" gorm:"size:20"` // 架构 - ForceUpdate bool `json:"force_update" gorm:"default:false"` // 强制更新标志 - Enabled *bool `json:"enabled" gorm:"default:true"` // 是否启用 - CreatedAt LocalTime `json:"created_at"` - UpdatedAt LocalTime `json:"updated_at"` + ID string `json:"id" gorm:"primaryKey;size:20"` + Name string `json:"name" gorm:"size:100;not null"` // Agent 名称 + Token string `json:"token" gorm:"size:64;index"` // 认证 Token（可重复使用） + MachineID string `json:"machine_id" gorm:"size:64;uniqueIndex"` // 机器识别码（唯一） + Description string `json:"description" gorm:"size:255"` // 描述 + Status string `json:"status" gorm:"size:20;default:'pending';index"` // 状态: constant.AgentStatusOnline, constant.AgentStatusOffline + LastSeen *LocalTime `json:"last_seen"` // 最后心跳时间 + IP string `json:"ip" gorm:"size:45"` // Agent IP 地址 + Version string `json:"version" gorm:"size:50"` // Agent 版本 + BuildTime string `json:"build_time" gorm:"size:30"` // Agent 构建时间 + Hostname string `json:"hostname" gorm:"size:100"` // Agent 主机名 + OS string `json:"os" gorm:"size:20"` // 操作系统 + Arch string `json:"arch" gorm:"size:20"` // 架构 + ForceUpdate bool `json:"force_update" gorm:"default:false"` // 强制更新标志 + Enabled *bool `json:"enabled" gorm:"default:true"` // 是否启用 + SchedulerConfig AgentSchedulerConfig `json:"scheduler_config" gorm:"type:text"` // 调度配置，以 JSON 字符串形式存储在 Text 类型字段中 + CreatedAt LocalTime `json:"created_at"` + UpdatedAt LocalTime `json:"updated_at"` } func (Agent) TableName() string { diff --git a/internal/models/vo/agent_vo.go b/internal/models/vo/agent_vo.go index bc05e34..8dbe4be 100644 --- a/internal/models/vo/agent_vo.go +++ b/internal/models/vo/agent_vo.go @@ -1,6 +1,8 @@ package vo import ( + "time" + "github.com/engigu/baihu-panel/internal/models" "github.com/engigu/baihu-panel/internal/utils" ) @@ -18,10 +20,11 @@ type AgentVO struct { Hostname string `json:"hostname"` OS string `json:"os"` Arch string `json:"arch"` - ForceUpdate bool `json:"force_update"` - Enabled bool `json:"enabled"` - CreatedAt models.LocalTime `json:"created_at"` - UpdatedAt models.LocalTime `json:"updated_at"` + ForceUpdate bool `json:"force_update"` + Enabled bool `json:"enabled"` + SchedulerConfig *AgentSchedulerConfigVO `json:"scheduler_config"` + CreatedAt models.LocalTime `json:"created_at"` + UpdatedAt models.LocalTime `json:"updated_at"` // 隐藏 Token 和 MachineID } @@ -30,6 +33,16 @@ func ToAgentVO(agent *models.Agent) *AgentVO { if agent == nil { return nil } + var schedulerConfigVO *AgentSchedulerConfigVO + if agent.SchedulerConfig.WorkerCount > 0 { + schedulerConfigVO = &AgentSchedulerConfigVO{ + WorkerCount: agent.SchedulerConfig.WorkerCount, + QueueSize: agent.SchedulerConfig.QueueSize, + RateInterval: int(agent.SchedulerConfig.RateInterval / time.Millisecond), + Verbose: agent.SchedulerConfig.Verbose, + StrictQueue: agent.SchedulerConfig.StrictQueue, + } + } return &AgentVO{ ID: agent.ID, Name: agent.Name, @@ -42,10 +55,11 @@ func ToAgentVO(agent *models.Agent) *AgentVO { Hostname: agent.Hostname, OS: agent.OS, Arch: agent.Arch, - ForceUpdate: agent.ForceUpdate, - Enabled: utils.DerefBool(agent.Enabled, true), - CreatedAt: agent.CreatedAt, - UpdatedAt: agent.UpdatedAt, + ForceUpdate: agent.ForceUpdate, + Enabled: utils.DerefBool(agent.Enabled, true), + SchedulerConfig: schedulerConfigVO, + CreatedAt: agent.CreatedAt, + UpdatedAt: agent.UpdatedAt, } } @@ -119,3 +133,12 @@ func ToAgentTokenVOListFromModels(tokens []models.AgentToken) []*AgentTokenVO { } return vos } + +// AgentSchedulerConfigVO 调度配置视图对象 +type AgentSchedulerConfigVO struct { + WorkerCount int `json:"worker_count"` + QueueSize int `json:"queue_size"` + RateInterval int `json:"rate_interval"` // 毫秒 + Verbose bool `json:"verbose"` + StrictQueue bool `json:"strict_queue"` +} diff --git a/internal/services/agent_service.go b/internal/services/agent_service.go index 2bf8266..c7d1f15 100644 --- a/internal/services/agent_service.go +++ b/internal/services/agent_service.go @@ -203,11 +203,12 @@ func (s *AgentService) Register(req *models.AgentRegisterRequest, ip string) (*m } // Update 更新 Agent -func (s *AgentService) Update(id string, name, description string, enabled bool) error { +func (s *AgentService) Update(id string, name, description string, enabled bool, schedulerConfig models.AgentSchedulerConfig) error { return database.DB.Model(&models.Agent{}).Where("id = ?", id).Updates(map[string]interface{}{ - "name": name, - "description": description, - "enabled": &enabled, + "name": name, + "description": description, + "enabled": &enabled, + "scheduler_config": schedulerConfig, }).Error } diff --git a/internal/services/tasks/executor_service.go b/internal/services/tasks/executor_service.go index fad2c4f..add68f2 100644 --- a/internal/services/tasks/executor_service.go +++ b/internal/services/tasks/executor_service.go @@ -937,7 +937,17 @@ func (es *ExecutorService) CheckConcurrency(taskID string) error { } if config.Concurrency == 0 && len(goids) > 0 { - return fmt.Errorf("任务正在运行中，拒绝并行执行，请前往日志查看") + // 检查目标 Agent 是否开启了排队机制 + var isAgentQueueing bool + if task.AgentID != nil && *task.AgentID != "" { + var agent models.Agent + if err := database.DB.Select("scheduler_config").Where("id = ?", *task.AgentID).First(&agent).Error; err == nil { + isAgentQueueing = agent.SchedulerConfig.StrictQueue + } + } + if !isAgentQueueing { + return fmt.Errorf("任务正在运行中，拒绝并行执行，请前往日志查看") + } } return nil } @@ -969,7 +979,17 @@ func (es *ExecutorService) AddRunningGo(taskID string) (int64, error) { // 如果并发为0(禁用)且已有执行中的任务，返回错误 if config.Concurrency == 0 && len(goids) > 0 { - return fmt.Errorf("task is running") + // 检查目标 Agent 是否开启了排队机制 + var isAgentQueueing bool + if task.AgentID != nil && *task.AgentID != "" { + var agent models.Agent + if err := tx.Select("scheduler_config").Where("id = ?", *task.AgentID).First(&agent).Error; err == nil { + isAgentQueueing = agent.SchedulerConfig.StrictQueue + } + } + if !isAgentQueueing { + return fmt.Errorf("task is running") + } } goids = append(goids, goid) diff --git a/web/src/api/index.ts b/web/src/api/index.ts index 76098cc..c473d8e 100644 --- a/web/src/api/index.ts +++ b/web/src/api/index.ts @@ -279,7 +279,7 @@ export const api = { agents: { list: () => request('/agents'), getVersion: () => request<{ version: string; platforms: { os: string; arch: string; filename: string }[] }>('/agents/version'), - update: (id: string, data: { name: string; description?: string; enabled: boolean }) => + update: (id: string, data: { name: string; description?: string; enabled: boolean; scheduler_config: SchedulerConfig | null }) => request('/agents/' + id, { method: 'PUT', body: JSON.stringify(data) }), delete: (id: string) => request('/agents/' + id, { method: 'DELETE' }), forceUpdate: (id: string) => request('/agents/' + id + '/update', { method: 'POST' }), @@ -582,10 +582,19 @@ export interface Agent { os: string arch: string enabled: boolean + scheduler_config: SchedulerConfig | null created_at: string updated_at: string } +export interface SchedulerConfig { + worker_count: number + queue_size: number + rate_interval: number + verbose: boolean + strict_queue: boolean +} + export interface AgentToken { id: string token: string diff --git a/web/src/views/agents/Agents.vue b/web/src/views/agents/Agents.vue index 0af3011..26a9c1f 100644 --- a/web/src/views/agents/Agents.vue +++ b/web/src/views/agents/Agents.vue @@ -3,6 +3,7 @@ import { ref, onMounted, computed, onUnmounted } from 'vue' import { Button } from '@/components/ui/button' import { Input } from '@/components/ui/input' import { Label } from '@/components/ui/label' +import { Switch } from '@/components/ui/switch' import { Dialog, DialogContent, DialogHeader, DialogTitle, DialogFooter, DialogDescription } from '@/components/ui/dialog' import { AlertDialog, AlertDialogAction, AlertDialogCancel, AlertDialogContent, AlertDialogDescription, AlertDialogFooter, AlertDialogHeader, AlertDialogTitle } from '@/components/ui/alert-dialog' import { Tabs, TabsContent, TabsList, TabsTrigger } from '@/components/ui/tabs' @@ -38,7 +39,18 @@ const showDownloadDialog = ref(false) const showTokenDialog = ref(false) const showEditTokenDialog = ref(false) const showDetailDialog = ref(false) -const formData = ref({ name: '', description: '' }) +const customScheduler = ref(false) +const formData = ref({ + name: '', + description: '', + scheduler_config: { + worker_count: 1, + queue_size: 100, + rate_interval: 200, + verbose: false, + strict_queue: false + } +}) const tokenForm = ref({ remark: '', max_uses: 0, expires_at: '' }) const editingToken = ref(null) const editTokenForm = ref({ remark: '', max_uses: 0, expires_at: '' }) @@ -89,14 +101,37 @@ function viewDetail(agent: Agent) { function openEditDialog(agent: Agent) { ;(document.activeElement as HTMLElement)?.blur() editingAgent.value = agent - formData.value = { name: agent.name, description: agent.description } + customScheduler.value = !!agent.scheduler_config + formData.value = { + name: agent.name, + description: agent.description, + scheduler_config: agent.scheduler_config ? { + worker_count: agent.scheduler_config.worker_count, + queue_size: agent.scheduler_config.queue_size, + rate_interval: agent.scheduler_config.rate_interval, + verbose: agent.scheduler_config.verbose, + strict_queue: agent.scheduler_config.strict_queue + } : { + worker_count: 1, + queue_size: 100, + rate_interval: 200, + verbose: false, + strict_queue: false + } + } showEditDialog.value = true } async function updateAgent() { if (!editingAgent.value || !formData.value.name.trim()) return try { - await api.agents.update(editingAgent.value.id, { ...formData.value, enabled: editingAgent.value.enabled }) + const payload = { + name: formData.value.name, + description: formData.value.description, + enabled: editingAgent.value.enabled, + scheduler_config: customScheduler.value ? formData.value.scheduler_config : null + } + await api.agents.update(editingAgent.value.id, payload) showEditDialog.value = false await loadAgents() toast.success('更新成功') @@ -108,7 +143,12 @@ async function updateAgent() { async function toggleEnabled(agent: Agent) { try { const newEnabled = !agent.enabled - await api.agents.update(agent.id, { name: agent.name, description: agent.description, enabled: newEnabled }) + await api.agents.update(agent.id, { + name: agent.name, + description: agent.description, + enabled: newEnabled, + scheduler_config: agent.scheduler_config + }) await loadAgents() toast.success(`${agent.name} 已${newEnabled ? '启用' : '禁用'}`) } catch (e: unknown) { @@ -612,6 +652,12 @@ onUnmounted(() => {
{{ viewingAgent.created_at || '-' }}
+
+ +
+ {{ viewingAgent.scheduler_config ? `自定义 (并发: ${viewingAgent.scheduler_config.worker_count}, 队列: ${viewingAgent.scheduler_config.queue_size}, 严格排队: ${viewingAgent.scheduler_config.strict_queue ? '是' : '否'})` : '继承全局' }} +
+
@@ -629,13 +675,55 @@ onUnmounted(() => { 修改 Agent 的名称和描述信息
-
- - +
+ +
-
- - +
+ + +
+
+
+ +
开启后可独立配置该 Agent 的并发限制与任务排队参数
+
+ +
+
+
+
+ + +

同一时间最大并行任务数

+
+
+ + +

并发满时等待排队的任务数

+
+
+
+
+ +
+ + ms +
+

两次调度启动的最小间隔时间

+
+
+ +
+ 队列满时拒绝执行 + +
+

开启后拒绝执行；关闭则同步直接执行

+
+
+
+ 提示：开启严格排队后，服务端将不再拦截此 Agent 上已运行任务的并行触发，而是交由 Agent 本地队列调度。若要严格保证任务不并行，请将 并发限制数 (Workers) 设为 1。 +