42e881576d
All checks were successful
构建并部署到测试环境（无 SSH） / build-and-deploy (push) Successful in 9m33s
- 新增网关限流：按本系统运营商账户（卡 carrier_id）→ 限流组 → 操作限制每秒请求数，每次 HTTP 尝试计量，默认限流、仅后台人工点击/通道阈值/自动续费/手动刷新豁免；支持只统计与拦截两种模式，默认只统计 - Asynq 遇限流不计失败、固定短延迟重试；轮询任务与停复机业务重试遇限流短延迟重排 - 停机、复机不再使用网关内部重试，只发一次请求 - 网关返回结构化错误；新增上游错误分类规则表与测试接口 - 新增按月分区的停复机记录表：系统自动停复机只写记录，不再写统一审计与集成日志；普通轮询查询不再写集成日志 - 新增限流组、错误分类、停复机记录、按账户调用统计与枚举后台接口 - 迁移 000243~000246；同步 OpenSpec 提案（含后续 6 个 change）与技术方案、工程约束
140 lines
3.7 KiB
Go
140 lines
3.7 KiB
Go
package queue
|
||
|
||
import (
|
||
"time"
|
||
|
||
"github.com/break/junhong_cmp_fiber/pkg/config"
|
||
"github.com/break/junhong_cmp_fiber/pkg/constants"
|
||
"github.com/hibiken/asynq"
|
||
"github.com/redis/go-redis/v9"
|
||
"go.uber.org/zap"
|
||
)
|
||
|
||
// Server Asynq Worker 服务器
|
||
type Server struct {
|
||
server *asynq.Server
|
||
logger *zap.Logger
|
||
}
|
||
|
||
// RateLimitedRetry 定义任务因运营商账户网关限流失败时的重试语义：不计为失败、按固定短延迟重试。
|
||
// Match 判断错误是否为限流拒绝；Delay 返回当前配置的固定延迟；OnDeferred 在每次因限流推迟时回调（用于累计推迟告警）。
|
||
type RateLimitedRetry struct {
|
||
Match func(error) bool
|
||
Delay func() time.Duration
|
||
OnDeferred func(task *asynq.Task)
|
||
}
|
||
|
||
// ServerOption 是 Asynq 服务器的可选配置。
|
||
type ServerOption func(*serverOptions)
|
||
|
||
type serverOptions struct {
|
||
rateLimited *RateLimitedRetry
|
||
}
|
||
|
||
// WithRateLimitedRetry 让限流错误不计入任务重试次数并按固定短延迟重试，其他错误行为不变。
|
||
func WithRateLimitedRetry(retry RateLimitedRetry) ServerOption {
|
||
return func(o *serverOptions) {
|
||
if retry.Match != nil {
|
||
o.rateLimited = &retry
|
||
}
|
||
}
|
||
}
|
||
|
||
// NewServer 创建新的 Asynq 服务器
|
||
func NewServer(redisClient *redis.Client, queueCfg *config.QueueConfig, logger *zap.Logger, options ...ServerOption) *Server {
|
||
var serverOpts serverOptions
|
||
for _, option := range options {
|
||
option(&serverOpts)
|
||
}
|
||
// 从 Redis 客户端获取配置
|
||
opts := redisClient.Options()
|
||
|
||
// 解析队列优先级配置
|
||
queues := ParseQueueConfig(queueCfg)
|
||
|
||
// 设置并发数
|
||
concurrency := queueCfg.Concurrency
|
||
if concurrency <= 0 {
|
||
concurrency = constants.DefaultConcurrency
|
||
}
|
||
|
||
asynqConfig := asynq.Config{
|
||
// 并发数
|
||
Concurrency: concurrency,
|
||
// 队列优先级配置
|
||
Queues: queues,
|
||
// 重试延迟函数（指数退避）
|
||
RetryDelayFunc: asynq.DefaultRetryDelayFunc,
|
||
// 是否记录详细日志
|
||
LogLevel: asynq.WarnLevel,
|
||
}
|
||
if retry := serverOpts.rateLimited; retry != nil {
|
||
// 限流错误不是任务本身的失败：IsFailure 返回 false 时 Asynq 不增加重试计数，任务不会因限流被归档；
|
||
// 重试间隔固定为短延迟，避免按指数退避把"流量用完停机"等推迟数小时。
|
||
asynqConfig.IsFailure = func(err error) bool {
|
||
return !retry.Match(err)
|
||
}
|
||
asynqConfig.RetryDelayFunc = func(n int, err error, task *asynq.Task) time.Duration {
|
||
if retry.Match(err) {
|
||
if retry.OnDeferred != nil {
|
||
retry.OnDeferred(task)
|
||
}
|
||
if retry.Delay != nil {
|
||
if delay := retry.Delay(); delay > 0 {
|
||
return delay
|
||
}
|
||
}
|
||
return 5 * time.Second
|
||
}
|
||
return asynq.DefaultRetryDelayFunc(n, err, task)
|
||
}
|
||
}
|
||
|
||
// 创建 Asynq 服务器配置
|
||
asynqServer := asynq.NewServer(
|
||
asynq.RedisClientOpt{
|
||
Addr: opts.Addr,
|
||
Password: opts.Password,
|
||
DB: opts.DB,
|
||
},
|
||
asynqConfig,
|
||
)
|
||
|
||
return &Server{
|
||
server: asynqServer,
|
||
logger: logger,
|
||
}
|
||
}
|
||
|
||
// Start 启动 Worker 服务器
|
||
func (s *Server) Start(mux *asynq.ServeMux) error {
|
||
s.logger.Info("Worker 服务器启动中...")
|
||
|
||
if err := s.server.Start(mux); err != nil {
|
||
s.logger.Error("Worker 服务器启动失败", zap.Error(err))
|
||
return err
|
||
}
|
||
|
||
s.logger.Info("Worker 服务器启动成功")
|
||
return nil
|
||
}
|
||
|
||
// Shutdown 优雅关闭服务器
|
||
func (s *Server) Shutdown() {
|
||
s.logger.Info("Worker 服务器关闭中...")
|
||
s.server.Shutdown()
|
||
s.logger.Info("Worker 服务器已关闭")
|
||
}
|
||
|
||
// Run 启动并阻塞运行（用于主函数）
|
||
func (s *Server) Run(mux *asynq.ServeMux) error {
|
||
s.logger.Info("Worker 服务器启动中...")
|
||
|
||
if err := s.server.Run(mux); err != nil {
|
||
s.logger.Error("Worker 服务器运行失败", zap.Error(err))
|
||
return err
|
||
}
|
||
|
||
return nil
|
||
}
|