package task import ( "context" "encoding/json" "fmt" "time" "github.com/hibiken/asynq" "go.uber.org/zap" ) // AsynqClient Asynq 客户端 type AsynqClient struct { client *asynq.Client logger *zap.Logger } // NewAsynqClient 创建 Asynq 客户端 func NewAsynqClient(redisAddr string, logger *zap.Logger) *AsynqClient { client := asynq.NewClient(asynq.RedisClientOpt{Addr: redisAddr}) return &AsynqClient{ client: client, logger: logger, } } // Close 关闭客户端 func (c *AsynqClient) Close() error { return c.client.Close() } // ScheduleArticlePublish 调度文章定时发布任务 func (c *AsynqClient) ScheduleArticlePublish(ctx context.Context, articleID string, publishTime time.Time) (string, error) { payload := map[string]interface{}{ "article_id": articleID, } payloadBytes, err := json.Marshal(payload) if err != nil { c.logger.Error("序列化任务载荷失败", zap.Error(err)) return "", fmt.Errorf("创建任务失败: %w", err) } task := asynq.NewTask(TaskTypeArticlePublish, payloadBytes) // 计算延迟时间 delay := publishTime.Sub(time.Now()) if delay <= 0 { return "", fmt.Errorf("定时发布时间不能早于当前时间") } // 设置任务选项 opts := []asynq.Option{ asynq.ProcessIn(delay), asynq.MaxRetry(3), asynq.Timeout(5 * time.Minute), } info, err := c.client.Enqueue(task, opts...) if err != nil { c.logger.Error("调度定时发布任务失败", zap.String("article_id", articleID), zap.Time("publish_time", publishTime), zap.Error(err)) return "", fmt.Errorf("调度任务失败: %w", err) } c.logger.Info("定时发布任务调度成功", zap.String("article_id", articleID), zap.Time("publish_time", publishTime), zap.String("task_id", info.ID)) return info.ID, nil } // CancelScheduledTask 取消已调度的任务 func (c *AsynqClient) CancelScheduledTask(ctx context.Context, taskID string) error { // 注意:Asynq不直接支持取消已调度的任务 // 这里我们记录日志,实际取消需要在数据库中标记 c.logger.Info("请求取消定时任务", zap.String("task_id", taskID)) // 在实际应用中,你可能需要: // 1. 在数据库中标记任务为已取消 // 2. 在任务执行时检查取消状态 // 3. 或者使用Redis的TTL机制 return nil } // RescheduleArticlePublish 重新调度文章定时发布任务 func (c *AsynqClient) RescheduleArticlePublish(ctx context.Context, articleID string, oldTaskID string, newPublishTime time.Time) (string, error) { // 1. 取消旧任务(标记为已取消) if err := c.CancelScheduledTask(ctx, oldTaskID); err != nil { c.logger.Warn("取消旧任务失败", zap.String("old_task_id", oldTaskID), zap.Error(err)) } // 2. 创建新任务 newTaskID, err := c.ScheduleArticlePublish(ctx, articleID, newPublishTime) if err != nil { return "", fmt.Errorf("重新调度任务失败: %w", err) } c.logger.Info("重新调度定时发布任务成功", zap.String("article_id", articleID), zap.String("old_task_id", oldTaskID), zap.String("new_task_id", newTaskID), zap.Time("new_publish_time", newPublishTime)) return newTaskID, nil }