diff --git a/README.md b/README.md index 3ad2e22..0e0446a 100644 --- a/README.md +++ b/README.md @@ -213,6 +213,25 @@ log_retention_days = 30 | 分页大小 | 列表每页显示条数 | 10 | | Cookie 有效期 | 登录状态保持天数 | 7 | +### 调度设置 + +以下设置可在管理面板「系统设置 > 调度设置」中配置，用于优化任务执行性能： + +| 设置项 | 说明 | 默认值 | +|--------|------|--------| +| Worker 数量 | 并发执行任务的 worker 数量 | 4 | +| 队列大小 | 任务队列缓冲区大小 | 100 | +| 速率间隔 | 任务启动间隔（毫秒），200ms = 每秒最多 5 个任务 | 200 | + +**调度机制说明：** + +系统采用 Worker Pool + 任务队列的架构来控制任务执行： + +1. **Worker Pool**：固定数量的 worker 从队列中取任务执行，避免 cron 触发时无限并发 +2. **任务队列**：cron 调度器触发的任务先入队，由 worker 按顺序执行 +3. **速率限制**：控制任务启动频率，防止秒级 cron 同时触发多个进程导致 CPU 峰值 +4. **热重载**：修改调度设置后立即生效，无需重启服务 + --- ## 🛠️ 技术栈 diff --git a/internal/constant/constant.go b/internal/constant/constant.go index 459f464..3f14c99 100644 --- a/internal/constant/constant.go +++ b/internal/constant/constant.go @@ -30,8 +30,9 @@ const ( DefaultTaskTimeout = 30 // Settings Section 常量 - SectionSite = "site" - SectionSystem = "system" + SectionSite = "site" + SectionSystem = "system" + SectionScheduler = "scheduler" // Site Settings Key 常量 KeyTitle = "title" @@ -42,6 +43,11 @@ const ( // System Settings Key 常量 KeyInitialized = "initialized" + + // Scheduler Settings Key 常量 + KeyWorkerCount = "worker_count" + KeyQueueSize = "queue_size" + KeyRateInterval = "rate_interval" ) // TablePrefix 表前缀，从配置文件读取 @@ -62,4 +68,9 @@ var DefaultSettings = map[string]map[string]string{ KeyPageSize: "10", KeyCookieDays: "7", }, + SectionScheduler: { + KeyWorkerCount: "4", + KeyQueueSize: "100", + KeyRateInterval: "200", + }, } diff --git a/internal/controllers/settings_controller.go b/internal/controllers/settings_controller.go index 70f25b4..022b9f0 100644 --- a/internal/controllers/settings_controller.go +++ b/internal/controllers/settings_controller.go @@ -22,14 +22,16 @@ type SettingsController struct { settingsService *services.SettingsService loginLogService *services.LoginLogService backupService *services.BackupService + executorService *services.ExecutorService } -func NewSettingsController(userService *services.UserService, loginLogService *services.LoginLogService) *SettingsController { +func NewSettingsController(userService *services.UserService, loginLogService *services.LoginLogService, executorService *services.ExecutorService) *SettingsController { return &SettingsController{ userService: userService, settingsService: services.NewSettingsService(), loginLogService: loginLogService, backupService: services.NewBackupService(), + executorService: executorService, } } @@ -115,6 +117,44 @@ func (sc *SettingsController) UpdateSiteSettings(c *gin.Context) { utils.SuccessMsg(c, "保存成功") } +// GetSchedulerSettings 获取调度设置 +func (sc *SettingsController) GetSchedulerSettings(c *gin.Context) { + settings := sc.settingsService.GetSection(constant.SectionScheduler) + utils.Success(c, settings) +} + +// UpdateSchedulerSettings 更新调度设置 +func (sc *SettingsController) UpdateSchedulerSettings(c *gin.Context) { + var req struct { + WorkerCount string `json:"worker_count"` + QueueSize string `json:"queue_size"` + RateInterval string `json:"rate_interval"` + } + + if err := c.ShouldBindJSON(&req); err != nil { + utils.BadRequest(c, "参数错误") + return + } + + values := map[string]string{ + constant.KeyWorkerCount: req.WorkerCount, + constant.KeyQueueSize: req.QueueSize, + constant.KeyRateInterval: req.RateInterval, + } + + if err := sc.settingsService.SetSection(constant.SectionScheduler, values); err != nil { + utils.ServerError(c, "保存失败") + return + } + + // 重新加载 executor service + if sc.executorService != nil { + sc.executorService.Reload() + } + + utils.SuccessMsg(c, "保存成功") +} + // GetAbout 获取关于信息 func (sc *SettingsController) GetAbout(c *gin.Context) { var taskCount, logCount, envCount int64 diff --git a/internal/router/register.go b/internal/router/register.go index 004c9bf..30e710f 100644 --- a/internal/router/register.go +++ b/internal/router/register.go @@ -37,7 +37,7 @@ func RegisterControllers() *Controllers { Dashboard: controllers.NewDashboardController(cronService, executorService), Log: controllers.NewLogController(), Terminal: controllers.NewTerminalController(), - Settings: controllers.NewSettingsController(userService, loginLogService), + Settings: controllers.NewSettingsController(userService, loginLogService, executorService), Runtime: controllers.NewRuntimeController(), } } diff --git a/internal/router/router.go b/internal/router/router.go index 0b35339..9f5189e 100644 --- a/internal/router/router.go +++ b/internal/router/router.go @@ -175,6 +175,8 @@ func Setup(c *Controllers) *gin.Engine { settings.POST("/password", c.Settings.ChangePassword) settings.GET("/site", c.Settings.GetSiteSettings) settings.PUT("/site", c.Settings.UpdateSiteSettings) + settings.GET("/scheduler", c.Settings.GetSchedulerSettings) + settings.PUT("/scheduler", c.Settings.UpdateSchedulerSettings) settings.GET("/about", c.Settings.GetAbout) settings.GET("/loginlogs", c.Settings.GetLoginLogs) settings.POST("/backup", c.Settings.CreateBackup) diff --git a/internal/services/cron_service.go b/internal/services/cron_service.go index f63a453..4347645 100644 --- a/internal/services/cron_service.go +++ b/internal/services/cron_service.go @@ -110,8 +110,8 @@ func (cs *CronService) runTask(taskID uint) { now := time.Now() database.DB.Model(&models.Task{}).Where("id = ?", taskID).Update("last_run", now) - // 执行任务 - cs.executorService.ExecuteTask(int(taskID)) + // 将任务加入队列执行（通过 worker pool 控制并发） + cs.executorService.EnqueueTask(int(taskID)) // 更新 next_run cs.updateNextRun(taskID) diff --git a/internal/services/executor_service.go b/internal/services/executor_service.go index 346ab40..8d6f7d7 100644 --- a/internal/services/executor_service.go +++ b/internal/services/executor_service.go @@ -1,19 +1,19 @@ package services import ( - "bytes" - "context" - "encoding/json" - "os" - "os/exec" - "sync" - "time" - "baihu/internal/constant" "baihu/internal/database" "baihu/internal/logger" "baihu/internal/models" "baihu/internal/utils" + "bytes" + "context" + "encoding/json" + "fmt" + "os" + "os/exec" + "sync" + "time" ) // ExecutionResult represents the result of a task execution @@ -29,6 +29,11 @@ type ExecutionResult struct { // ExecutionCallback 任务执行完成后的回调函数类型 type ExecutionCallback func(taskID uint, command string, result *ExecutionResult) +// taskJob 任务队列项 +type taskJob struct { + taskID int +} + // ExecutorService handles task execution type ExecutorService struct { taskService *TaskService @@ -36,23 +41,119 @@ type ExecutorService struct { runningTasks map[int]bool callbacks []ExecutionCallback mu sync.RWMutex + resultsMu sync.RWMutex + + // 任务队列和 worker pool + taskQueue chan taskJob + workerCount int + rateLimiter <-chan time.Time + stopCh chan struct{} + wg sync.WaitGroup } // NewExecutorService creates a new executor service func NewExecutorService(taskService *TaskService) *ExecutorService { + // 从设置中读取调度配置 + settingsService := NewSettingsService() + workerCount := getIntSetting(settingsService, constant.SectionScheduler, constant.KeyWorkerCount, 4) + queueSize := getIntSetting(settingsService, constant.SectionScheduler, constant.KeyQueueSize, 100) + rateInterval := getIntSetting(settingsService, constant.SectionScheduler, constant.KeyRateInterval, 200) + + logger.Infof("Executor service config: workers=%d, queue=%d, rate=%dms", workerCount, queueSize, rateInterval) + es := &ExecutorService{ taskService: taskService, - results: make([]ExecutionResult, 0), + results: make([]ExecutionResult, 0, 100), runningTasks: make(map[int]bool), callbacks: make([]ExecutionCallback, 0), + taskQueue: make(chan taskJob, queueSize), + workerCount: workerCount, + rateLimiter: time.Tick(time.Duration(rateInterval) * time.Millisecond), + stopCh: make(chan struct{}), } + // 注册默认回调 es.RegisterCallback(es.saveTaskLogCallback) es.RegisterCallback(es.updateStatsCallback) es.RegisterCallback(es.cleanLogsCallback) + + // 启动 worker pool + es.startWorkers() + return es } +// getIntSetting 从设置中获取整数值 +func getIntSetting(s *SettingsService, section, key string, defaultVal int) int { + val := s.Get(section, key) + if val == "" { + return defaultVal + } + var result int + if _, err := fmt.Sscanf(val, "%d", &result); err != nil { + return defaultVal + } + return result +} + +// startWorkers 启动 worker pool +func (es *ExecutorService) startWorkers() { + for i := 0; i < es.workerCount; i++ { + es.wg.Add(1) + go es.worker(i) + } +} + +// worker 从队列中取任务执行 +func (es *ExecutorService) worker(id int) { + defer es.wg.Done() + for { + select { + case <-es.stopCh: + return + case job := <-es.taskQueue: + // 速率限制 + <-es.rateLimiter + es.executeTaskInternal(job.taskID) + } + } +} + +// Stop 停止 executor service +func (es *ExecutorService) Stop() { + close(es.stopCh) + es.wg.Wait() +} + +// Reload 重新加载配置并重建 worker pool +func (es *ExecutorService) Reload() { + logger.Info("Reloading executor service...") + + // 停止现有 workers + close(es.stopCh) + es.wg.Wait() + logger.Info("Stopped executor service...") + + // 从设置中读取新配置 + settingsService := NewSettingsService() + workerCount := getIntSetting(settingsService, constant.SectionScheduler, constant.KeyWorkerCount, 4) + queueSize := getIntSetting(settingsService, constant.SectionScheduler, constant.KeyQueueSize, 100) + rateInterval := getIntSetting(settingsService, constant.SectionScheduler, constant.KeyRateInterval, 200) + + // 重建 channel 和配置 + es.mu.Lock() + es.taskQueue = make(chan taskJob, queueSize) + es.workerCount = workerCount + es.rateLimiter = time.Tick(time.Duration(rateInterval) * time.Millisecond) + es.stopCh = make(chan struct{}) + es.mu.Unlock() + + // 启动新的 workers + es.startWorkers() + + logger.Infof("Executor service reloaded: workers=%d, queue=%d, rate=%dms", workerCount, queueSize, rateInterval) +} + // RegisterCallback 注册执行完成回调 func (es *ExecutorService) RegisterCallback(cb ExecutionCallback) { es.mu.Lock() @@ -60,19 +161,21 @@ func (es *ExecutorService) RegisterCallback(cb ExecutionCallback) { es.mu.Unlock() } -// executeCallbacks 执行所有回调 -func (es *ExecutorService) executeCallbacks(taskID uint, command string, result *ExecutionResult) { +// executeCallbacksAsync 异步执行所有回调 +func (es *ExecutorService) executeCallbacksAsync(taskID uint, command string, result *ExecutionResult) { es.mu.RLock() callbacks := make([]ExecutionCallback, len(es.callbacks)) copy(callbacks, es.callbacks) es.mu.RUnlock() - for _, cb := range callbacks { - cb(taskID, command, result) - } + go func() { + for _, cb := range callbacks { + cb(taskID, command, result) + } + }() } -// saveTaskLogCallback 保存任务日志的回调 +// saveTaskLogCallback 保存任务日志的回调（异步执行，压缩在此处进行） func (es *ExecutorService) saveTaskLogCallback(taskID uint, command string, result *ExecutionResult) { output := result.Output if result.Error != "" { @@ -141,17 +244,13 @@ func (es *ExecutorService) cleanLogsCallback(taskID uint, _ string, _ *Execution var deleted int64 switch config.Type { case "day": - // 按天清理：删除 N 天前的日志 cutoff := time.Now().AddDate(0, 0, -config.Keep) result := database.DB.Where("task_id = ? AND created_at < ?", taskID, cutoff).Delete(&models.TaskLog{}) deleted = result.RowsAffected case "count": - // 按条数清理：使用子查询删除超出保留数量的旧日志 - // 先获取第 N 条的 ID 作为边界 var boundaryLog models.TaskLog err := database.DB.Where("task_id = ?", taskID).Order("id DESC").Offset(config.Keep - 1).Limit(1).First(&boundaryLog).Error if err == nil { - // 删除 ID 小于边界的所有日志 result := database.DB.Where("task_id = ? AND id < ?", taskID, boundaryLog.ID).Delete(&models.TaskLog{}) deleted = result.RowsAffected } @@ -162,8 +261,25 @@ func (es *ExecutorService) cleanLogsCallback(taskID uint, _ string, _ *Execution } } -// ExecuteTask executes a task by ID +// EnqueueTask 将任务加入队列（供 cron 调度器调用） +func (es *ExecutorService) EnqueueTask(taskID int) { + select { + case es.taskQueue <- taskJob{taskID: taskID}: + // 成功入队 + default: + // 队列满，直接执行（降级处理） + logger.Warnf("Task queue full, executing task %d directly", taskID) + go es.executeTaskInternal(taskID) + } +} + +// ExecuteTask executes a task by ID（同步执行，供 API 调用） func (es *ExecutorService) ExecuteTask(taskID int) *ExecutionResult { + return es.executeTaskInternal(taskID) +} + +// executeTaskInternal 内部执行任务逻辑 +func (es *ExecutorService) executeTaskInternal(taskID int) *ExecutionResult { task := es.taskService.GetTaskByID(taskID) if task == nil { return &ExecutionResult{ @@ -197,8 +313,8 @@ func (es *ExecutorService) ExecuteTask(taskID int) *ExecutionResult { delete(es.runningTasks, taskID) es.mu.Unlock() - // 执行回调 - es.executeCallbacks(uint(taskID), task.Command, result) + // 异步执行回调（日志压缩、统计更新、日志清理） + es.executeCallbacksAsync(uint(taskID), task.Command, result) return result } @@ -255,20 +371,21 @@ func (es *ExecutorService) ExecuteCommandWithEnv(command string, timeout time.Du result.Success = true } - es.mu.Lock() + // 使用独立锁保存结果 + es.resultsMu.Lock() es.results = append(es.results, *result) if len(es.results) > 100 { es.results = es.results[1:] } - es.mu.Unlock() + es.resultsMu.Unlock() return result } // GetLastResults returns the last execution results func (es *ExecutorService) GetLastResults(count int) []ExecutionResult { - es.mu.RLock() - defer es.mu.RUnlock() + es.resultsMu.RLock() + defer es.resultsMu.RUnlock() start := 0 if len(es.results) > count { diff --git a/web/src/api/index.ts b/web/src/api/index.ts index 74aeebb..74413e3 100644 --- a/web/src/api/index.ts +++ b/web/src/api/index.ts @@ -114,6 +114,9 @@ export const api = { getPublicSite: () => request<{ title: string; subtitle: string; icon: string }>('/settings/public'), updateSite: (data: SiteSettings) => request('/settings/site', { method: 'PUT', body: JSON.stringify(data) }), + getScheduler: () => request('/settings/scheduler'), + updateScheduler: (data: SchedulerSettings) => + request('/settings/scheduler', { method: 'PUT', body: JSON.stringify(data) }), getAbout: () => request('/settings/about'), getLoginLogs: (params?: { page?: number; page_size?: number; username?: string }) => { const query = new URLSearchParams() @@ -307,6 +310,12 @@ export interface SiteSettings { cookie_days: string } +export interface SchedulerSettings { + worker_count: string + queue_size: string + rate_interval: string +} + export interface LoginLog { id: number diff --git a/web/src/assets/index.css b/web/src/assets/index.css index 4fbd447..cc3670f 100644 --- a/web/src/assets/index.css +++ b/web/src/assets/index.css @@ -143,3 +143,24 @@ line-height: 1.5; } } + +/* Number input spinner buttons - dark mode */ +input[type="number"]::-webkit-inner-spin-button, +input[type="number"]::-webkit-outer-spin-button { + opacity: 0.6; +} + +input[type="number"]:hover::-webkit-inner-spin-button, +input[type="number"]:hover::-webkit-outer-spin-button { + opacity: 1; +} + +.dark input[type="number"]::-webkit-inner-spin-button, +.dark input[type="number"]::-webkit-outer-spin-button { + filter: invert(0.7); +} + +.dark input[type="number"]:hover::-webkit-inner-spin-button, +.dark input[type="number"]:hover::-webkit-outer-spin-button { + filter: invert(0.85); +} diff --git a/web/src/views/settings/BackupSettings.vue b/web/src/views/settings/BackupSettings.vue index 3834958..453b735 100644 --- a/web/src/views/settings/BackupSettings.vue +++ b/web/src/views/settings/BackupSettings.vue @@ -21,7 +21,6 @@ const backupLoading = ref(false) const restoreLoading = ref(false) const fileInput = ref() const showConfirm = ref(false) -const pendingFile = ref(null) async function checkBackupStatus() { try { @@ -49,11 +48,20 @@ function downloadBackup() { setTimeout(checkBackupStatus, 6000) } -function triggerFileSelect() { +function showRestoreConfirm() { + showConfirm.value = true +} + +function confirmRestore() { + showConfirm.value = false fileInput.value?.click() } -function handleFileSelect(e: Event) { +function cancelRestore() { + showConfirm.value = false +} + +async function handleFileSelect(e: Event) { const target = e.target as HTMLInputElement const file = target.files?.[0] if (!file) return @@ -64,33 +72,19 @@ function handleFileSelect(e: Event) { return } - pendingFile.value = file - showConfirm.value = true - target.value = '' -} - -async function confirmRestore() { - if (!pendingFile.value) return - - showConfirm.value = false restoreLoading.value = true try { - await api.settings.restoreBackup(pendingFile.value) + await api.settings.restoreBackup(file) toast.success('恢复成功，页面即将刷新') setTimeout(() => window.location.reload(), 1500) } catch (e: any) { toast.error(e.message || '恢复失败') } finally { restoreLoading.value = false - pendingFile.value = null + target.value = '' } } -function cancelRestore() { - showConfirm.value = false - pendingFile.value = null -} - onMounted(checkBackupStatus) @@ -112,7 +106,7 @@ onMounted(checkBackupStatus)
- diff --git a/web/src/views/settings/SchedulerSettings.vue b/web/src/views/settings/SchedulerSettings.vue new file mode 100644 index 0000000..e09acf8 --- /dev/null +++ b/web/src/views/settings/SchedulerSettings.vue @@ -0,0 +1,108 @@ + + + diff --git a/web/src/views/settings/Settings.vue b/web/src/views/settings/Settings.vue index b2d8af6..0ed1ce9 100644 --- a/web/src/views/settings/Settings.vue +++ b/web/src/views/settings/Settings.vue @@ -4,6 +4,7 @@ import { Card, CardContent, CardDescription, CardHeader, CardTitle } from '@/com import { Tabs, TabsContent, TabsList, TabsTrigger } from '@/components/ui/tabs' import PasswordSettings from './PasswordSettings.vue' import SiteSettings from './SiteSettings.vue' +import SchedulerSettings from './SchedulerSettings.vue' import BackupSettings from './BackupSettings.vue' import AboutSettings from './AboutSettings.vue' @@ -21,6 +22,7 @@ const activeTab = ref('password') 密码修改 站点设置 + 调度设置 备份恢复 关于 @@ -49,6 +51,18 @@ const activeTab = ref('password') + + + + 调度设置 + 配置任务调度器参数 + + + + + + +