// asynq_service.go package service import ( "encoding/json" "tydata-server/app/main/api/internal/config" "tydata-server/app/main/api/internal/types" "github.com/hibiken/asynq" "github.com/zeromicro/go-zero/core/logx" ) type AsynqService struct { client *asynq.Client config config.Config } // NewAsynqService 创建并初始化 Asynq 客户端 func NewAsynqService(c config.Config) *AsynqService { client := asynq.NewClient(asynq.RedisClientOpt{ Addr: c.CacheRedis[0].Host, Password: c.CacheRedis[0].Pass, }) return &AsynqService{client: client, config: c} } // Close 关闭 Asynq 客户端 func (s *AsynqService) Close() error { return s.client.Close() } func (s *AsynqService) SendQueryTask(orderID int64) error { // 准备任务的 payload payload := types.MsgPaySuccessQueryPayload{ OrderID: orderID, } payloadBytes, err := json.Marshal(payload) if err != nil { logx.Errorf("发送异步任务失败 (无法编码 payload): %v, 订单号: %d", err, orderID) return err // 直接返回错误,避免继续执行 } options := []asynq.Option{ asynq.MaxRetry(5), // 设置最大重试次数 } // 创建任务 task := asynq.NewTask(types.MsgPaySuccessQuery, payloadBytes, options...) // 将任务加入队列并获取任务信息 info, err := s.client.Enqueue(task) if err != nil { logx.Errorf("发送异步任务失败 (加入队列失败): %+v, 订单号: %d", err, orderID) return err } // 记录成功日志,带上任务 ID 和队列信息 logx.Infof("发送异步任务成功,任务ID: %s, 队列: %s, 订单号: %d", info.ID, info.Queue, orderID) return nil }