huang
984ccccc63
在项目宪章中新增第九条原则"数据库设计原则"，明确禁止使用数据库外键约束和ORM关联标签。 主要变更： - 新增原则IX：数据库设计原则（Database Design Principles） - 强制要求：数据库表不得使用外键约束 - 强制要求：GORM模型不得使用ORM关联标签（foreignKey、hasMany等） - 强制要求：表关系必须通过ID字段手动维护 - 强制要求：关联数据查询必须显式编写，避免ORM魔法 - 强制要求：时间字段由GORM处理，不使用数据库触发器 设计理念： - 提升业务逻辑灵活性（无数据库约束限制） - 优化高并发性能（无外键检查开销） - 增强代码可读性（显式查询，无隐式预加载） - 简化数据库架构和迁移流程 - 支持分布式和微服务场景 版本升级：2.3.0 → 2.4.0（MINOR）
89 lines
2.0 KiB
Go
89 lines
2.0 KiB
Go
package queue
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
|
|
"github.com/break/junhong_cmp_fiber/pkg/config"
|
|
"github.com/bytedance/sonic"
|
|
"github.com/hibiken/asynq"
|
|
"github.com/redis/go-redis/v9"
|
|
"go.uber.org/zap"
|
|
)
|
|
|
|
// Client Asynq 任务提交客户端
|
|
type Client struct {
|
|
client *asynq.Client
|
|
logger *zap.Logger
|
|
}
|
|
|
|
// NewClient 创建新的 Asynq 客户端
|
|
func NewClient(redisClient *redis.Client, logger *zap.Logger) *Client {
|
|
// 从 Redis 客户端获取配置
|
|
opts := redisClient.Options()
|
|
|
|
asynqClient := asynq.NewClient(asynq.RedisClientOpt{
|
|
Addr: opts.Addr,
|
|
Password: opts.Password,
|
|
DB: opts.DB,
|
|
})
|
|
|
|
return &Client{
|
|
client: asynqClient,
|
|
logger: logger,
|
|
}
|
|
}
|
|
|
|
// EnqueueTask 提交任务到队列
|
|
func (c *Client) EnqueueTask(ctx context.Context, taskType string, payload interface{}, opts ...asynq.Option) error {
|
|
// 序列化载荷
|
|
payloadBytes, err := sonic.Marshal(payload)
|
|
if err != nil {
|
|
c.logger.Error("任务载荷序列化失败",
|
|
zap.String("task_type", taskType),
|
|
zap.Error(err))
|
|
return fmt.Errorf("failed to marshal task payload: %w", err)
|
|
}
|
|
|
|
// 创建任务
|
|
task := asynq.NewTask(taskType, payloadBytes, opts...)
|
|
|
|
// 提交任务
|
|
info, err := c.client.EnqueueContext(ctx, task)
|
|
if err != nil {
|
|
c.logger.Error("任务提交失败",
|
|
zap.String("task_type", taskType),
|
|
zap.Error(err))
|
|
return fmt.Errorf("failed to enqueue task: %w", err)
|
|
}
|
|
|
|
c.logger.Info("任务已提交",
|
|
zap.String("task_id", info.ID),
|
|
zap.String("task_type", taskType),
|
|
zap.String("queue", info.Queue),
|
|
zap.Int("max_retry", info.MaxRetry))
|
|
|
|
return nil
|
|
}
|
|
|
|
// Close 关闭客户端
|
|
func (c *Client) Close() error {
|
|
if c.client != nil {
|
|
return c.client.Close()
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// ParseQueueConfig 解析队列配置为 Asynq 格式
|
|
func ParseQueueConfig(cfg *config.QueueConfig) map[string]int {
|
|
if cfg.Queues != nil && len(cfg.Queues) > 0 {
|
|
return cfg.Queues
|
|
}
|
|
// 默认队列优先级
|
|
return map[string]int{
|
|
"critical": 6,
|
|
"default": 3,
|
|
"low": 1,
|
|
}
|
|
}
|