Task Scheduling in Practice

// app/core/service/internal/server/asynq_server.go
func NewAsynqServer(ctx *bootstrap.Context, logger log.Logger) *asynq.Server {
    cfg := ctx.GetConfig().GetServer().GetAsynq()

    srv := asynq.NewServer(
        asynq.RedisClientOpt{Addr: cfg.Uri},
        asynq.Config{
            Concurrency: int(cfg.Concurrency),
            Queues:      parseQueues(cfg.Queues),
        },
    )

    mux := asynq.NewServeMux()

    // 注册任务处理器
    mux.HandleFunc(task.TypeIndexPost, HandleIndexPost)
    mux.HandleFunc(task.TypeSendNotification, HandleSendNotification)
    mux.HandleFunc(task.TypeCleanupExpiredData, HandleCleanupExpiredData)
    mux.HandleFunc(task.TypeGenerateReport, HandleGenerateReport)
    mux.HandleFunc(task.TypeSyncTranslations, HandleSyncTranslations)

    return srv
}

// pkg/task/task_types.go
const (
    // 内容索引任务
    TypeIndexPost        = "post:index"
    TypeIndexPage        = "page:index"
    TypeRebuildIndex     = "search:rebuild_index"

    // 通知任务
    TypeSendNotification = "notification:send"
    TypeSendEmail        = "email:send"

    // 维护任务
    TypeCleanupExpiredData  = "cleanup:expired_data"
    TypeCleanupTempFiles    = "cleanup:temp_files"
    TypeSyncTranslations    = "translation:sync"

    // 统计任务
    TypeGenerateReport   = "report:generate"
    TypeUpdateViewCount  = "stats:update_view_count"
)

// pkg/task/payloads.go

// 内容索引任务
type IndexPostPayload struct {
    PostId uint32 `json:"postId"`
    Action string `json:"action"` // create / update / delete
}

// 通知推送任务
type SendNotificationPayload struct {
    UserId  uint32 `json:"userId"`
    Title   string `json:"title"`
    Content string `json:"content"`
    Type    string `json:"type"` // comment / system / mention
}

// 数据清理任务
type CleanupExpiredDataPayload struct {
    EntityType string `json:"entityType"` // posts / comments / files
    OlderThan  int64  `json:"olderThan"`  // unix timestamp
}

// 统计任务
type UpdateViewCountPayload struct {
    PostId    uint32 `json:"postId"`
    Increment uint32 `json:"increment"`
}

func HandleIndexPost(ctx context.Context, t *asynq.Task) error {
    var payload task.IndexPostPayload
    if err := json.Unmarshal(t.Payload(), &payload); err != nil {
        return fmt.Errorf("解析载荷失败: %w", err)
    }

    switch payload.Action {
    case "create", "update":
        post, err := postRepo.Get(ctx, &contentV1.GetPostRequest{
            QueryBy: &contentV1.GetPostRequest_Id{Id: payload.PostId},
        })
        if err != nil {
            return err
        }
        return searchClient.Index(ctx, "cms_posts", post)

    case "delete":
        return searchClient.Delete(ctx, "cms_posts", fmt.Sprint(payload.PostId))
    }

    return nil
}

func HandleSendNotification(ctx context.Context, t *asynq.Task) error {
    var payload task.SendNotificationPayload
    if err := json.Unmarshal(t.Payload(), &payload); err != nil {
        return err
    }

    // 1. 创建站内信
    _, err := internalMessageRepo.Create(ctx, &internalMessageV1.CreateInternalMessageRequest{
        Data: &internalMessageV1.InternalMessage{
            UserId:  payload.UserId,
            Title:   payload.Title,
            Content: payload.Content,
            Type:    payload.Type,
        },
    })
    if err != nil {
        return err
    }

    // 2. 通过 SSE 推送实时通知
    sseServer.Broadcast(ctx, payload.UserId, map[string]any{
        "event":   "notification",
        "title":   payload.Title,
        "content": payload.Content,
    })

    return nil
}

func HandleCleanupExpiredData(ctx context.Context, t *asynq.Task) error {
    var payload task.CleanupExpiredDataPayload
    if err := json.Unmarshal(t.Payload(), &payload); err != nil {
        return err
    }

    switch payload.EntityType {
    case "posts":
        // 清理回收站中超过 30 天的帖子
        _, err := postRepo.PermanentlyDelete(ctx, payload.OlderThan)
        return err

    case "comments":
        // 清理被拒绝的评论
        _, err := commentRepo.CleanupRejected(ctx, payload.OlderThan)
        return err

    case "files":
        // 清理临时上传文件
        return cleanupTempFiles(payload.OlderThan)
    }

    return nil
}

func HandleUpdateViewCount(ctx context.Context, t *asynq.Task) error {
    var payload task.UpdateViewCountPayload
    if err := json.Unmarshal(t.Payload(), &payload); err != nil {
        return err
    }

    // 异步更新浏览量（Redis 缓存 + 定期刷入数据库）
    cacheKey := fmt.Sprintf("post:view:%d", payload.PostId)
    count, err := redisClient.IncrBy(ctx, cacheKey, int64(payload.Increment)).Result()
    if err != nil {
        return err
    }

    // 每积累 100 次浏览刷入数据库
    if count%100 == 0 {
        return postRepo.UpdateViewCount(ctx, payload.PostId, count)
    }

    return nil
}

// app/core/service/internal/service/post_service.go
func (s *PostService) Create(ctx context.Context, req *contentV1.CreatePostRequest) (*contentV1.Post, error) {
    post, err := s.postRepo.Create(ctx, req)
    if err != nil {
        return nil, err
    }

    // 投递索引任务（default 队列）
    payload, _ := json.Marshal(task.IndexPostPayload{
        PostId: post.Id,
        Action: "create",
    })
    s.asynqClient.Enqueue(
        asynq.NewTask(task.TypeIndexPost, payload),
        asynq.Queue("default"),
    )

    return post, nil
}

// 评论创建后通知文章作者
func (s *CommentService) Create(ctx context.Context, req *commentV1.CreateCommentRequest) (*commentV1.Comment, error) {
    comment, err := s.commentRepo.Create(ctx, req)
    if err != nil {
        return nil, err
    }

    // 异步通知文章作者
    payload, _ := json.Marshal(task.SendNotificationPayload{
        UserId:  post.AuthorId,
        Title:   "新评论通知",
        Content: fmt.Sprintf("您的文章收到了新评论"),
        Type:    "comment",
    })
    s.asynqClient.Enqueue(
        asynq.NewTask(task.TypeSendNotification, payload),
        asynn.Queue("critical"),  // 通知类任务高优先级
    )

    return comment, nil
}

// app/core/service/internal/server/asynq_server.go
func RegisterScheduledTasks(scheduler *asynq.Scheduler) {
    // 每天凌晨 3 点清理过期数据
    scheduler.Register("0 3 * * *", asynq.NewTask(
        task.TypeCleanupExpiredData,
        jsonMustMarshal(task.CleanupExpiredDataPayload{
            EntityType: "posts",
            OlderThan:  time.Now().AddDate(0, 0, -30).Unix(),  // 30 天前
        }),
    ))

    // 每小时同步翻译缓存
    scheduler.Register("0 * * * *", asynq.NewTask(
        task.TypeSyncTranslations,
        nil,
    ))

    // 每周一生成周报
    scheduler.Register("0 9 * * 1", asynq.NewTask(
        task.TypeGenerateReport,
        jsonMustMarshal(task.GenerateReportPayload{
            Type: "weekly",
        }),
    ))
}

s.asynqClient.Enqueue(
    asynq.NewTask(task.TypeIndexPost, payload),
    asynq.Queue("default"),
    asynq.MaxRetry(5),                    // 最多重试 5 次
    asynq.RetryDelay(30*time.Second),     // 初始延迟 30 秒
    asynq.Timeout(2*time.Minute),         // 超时 2 分钟
)