修复配置更新任务解析失败问题

web/events/proxy.go

@@ -1,9 +1,13 @@
 package events
 
-import "github.com/hibiken/asynq"
+import (
+	"time"
+
+	"github.com/hibiken/asynq"
+)
 
 const FlushGateway = "gateway:flush"
 
-func NewFlushGateway() *asynq.Task {
-	return asynq.NewTask(FlushGateway, nil)
+func NewFlushGateway(ttl time.Duration) *asynq.Task {
+	return asynq.NewTask(FlushGateway, nil, asynq.MaxRetry(0), asynq.Unique(ttl))
 }

web/tasks/task.go

@@ -59,8 +59,14 @@ func HandleRemoveChannel(_ context.Context, task *asynq.Task) (err error) {
 	return nil
 }
 
-func HandleFlushGateway(_ context.Context, task *asynq.Task) (err error) {
+func HandleFlushGateway(_ context.Context, task *asynq.Task) error {
 	start := time.Now()
+	defer func() {
+		duration := time.Since(start)
+		if duration > time.Second {
+			slog.Warn("更新代理后备配置耗时过长", "time", duration.String())
+		}
+	}()
 
 	// 获取所有网关:配置组
 	proxies, err := s.Proxy.AllProxies(m.ProxyTypeBaiYin, true)
@@ -82,12 +88,16 @@ func HandleFlushGateway(_ context.Context, task *asynq.Task) (err error) {
 		// 获取之前的后备配置
 		remotes := map[string]int{}
 		if proxy.Meta != nil {
-			meta, ok := proxy.Meta.Data().([]g.AutoConfig)
+			meta, ok := proxy.Meta.Data().([]any)
 			if !ok {
-				return fmt.Errorf("解析网关数据失败")
+				return fmt.Errorf("解析网关数据失败: %T", proxy.Meta.Data())
 			}
-			for _, m := range meta {
-				remotes[fmt.Sprintf("%s:%s:%s", m.Isp, m.Province, m.City)] = m.Count
+			for _, rawM := range meta {
+				m, ok := rawM.(map[string]any)
+				if !ok {
+					return fmt.Errorf("解析网关数据失败: %T", rawM)
+				}
+				remotes[fmt.Sprintf("%s:%s:%s", m["isp"], m["province"], m["city"])] = int(m["count"].(float64))
 			}
 		}
 
@@ -142,6 +152,5 @@ func HandleFlushGateway(_ context.Context, task *asynq.Task) (err error) {
 		}
 	}
 
-	slog.Debug("更新代理后备配置", "time", time.Since(start).String())
 	return nil
 }

web/web.go

@@ -81,7 +81,7 @@ func RunSchedule(ctx context.Context) error {
 		Location: time.Local,
 	})
 
-	scheduler.Register("* * * * *", events.NewFlushGateway())
+	scheduler.Register("@every 5s", events.NewFlushGateway(5*time.Second))
 
 	// 停止服务
 	go func() {
@@ -99,7 +99,11 @@ func RunSchedule(ctx context.Context) error {
 }
 
 func RunTask(ctx context.Context) error {
-	var server = asynq.NewServerFromRedisClient(base.Redis, asynq.Config{})
+	var server = asynq.NewServerFromRedisClient(base.Redis, asynq.Config{
+		ErrorHandler: asynq.ErrorHandlerFunc(func(ctx context.Context, task *asynq.Task, err error) {
+			slog.Error("任务执行失败", "task", task.Type(), "error", err)
+		}),
+	})
 
 	var mux = asynq.NewServeMux()
 	mux.HandleFunc(events.RemoveChannel, tasks.HandleRemoveChannel)