基于 Redis 的通用接口排队服务 🪜

type TaskRequestDetails struct {
    TargetURL     string              `json:"target_url"`     // 业务接口 URL
    ExtractedPath string              `json:"extracted_path"` // 提取的路径
    HTTPMethod    string              `json:"http_method"`    // HTTP 方法
    Headers       map[string][]string `json:"headers"`        // 请求头
    RequestBody   string              `json:"request_body"`   // 请求体
    QueryParams   url.Values          `json:"query_params"`   // 路由参数或查询参数
}

type TaskExecutionStatus struct {
    Status    string    `json:"status"`     // 任务的当前状态（如 "pending", "processing", "completed", "failed" 等）
    Result    string    `json:"result"`     // 接口的响应内容，包括成功结果或错误信息，作为 JSON 字符串存储
    CreatedAt time.Time `json:"created_at"` // 任务的创建时间，即任务被添加到队列的时间
    StartedAt time.Time `json:"started_at"` // 任务的开始时间，即任务正式处理的时间
    EndedAt   time.Time `json:"ended_at"`   // 任务的结束时间，任务完成或失败的时间
}

type Task struct {
    UserId       int64               `json:"user_id"`        // 用户 ID
    TaskId       string              `json:"taskID"`         // 任务的唯一标识符
    TaskQueueKey string              `json:"task_queue_Key"` // 任务队列的唯一标识符
    Priority     int64               `json:"priority"`       // 任务的优先级
    Type         string              `json:"type"`           // 任务类型，如果是 offline 类型的任务，需要保存任务请求信息
    Request      TaskRequestDetails  `json:"task_request"`   // 任务请求信息
    Execution    TaskExecutionStatus `json:"task_status"`    // 任务状态信息
}

    // 触发限流
    if limited {
       // 任务放回队列
       _, err := z.Redis.ZAdd(ctx, task.TaskQueueKey, redis.Z{
          Score:  res.Score,
          Member: res.Member,
       }).Result()
       if err != nil {
          logx.Errorf("触发限流，任务还原失败: %v，被丢弃", err)
          // 从任务状态 String 中删除任务状态，防止状态泄露
          z.Redis.Del(ctx, taskKey)
          continue
       }
       retryCount := 0
       maxRetryCount := 5 // 最大重试次数
       baseWaitTime := time.Second
       for retryCount < maxRetryCount {
          // 计算指数退避时间，最大为 TaskInterval
          backoffTime := time.Duration(float64(baseWaitTime) * math.Pow(2, float64(retryCount)))
          if backoffTime > TaskInterval {
             backoffTime = TaskInterval
          }
          // 退避一段时间，等待下一个窗口
          logx.Infof("限流触发，等待 %v 后重试，任务: %s", backoffTime, taskKey)
          time.Sleep(backoffTime)
          // 重新检查限流状态
          //limited, err = z.RateLimiter.LimitDefault(ctx, rateLimitedKey)
          now := time.Now()
          limited, err = z.RateLimiter.Limit(ctx, rateLimitedKey, interval, rate, now)
          logx.Infof("重新检查限流状态: %v", limited)
          if err != nil {
             logx.Errorf("限流判断失败: %v", err)
             break
          }
          // 如果不再限流，退出重试循环
          // 这边要注意，就算不再限流了，外层可能会因此又触发限流
          if !limited {
             // 删除内层循环检查限流为 false 情况时产生的元素
             z.Redis.ZRem(ctx, rateLimitedKey, now.UnixMilli())
             break
          }
          // 增加重试次数，准备下一次退避
          retryCount++
       }
       continue
    }
    go func(task Task) {
       // 执行对应的任务
       err = z.Execute(&task)
       if err != nil {
          logx.Errorf("执行任务失败: %v", err)
       }
    }(task)
}

SubmitReq {
    TargetUrl  string `json:"target_url"` // 目标接口地址，完整的 url
    HttpMethod string `json:"http_method"` // 请求方法，目前只考虑了 GET POST
    Body       string `json:"body"` // post 请求体 json 字符串，注意是带转义字符的字符串
    Priority   int64  `json:"priority"` // 优先级 ＞ 0，数字越大优先级越高，例如,普通用户 0 VIP 2 SVIP 3
    Timestamp  int64  `json:"timestamp"` // 请求时间戳
    TaskType    string `json:"task_type"`  // "online" 在线排队 "offline" 离线排队
}
SubmitResp {
    TaskId     string `json:"task_id"` // 任务Id
    TaskStatus string `json:"task_status"` // 任务状态
    CreatedAt  string `json:"created_at"` // 任务创建时间
    WaitNum    int64  `json:"wait_num"` // 排队人数
}

StatusReq {
    TaskId string `form:"task_id"` // 任务Id
}
StatusResp {
    TaskStatus string `json:"task_status"` // 任务状态
    TaskResult string `json:"task_result"` // 任务结果，对应接口的响应体 json
    WaitNum    int64  `json:"wait_num"` // 排队人数
    WaitTime   int64  `json:"wait_time"` // 预计等待时间
    CreatedAt  string `json:"created_at"` // 任务创建时间
    StartedAt  string `json:"started_at"` // 任务开始时间
    EndedAt    string `json:"ended_at"` // 任务结束时间
}

CancelReq {
    TaskId string `json:"task_id"` // 任务Id
}
CancelResp {
    TaskId     string `json:"task_id"` // 任务Id
    TaskStatus string `json:"task_status"` // 任务状态
}