x_admin

异步队列 ​

后端异步任务队列位于 core/queue，提供与业务解耦的「点对点工作队列」，底层内置 Redis 后端。典型用途：AI 处理、流程邮件/Webhook 通知、邮箱验证码、通知邮件补推。

一、设计目标 ​

  • 统一接口：调用方只依赖 queue.Queue 接口，不感知底层后端。
  • 可插拔后端：内置 redis 实现，新增后端只需实现 Queue 接口并在 config.New 注册。
  • 无业务语义：消息体统一为 []byte，由业务层自行序列化（JSON 等）。
  • 点对点语义：同一条消息只会被一个消费者取出（多 worker 竞争消费）。广播语义请使用 core/pubsub。

二、接口定义 ​

go
// core/queue/interface.go
type Handler func(ctx context.Context, body []byte) error

type Queue interface {
    Enqueue(name string, body any) error                                   // 投递消息
    Consume(ctx context.Context, name string, handler Handler, concurrency int) error // 启动消费者
    Close() error
}
  • Enqueue：任意实例投递的任务可被任一实例的 worker 取到（适合多实例部署）。
  • Consume：concurrency 为 worker 并发数（<=0 回落为 1）；ctx 取消后所有 worker 优雅退出。
  • Handler 返回 error 仅记录日志，当前实现不自动重试（重试/死信队列需业务层自行实现）。

三、Redis 后端实现 ​

位置：core/queue/redis.go，基于 Redis List：

  • 投递：RPUSH 追加到队列尾部。
  • 消费：单连接 BRPOP 阻塞拉取，投入有界 channel（cap = 并发数），再由 concurrency 个 worker 并发处理。
    • 有界 channel 提供天然背压：channel 满则阻塞取消息协程，避免内存无限堆积。
    • 仅占用 1 条 Redis 连接（而非 concurrency 条），大幅减少连接数。
    • Redis 保证同一消息只被一个 worker 取到。
  • 网络抖动时短暂退避（100ms）重试，避免 CPU 空转。
  • ctx 取消后先停止取消息，再 drain 在途任务后退出。
go
core.Queue.Enqueue("transcribe", payload)          // 投递
core.Queue.Consume(ctx, "transcribe", handler, 3)  // 3 并发消费

四、消费者注册（app/queue） ​

所有消费者集中在 app/queue/start.go 的 Start(ctx) 中统一注册并启动，由 main.go 在进程启动时调用。当前已注册队列：

队列名并发处理内容
flow_notify_email3流程邮件通知（投递体为 util.EmailOptions，消费时 json.Unmarshal 后 util.EmailUtil.SendEmail）
flow_notify_webhook3流程 Webhook 通知（投递体 flow_schema.FlowNotifyWebhookPayload，调用 FlowHistoryService.ProcessFlowWebhook）
email:code:send（util.QueueEmailCode）3邮箱验证码发送，投递体 util.EmailCodeTask；入队超过 60 秒则丢弃（验证码时效性）
notice:email（notice_service.QueueNoticeEmail）3通知邮件补推，投递体 notice_service.NoticeEmailTask；失败/成功均回写状态（2 成功 / 3 失败），不重试
notice:email:delay（notice_service.QueueNoticeEmailDelay）3通知邮件延迟补推，投递体 queue_schema.NoticeEmailDelayTask（receiver_id/preview_count）；由 core.QueueDelay.ConsumeDelayed 消费，延迟 EmailDelaySeconds 后合成邮件再入 notice:email
image_webp3图片异步转 webp，投递体 queue_schema.ImageWebpPayload（file_id），转码完成后回写文件记录，失败则标记原图
operate_log32操作日志落库，投递体 queue_schema.OperateLogPayload；由中间件投递，defer 兜底写库，消费幂等

示例（来自 app/queue/start.go）：

go
// app/queue/start.go
func Start(ctx context.Context) {
    core.Queue.Consume(ctx, queue_schema.QueueFlowNotifyEmail, flowNotifyEmailHandler, 3)
    core.Queue.Consume(ctx, queue_schema.QueueFlowNotifyWebhook, flowNotifyWebhookHandler, 3)
    core.Queue.Consume(ctx, util.QueueEmailCode, emailCodeHandler, 3)
    core.Queue.Consume(ctx, queue_schema.QueueNoticeEmail, noticeEmailHandler, 3)
    core.Queue.Consume(ctx, queue_schema.QueueImageWebp, imageWebpHandler, 3)
    core.Queue.Consume(ctx, queue_schema.QueueOperateLog, operateLogHandler, 3)
    // 延迟队列（与即时队列独立，基于 core/queue_delay）
    core.QueueDelay.ConsumeDelayed(ctx, queue_schema.QueueNoticeEmailDelay, noticeEmailDelayHandler, 3)
}

五、扩展新的异步任务 ​

  1. 定义 payload：在对应 schema 包定义结构体（或直接复用 util.EmailOptions 等）。
  2. 投递：业务代码中 core.Queue.Enqueue("your_queue", payload)。
  3. 注册消费者：在 app/queue/start.go 的 Start 中 core.Queue.Consume(ctx, "your_queue", handler, N)。
  4. 无需改动 core/queue 本身——队列不关心业务类型，仅传输字节。

六、与 pubsub 的区别 ​

能力core/queuecore/pubsub
语义点对点（一条消息一个消费者）广播（所有订阅者都收到）
用途异步作业、后台任务实时事件通知（如 WebSocket 跨实例推送）
后端Redis ListRedis Pub/Sub

详见 邮件发送 与「异步队列与核心组件」中的 pubsub 说明。