sugarBecomer
107 lines
2.8 KiB
Go
107 lines
2.8 KiB
Go
package schedule
|
|
|
|
import (
|
|
"context"
|
|
cst2 "likedream-goframe/internal/common/cst"
|
|
"likedream-goframe/internal/common/db"
|
|
"likedream-goframe/internal/ent/cptask"
|
|
"likedream-goframe/internal/framework/queue"
|
|
"likedream-goframe/internal/utils"
|
|
|
|
"github.com/gogf/gf/v2/frame/g"
|
|
"github.com/gogf/gf/v2/os/gctx"
|
|
"github.com/hibiken/asynq"
|
|
)
|
|
|
|
func Init(ctx context.Context) {
|
|
queue.InitClient(ctx)
|
|
initServer(ctx)
|
|
initInspector(ctx)
|
|
cleanupStuckTasks(ctx) // 清理卡住的任务
|
|
initScanner()
|
|
initPeriodicTasks(ctx)
|
|
}
|
|
|
|
// cleanupStuckTasks 清理服务重启时卡在 queue 或 doing 状态的任务
|
|
// 检查 asynq 队列中是否有 archived 的任务，如果有则删除并重置数据库状态
|
|
func cleanupStuckTasks(ctx context.Context) {
|
|
// 获取所有状态为 queue 或 doing 的任务
|
|
stuckTasks, err := db.Global.CpTask.Query().
|
|
Where(
|
|
cptask.Or(
|
|
cptask.TaskStatusEQ(cptask.TaskStatusQueue),
|
|
cptask.TaskStatusEQ(cptask.TaskStatusDoing),
|
|
),
|
|
).
|
|
All(ctx)
|
|
|
|
if err != nil {
|
|
g.Log().Warningf(ctx, "查询卡住的任务失败: %v", err)
|
|
return
|
|
}
|
|
|
|
if len(stuckTasks) == 0 {
|
|
return
|
|
}
|
|
|
|
// 检查 archived 队列中的任务数量
|
|
archivedTasks, err := globalInspector.ListArchivedTasks(cst2.TaskQueueCpdaily, asynq.PageSize(10000))
|
|
if err != nil {
|
|
g.Log().Warningf(ctx, "获取 archived 任务列表失败: %v", err)
|
|
}
|
|
|
|
// 删除所有 archived 的任务，让它们可以重新入队
|
|
for _, task := range archivedTasks {
|
|
if err := globalInspector.DeleteTask(cst2.TaskQueueCpdaily, task.ID); err != nil {
|
|
g.Log().Debugf(ctx, "删除 archived 任务失败: %v", err)
|
|
}
|
|
}
|
|
|
|
if len(archivedTasks) > 0 {
|
|
g.Log().Infof(ctx, "已清理 %d 个 archived 状态的 asynq 任务", len(archivedTasks))
|
|
}
|
|
|
|
// 重置数据库中卡住的任务状态（queue 和 doing 都需要重置）
|
|
affected, err := db.Global.CpTask.Update().
|
|
Where(
|
|
cptask.Or(
|
|
cptask.TaskStatusEQ(cptask.TaskStatusQueue),
|
|
cptask.TaskStatusEQ(cptask.TaskStatusDoing),
|
|
),
|
|
).
|
|
SetTaskStatus(cptask.TaskStatusNotStart).
|
|
Save(ctx)
|
|
|
|
if err != nil {
|
|
g.Log().Warningf(ctx, "重置卡住的任务失败: %v", err)
|
|
return
|
|
}
|
|
|
|
if affected > 0 {
|
|
g.Log().Infof(ctx, "已重置 %d 个卡住的任务状态为 not_start", affected)
|
|
}
|
|
}
|
|
|
|
func Stop() {
|
|
globalScanner.StopScan()
|
|
g.Log().Info(gctx.GetInitCtx(), "task scanner stopped")
|
|
|
|
if globalPeriodicTaskManager != nil {
|
|
globalPeriodicTaskManager.Shutdown()
|
|
}
|
|
|
|
e := queue.GetClient().Close()
|
|
utils.OkOrPanic(e)
|
|
g.Log().Info(gctx.GetInitCtx(), "asynq client closed")
|
|
|
|
globalServer.Shutdown()
|
|
g.Log().Info(gctx.GetInitCtx(), "asynq main server stopped")
|
|
|
|
auxiliaryServer.Shutdown()
|
|
g.Log().Info(gctx.GetInitCtx(), "asynq auxiliary server stopped")
|
|
|
|
e = globalInspector.Close()
|
|
utils.OkOrPanic(e)
|
|
g.Log().Info(gctx.GetInitCtx(), "asynq inspector closed")
|
|
}
|