admin/TaskPool
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

chore: add sharedworker for task stauts

Browse Source
This commit is contained in:
duorameng
2026-05-07 20:12:30 +08:00
parent e5f9081a64
commit 4bf9ea8802
12 changed files with 542 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/constant/constant.go
+2
View File
@@ -99,6 +99,8 @@ const (
EventTaskSuccess = "task_success"
EventTaskFailed = "task_failed"
EventTaskTimeout = "task_timeout"
EventTaskRunning = "task_running"
EventTaskQueued = "task_queued"
// 其他事件类型
EventSystemNotice = "system_notice"
internal/controllers/system_ws_controller.go
+90
View File
@@ -0,0 +1,90 @@
package controllers
import (
"time"
"github.com/engigu/baihu-panel/internal/constant"
"github.com/engigu/baihu-panel/internal/logger"
"github.com/engigu/baihu-panel/internal/services"
"github.com/gin-gonic/gin"
"github.com/gorilla/websocket"
)
type SystemWSController struct {
manager *services.SystemWSManager
}
func NewSystemWSController() *SystemWSController {
return &SystemWSController{
manager: services.GetSystemWSManager(),
}
}
func (sc *SystemWSController) HandleEvents(c *gin.Context) {
conn, err := upgrader.Upgrade(c.Writer, c.Request, nil)
if err != nil {
logger.Errorf("[SystemWS] 升级 WebSocket 失败: %v", err)
return
}
client := sc.manager.Register(conn)
defer sc.manager.Unregister(client)
// 启动写循环
go sc.writeLoop(client)
// 启动读循环 (主要用于检测连接断开和维持心跳)
sc.readLoop(client)
}
func (sc *SystemWSController) readLoop(client *services.ClientConnection) {
defer client.Close()
client.Conn.SetReadLimit(constant.MaxMessageSize)
client.Conn.SetReadDeadline(time.Now().Add(constant.PongWait))
client.Conn.SetPongHandler(func(string) error {
client.Conn.SetReadDeadline(time.Now().Add(constant.PongWait))
return nil
})
for {
_, _, err := client.Conn.ReadMessage()
if err != nil {
if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway, websocket.CloseAbnormalClosure) {
logger.Warnf("[SystemWS] 客户端异常断开: %v", err)
}
break
}
// 暂时不处理来自前端的消息，前端仅作为接收方
}
}
func (sc *SystemWSController) writeLoop(client *services.ClientConnection) {
ticker := time.NewTicker(constant.PingPeriod)
defer func() {
ticker.Stop()
client.Close()
}()
for {
select {
case message, ok := <-client.Send:
client.Conn.SetWriteDeadline(time.Now().Add(10 * time.Second))
if !ok {
// 通道关闭
client.Conn.WriteMessage(websocket.CloseMessage, []byte{})
return
}
if err := client.Conn.WriteMessage(websocket.TextMessage, message); err != nil {
return
}
case <-ticker.C:
client.Conn.SetWriteDeadline(time.Now().Add(10 * time.Second))
if err := client.Conn.WriteMessage(websocket.PingMessage, nil); err != nil {
return
}
}
}
}
internal/models/vo/ws_vo.go
+8
View File
@@ -0,0 +1,8 @@
package vo
// WSMessage 通用 WebSocket 消息结构
type WSMessage struct {
Type string `json:"type"` // 事件类型: task_status, notice, system_stats
Timestamp int64 `json:"timestamp"` // 毫秒时间戳
Payload interface{} `json:"payload"` // 负载数据
}
internal/router/api_routes.go
+5
View File
@@ -61,6 +61,7 @@ func initAuthorizedAPIRoutes(api *gin.RouterGroup, c *Controllers) {
registerMiseRoutes(adminOnly, c)
registerNotificationRoutes(adminOnly, c)
registerAppLogRoutes(adminOnly, c)
registerSystemWSRoutes(adminOnly, c)
}
}
@@ -256,6 +257,10 @@ func registerAppLogRoutes(g *gin.RouterGroup, c *Controllers) {
}
}
func registerSystemWSRoutes(g *gin.RouterGroup, c *Controllers) {
g.GET("/ws/events", c.SystemWS.HandleEvents)
}
func initAgentAPIRoutes(root *gin.RouterGroup, c *Controllers) {
// Agent API（供远程 Agent 调用，不使用 /v1 版本号）
agentAPI := root.Group("/api/agent")
internal/router/register.go
+3 -1
View File
@@ -23,6 +23,7 @@ func RegisterControllers() *Controllers {
scriptService := services.NewScriptService()
sendStatsService := services.NewSendStatsService()
agentWSManager := services.GetAgentWSManager()
systemWSManager := services.GetSystemWSManager()
taskLogService := tasks.NewTaskLogService(sendStatsService)
// 创建任务执行服务（需要依赖注入）
@@ -40,7 +41,7 @@ func RegisterControllers() *Controllers {
executorService.StartCron()
// 初始化所有关注系统总线的服务
setupEventHandlers(appLogService, notifyService, loginLogService)
setupEventHandlers(appLogService, notifyService, loginLogService, systemWSManager)
go startAppLogCleanup(appLogService)
// 初始化并返回控制器
@@ -61,6 +62,7 @@ func RegisterControllers() *Controllers {
Mise: controllers.NewMiseController(services.NewMiseService()),
Notification: controllers.NewNotificationController(),
AppLog: controllers.NewAppLogController(),
SystemWS: controllers.NewSystemWSController(),
}
}
internal/router/router.go
+1
View File
@@ -29,6 +29,7 @@ type Controllers struct {
Mise *controllers.MiseController
Notification *controllers.NotificationController
AppLog *controllers.AppLogController
SystemWS *controllers.SystemWSController
}
func Setup(c *Controllers) *gin.Engine {
internal/services/system_ws_service.go
+126
View File
@@ -0,0 +1,126 @@
package services
import (
"encoding/json"
"sync"
"time"
"github.com/engigu/baihu-panel/internal/constant"
"github.com/engigu/baihu-panel/internal/eventbus"
"github.com/engigu/baihu-panel/internal/logger"
"github.com/engigu/baihu-panel/internal/models/vo"
"github.com/gorilla/websocket"
)
// SystemWSManager 前端系统事件 WebSocket 管理器 (单例)
type SystemWSManager struct {
clients map[*ClientConnection]bool
mu sync.RWMutex
}
// ClientConnection 代表一个前端页面的 WebSocket 连接
type ClientConnection struct {
Conn *websocket.Conn
Send chan []byte
closed bool
mu sync.Mutex
}
var systemWSManager *SystemWSManager
var systemWSOnce sync.Once
// GetSystemWSManager 获取系统 WebSocket 管理器单例
func GetSystemWSManager() *SystemWSManager {
systemWSOnce.Do(func() {
systemWSManager = &SystemWSManager{
clients: make(map[*ClientConnection]bool),
}
})
return systemWSManager
}
// Register 注册一个新的前端连接
func (m *SystemWSManager) Register(conn *websocket.Conn) *ClientConnection {
m.mu.Lock()
defer m.mu.Unlock()
client := &ClientConnection{
Conn: conn,
Send: make(chan []byte, 256),
}
m.clients[client] = true
return client
}
// Unregister 注销一个前端连接
func (m *SystemWSManager) Unregister(client *ClientConnection) {
m.mu.Lock()
defer m.mu.Unlock()
if _, ok := m.clients[client]; ok {
delete(m.clients, client)
client.Close()
}
}
// Broadcast 广播消息给所有在线前端
func (m *SystemWSManager) Broadcast(msgType string, payload interface{}) {
msg := vo.WSMessage{
Type: msgType,
Timestamp: time.Now().UnixMilli(),
Payload: payload,
}
data, err := json.Marshal(msg)
if err != nil {
logger.Errorf("[SystemWS] 序列化消息失败: %v", err)
return
}
m.mu.RLock()
defer m.mu.RUnlock()
for client := range m.clients {
select {
case client.Send <- data:
default:
// 缓冲区满，可能该客户端连接已死
go m.Unregister(client)
}
}
}
// SubscribeEvents 订阅系统事件总线并分发给 WebSocket
func (m *SystemWSManager) SubscribeEvents(bus *eventbus.EventBus) {
// 任务相关事件
taskEvents := []string{
constant.EventTaskSuccess,
constant.EventTaskFailed,
constant.EventTaskTimeout,
constant.EventTaskRunning,
constant.EventTaskQueued,
}
for _, evt := range taskEvents {
bus.Subscribe(evt, func(e eventbus.Event) {
m.Broadcast(e.Type, e.Payload)
})
}
// 系统通知事件
bus.Subscribe(constant.EventSystemNotice, func(e eventbus.Event) {
m.Broadcast("notice", e.Payload)
})
}
func (c *ClientConnection) Close() {
c.mu.Lock()
defer c.mu.Unlock()
if c.closed {
return
}
c.closed = true
c.Conn.Close()
close(c.Send)
}
internal/services/tasks/executor_service.go
+19 -1
View File
@@ -116,7 +116,15 @@ type ServerSchedulerHandler struct {
}
func (h *ServerSchedulerHandler) OnTaskScheduled(req *executor.ExecutionRequest) {
// 任务入队事件，可以在此处更新数据库状态为 "pending"
if req.TaskID != "" {
eventbus.DefaultBus.Publish(eventbus.Event{
Type: constant.EventTaskQueued,
Payload: map[string]interface{}{
"task_id": req.TaskID,
"status": constant.TaskStatusQueued,
},
})
}
}
func (h *ServerSchedulerHandler) OnTaskExecuting(req *executor.ExecutionRequest) (io.Writer, io.Writer, error) {
@@ -167,6 +175,16 @@ func (h *ServerSchedulerHandler) OnTaskExecuting(req *executor.ExecutionRequest)
StartTime: time.Now(),
})
// 发布任务开始运行事件
eventbus.DefaultBus.Publish(eventbus.Event{
Type: constant.EventTaskRunning,
Payload: map[string]interface{}{
"task_id": req.TaskID,
"status": constant.TaskStatusRunning,
"log_id": req.LogID,
},
})
if req.Metadata.RetryIndex > 0 {
tl.Write([]byte(fmt.Sprintf("\n[System] 此为任务失败后的第 %d 次重试执行...\n\n", req.Metadata.RetryIndex)))
}
web/public/workers/event-worker.js
+109
View File
@@ -0,0 +1,109 @@
/**
* Baihu Panel System Event SharedWorker
*
* 这个 Worker 维护全浏览器唯一的一个 WebSocket 连接，
* 并通过 postMessage 将收到的事件同步给所有打开的标签页。
*/
let ws = null;
const ports = new Set();
let reconnectTimer = null;
let lockReconnect = false;
// WebSocket 配置 (由第一个连接的页面通过消息传过来，或者使用默认值)
let wsUrl = '';
/**
* 建立 WebSocket 连接
*/
function connect() {
if (!wsUrl || ws) return;
try {
ws = new WebSocket(wsUrl);
ws.onopen = () => {
console.log('[SharedWorker] WebSocket 已连接');
lockReconnect = false;
};
ws.onmessage = (event) => {
// 广播消息给所有连接的端口
broadcast(event.data);
};
ws.onclose = () => {
console.log('[SharedWorker] WebSocket 已断开');
ws = null;
reconnect();
};
ws.onerror = (err) => {
console.error('[SharedWorker] WebSocket 错误:', err);
ws = null;
reconnect();
};
} catch (e) {
console.error('[SharedWorker] 建立连接失败:', e);
reconnect();
}
}
/**
* 重连逻辑
*/
function reconnect() {
if (lockReconnect) return;
lockReconnect = true;
if (reconnectTimer) clearTimeout(reconnectTimer);
reconnectTimer = setTimeout(() => {
connect();
lockReconnect = false;
}, 5000); // 5秒后尝试重连
}
/**
* 广播消息给所有活跃标签页
*/
function broadcast(data) {
const msg = typeof data === 'string' ? JSON.parse(data) : data;
ports.forEach(port => {
try {
port.postMessage(msg);
} catch (e) {
// 如果端口失效，移除它
ports.delete(port);
}
});
}
/**
* 处理 SharedWorker 连接
*/
self.onconnect = (e) => {
const port = e.ports[0];
ports.add(port);
port.onmessage = (event) => {
const { type, data } = event.data;
switch (type) {
case 'init':
// 初始化配置
if (data.url) {
wsUrl = data.url;
if (!ws) connect();
}
break;
case 'ping':
port.postMessage({ type: 'pong' });
break;
}
};
port.start();
// 发送一条初始消息确认连接成功
port.postMessage({ type: 'worker_ready' });
};
web/src/composables/useEventBus.ts
+35
View File
@@ -0,0 +1,35 @@
import { onMounted, onUnmounted } from 'vue';
import { eventBus, type WSMessage } from '../utils/event-bus';
/**
* Vue Composable: 使用系统事件总线
*
* @param eventType 可选，指定监听的消息类型
* @param handler 消息处理回调
*/
export function useEventBus(eventType: string | string[] | null, handler: (payload: any, type: string) => void) {
let unsubscribe: (() => void) | null = null;
const handleMessage = (msg: WSMessage) => {
if (!eventType) {
// 监听所有消息
handler(msg.payload, msg.type);
return;
}
const types = Array.isArray(eventType) ? eventType : [eventType];
if (types.includes(msg.type)) {
handler(msg.payload, msg.type);
}
};
onMounted(() => {
// 确保驱动已初始化 (单例内部会处理多次调用)
eventBus.init();
unsubscribe = eventBus.subscribe(handleMessage);
});
onUnmounted(() => {
if (unsubscribe) unsubscribe();
});
}
web/src/utils/event-bus.ts
+129
View File
@@ -0,0 +1,129 @@
/**
* 系统事件总线驱动 (带 SharedWorker 降级逻辑)
*/
export interface WSMessage {
type: string;
timestamp: number;
payload: any;
}
export type MessageHandler = (msg: WSMessage) => void;
class EventBusDriver {
private wsUrl: string;
private handlers: Set<MessageHandler> = new Set();
private worker: SharedWorker | null = null;
private socket: WebSocket | null = null;
private reconnectTimer: any = null;
constructor() {
// 获取基础路径配置
const baseUrl = (window as any).__BASE_URL__ || '';
const apiVersion = (window as any).__API_VERSION__ || '/api/v1';
// 自动计算 WebSocket 地址
let host = window.location.host;
let protocol = window.location.protocol === 'https:' ? 'wss:' : 'ws:';
// 如果 BASE_URL 是绝对路径 (如 http://api.example.com)
if (baseUrl.startsWith('http')) {
const url = new URL(baseUrl);
host = url.host;
protocol = url.protocol === 'https:' ? 'wss:' : 'ws:';
}
// 拼接最终的 WS 地址
const path = baseUrl.startsWith('http') ? '' : baseUrl;
this.wsUrl = `${protocol}//${host}${path}${apiVersion}/ws/events`;
// 记录基础路径用于 worker 加载
this.workerPath = `${path}/workers/event-worker.js`.replace(/\/+/g, '/');
}
private workerPath: string;
private initialized = false;
/**
* 初始化连接
*/
init() {
if (this.initialized) return;
if (window.SharedWorker) {
this.initSharedWorker();
} else {
this.initStandardWebSocket();
}
this.initialized = true;
}
private initSharedWorker() {
try {
this.worker = new SharedWorker(this.workerPath);
this.worker.port.postMessage({
type: 'init',
data: { url: this.wsUrl }
});
this.worker.port.onmessage = (e) => {
this.emit(e.data);
};
this.worker.port.start();
console.log('[EventBus] SharedWorker initialized');
} catch (e) {
console.error('[EventBus] SharedWorker failed, falling back', e);
this.initStandardWebSocket();
}
}
private initStandardWebSocket() {
if (this.socket) return;
this.socket = new WebSocket(this.wsUrl);
this.socket.onmessage = (e) => {
try {
const data = JSON.parse(e.data);
this.emit(data);
} catch (err) {
console.error('[EventBus] Parse message error', err);
}
};
this.socket.onclose = () => {
this.socket = null;
this.reconnect();
};
this.socket.onerror = () => {
this.socket = null;
this.reconnect();
};
console.log('[EventBus] Standard WebSocket initialized (Fallback)');
}
private reconnect() {
if (this.reconnectTimer) clearTimeout(this.reconnectTimer);
this.reconnectTimer = setTimeout(() => this.init(), 5000);
}
private emit(msg: WSMessage) {
if (!msg || !msg.type) return;
this.handlers.forEach(handler => handler(msg));
}
/**
* 订阅消息
*/
subscribe(handler: MessageHandler) {
this.handlers.add(handler);
return () => this.handlers.delete(handler);
}
}
// 导出单例
export const eventBus = new EventBusDriver();
web/src/views/tasks/Tasks.vue
+15 -1
View File
@@ -24,6 +24,7 @@ import { Label } from '@/components/ui/label'
import { api, type Agent, type Task, type TaskLog } from '@/api'
import { toast } from 'vue-sonner'
import { useSiteSettings } from '@/composables/useSiteSettings'
import { useEventBus } from '@/composables/useEventBus'
import { useRouter, useRoute } from 'vue-router'
import { TASK_TYPE, AGENT_STATUS, TRIGGER_TYPE, TASK_STATUS } from '@/constants'
import TextOverflow from '@/components/TextOverflow.vue'
@@ -275,7 +276,9 @@ function cleanupLogSocket() {
logSocket.onmessage = null
logSocket.onerror = null
logSocket.onclose = null
logSocket.close()
if (logSocket.readyState === WebSocket.CONNECTING || logSocket.readyState === WebSocket.OPEN) {
logSocket.close()
}
logSocket = null
}
}
@@ -472,6 +475,14 @@ onMounted(async () => {
loadViewsFromSettings()
})
// 订阅任务状态实时更新
useEventBus(['task_running', 'task_queued', 'task_success', 'task_failed', 'task_timeout'], (payload) => {
const task = tasks.value.find(t => t.id === payload.task_id)
if (task) {
task.running_status = payload.status
}
})
// 监听路由参数变化
watch(() => route.query.agent_id, (newVal: any) => {
filterAgentId.value = newVal ? String(newVal) : null
@@ -624,6 +635,7 @@ watch(() => route.query.agent_id, (newVal: any) => {
<div v-for="(task, index) in tasks" :key="`large-${task.id}`"
class="flex items-center gap-2 px-4 py-1.5 hover:bg-muted/30 transition-colors">
<div v-if="task.running_status === 'running'" class="h-2 w-2 rounded-full bg-amber-500 animate-pulse shadow-[0_0_8px_rgba(245,158,11,0.5)] shrink-0" title="运行中" />
<div v-else-if="task.running_status === 'queued' || task.running_status === 'pending'" class="h-2 w-2 rounded-full bg-blue-400 animate-pulse shrink-0" title="排队中" />
<div v-else class="h-1.5 w-1.5 rounded-full bg-muted-foreground/20 shrink-0" />
<div class="w-12 shrink-0 text-muted-foreground tabular-nums">#{{ total - (currentPage - 1) * pageSize - index }}</div>
<span class="w-8 shrink-0 flex justify-center" :title="getTaskTypeTitle(task.type || 'task')">
@@ -720,6 +732,7 @@ watch(() => route.query.agent_id, (newVal: any) => {
<div v-for="(task, index) in tasks" :key="`medium-${task.id}`"
class="flex items-center gap-2 px-4 py-2.5 hover:bg-muted/30 transition-colors">
<div v-if="task.running_status === 'running'" class="h-1.5 w-1.5 rounded-full bg-amber-500 animate-pulse shadow-[0_0_8px_rgba(245,158,11,0.5)] shrink-0" />
<div v-else-if="task.running_status === 'queued' || task.running_status === 'pending'" class="h-1.5 w-1.5 rounded-full bg-blue-400 animate-pulse shrink-0" />
<div v-else class="h-1 w-1 rounded-full bg-muted-foreground/20 shrink-0" />
<div class="w-12 shrink-0 text-muted-foreground tabular-nums text-xs">#{{ total - (currentPage - 1) * pageSize - index }}</div>
<div class="w-48 shrink-0 flex items-center gap-2 overflow-hidden">
@@ -789,6 +802,7 @@ watch(() => route.query.agent_id, (newVal: any) => {
<div class="flex items-start justify-between mb-3 border-b border-border/40 pb-2">
<div class="flex items-center gap-2 flex-1 min-w-0 pr-2">
<div v-if="task.running_status === 'running'" class="h-1.5 w-1.5 rounded-full bg-amber-500 animate-pulse shadow-[0_0_8px_rgba(245,158,11,0.5)] shrink-0" />
<div v-else-if="task.running_status === 'queued' || task.running_status === 'pending'" class="h-1.5 w-1.5 rounded-full bg-blue-400 animate-pulse shrink-0" />
<div v-else class="h-1 w-1 rounded-full bg-muted-foreground/20 shrink-0" />
<span class="text-xs text-muted-foreground tabular-nums flex-shrink-0">#{{ total - (currentPage - 1) * pageSize - index }}</span>
<span class="shrink-0">