Files

T

ryan 1f4fe2fb75 docs: add RabbitMQ subscriber design spec

Lock the notify-service MQ module: configurable subscriptions, trade-signal formatting, and shared NotifyService.

2026-08-15 16:42:30 +08:00

8.2 KiB

Raw Blame History

RabbitMQ 订阅模块设计

Date: 2026-08-15
Status: Approved

Goal

在通知服务内增加可配置的 RabbitMQ 订阅：消费交易信号、格式化文案、再走现有规则/模板/渠道发送。第一路订阅 fanout exchange trade.signal.executor.queue，与 executor 队列独立消费、互不争抢。配置做成订阅列表，后续加 exchange 只加配置、不改代码。

Non-goals

移植参考项目的钉钉直发、按目标过滤、x-dingtalk-sent / x-dingtalk-abandoned
在 MQ 层等待渠道发送结果
均价持久化（进程内即可，重启丢失）
连真实 CloudAMQP 的集成测试
把 AMQP 账号写进仓库
新增管理 API 或数据库表
订阅侧解析 Source 的 parse_mode（MQ 路径自带 event + data）

Decisions

项	选择
进 notify 的方式	抽出内部 `NotifyService`，MQ 与 HTTP 共用
订阅 ↔ Source	每条订阅绑定一个 `source` 名
event	`trade.<action小写>`，如 `OPEN` → `trade.open`
格式化	移植交易信号格式化器；`data.formatted` + 原始字段
均价 / 策略覆盖	都移植；均价按 `strategyCode + symbol + side` 进程内追踪
周期	MQ `period` 原样使用（如 `1h`），不换算
过滤	交给 Rule 条件（`strategyCode` / `symbol` 等）
凭据	`url: "${RABBITMQ_URL}"`

Architecture

进程内 HTTP 与 MQ 消费并行。两者都进入同一个内部入口，规则、模板、渠道只维护一份。

RabbitMQ fanout exchange
        │
        ├─ executor 队列（已有，不改）
        └─ trade.signal.notify.queue（本服务独占）
                │
                ▼
        Subscriber（声明/绑定/重连/消费/重试/DLQ）
                │
                ▼
        trade_signal：解析 → 策略覆盖 → 均价 → 格式化
                │
                ▼
        NotifyService
          匹配规则 → 条件过滤 → 渲染模板 → 异步发渠道

订阅列表为空，或某条订阅的 url 未展开到非空值：该条不启动。全部未启动时 HTTP 通知不受影响。

Config

config/config.yaml 增加 subscriptions 列表。第一路示例（URL 必须走环境变量）：

subscriptions:
  - name: trade-signal
    url: "${RABBITMQ_URL}"
    queue: trade.signal.notify.queue
    dead_letter_queue: trade.signal.notify.dlq
    exchange: trade.signal.executor.queue
    exchange_type: fanout
    routing_key: ""
    max_retry: 3
    source: trade-signal
    formatter: trade_signal
    strategy_overrides:
      BLONG:
        quantity_multipliers:
          open: 100
          add: 100
          reduce: 100
          close: 100
        leverage: 100

字段	说明
`name`	订阅标识，日志用；缺省用 `queue`
`url`	AMQP(S) URL，支持 `${ENV}`
`queue`	本服务独占队列
`dead_letter_queue`	超过 `max_retry` 后投入；空则丢弃并 Ack
`exchange` / `exchange_type` / `routing_key`	声明并绑定；`exchange` 为空则只声明队列
`max_retry`	`Process` 内部错误的重投次数；`<=0` 时默认 3
`source`	已有 `notification_source.name`
`formatter`	空则默认 `trade_signal`；未知值启动时报错
`strategy_overrides`	按 `strategyCode`；数量倍数 `<=0` 视为 1

exchange_type 为空时默认 fanout。

Modules

模块	职责	接口
`internal/notify`	匹配规则、条件、渲染、路由、记 message_log	`Process(ctx, Request) (Result, error)`
`internal/subscriber`	连 MQ、声明/绑定、重连、重试/DLQ；按订阅启动 goroutine	`Run(ctx)`，依赖 `Process`
`internal/subscriber/tradesignal`	解析信号、策略覆盖、均价、格式化	输入 JSON body，输出 `event` + `data`

POST /api/v1/notify 按 Source parse_mode 解析 body 后调用 notify.Process，不再内嵌匹配/渲染/路由。

subscriber 不碰渠道，不读 parse_mode。

NotifyService

type Request struct {
    Source *model.Source
    Event  string
    Data   map[string]interface{}
}

type Result struct {
    Matched  bool
    Filtered bool
    Channels []string
    Reason   string
}

行为与现有 Handler 一致：无规则 → Matched=false；条件未过 → Matched=true, Filtered=true；命中则渲染、Route、写 log。标题仍为 {source.name}: {event}。Process 只在查库/渲染等内部失败时返回 error。

Source 查找

订阅用 source 名调用已有 GetSourceByName。status != 1 或未找到：Ack，记日志，不进 DLQ。

Data flow（trade_signal）

消费 delivery，JSON 反序列化为信号。
按 strategyCode 套 strategy_overrides（数量倍数、杠杆覆盖）。
按 strategyCode + symbol + side 更新进程内均价；同一 signalId 只应用一次。
生成 formatted 文本。
event = "trade." + strings.ToLower(action)。action 为空视为无效消息。
用订阅的 source 名取启用 Source，调用 Process。

`data` 字段

data 来自覆盖后的信号（与 formatted 数字一致），camelCase，并附加：

键	来源
`signalId` / `sourcePosId` / `strategyCode` / `symbol` / `side` / `action`	信号
`quantity` / `amountMarginRatio` / `posMarginRatio`	信号（quantity 可能被倍数改写）
`price` / `leverage` / `period` / `eventTime`	信号（leverage 可能被覆盖）
`takeProfitPrice` / `stopLossPrice` / `takeProfitRatio` / `stopLossRatio`	信号
`pnl` / `accountBalance`	信号（可选）
`formatted`	格式化全文
`avgPrice`	有均价才写入

规则条件可写 strategyCode、symbol、period 等。模板可用 {{.formatted}}，也可自己拼字段。

格式化文案

移植参考项目 test-mq-to-ali 的格式化器，并在「交易品种」后增加周期（period 非空才输出）：

多单开仓
交易品种: BTC
周期: 1h
开仓价格: 65000.00
...
策略: BLONG
Time: 2026.08.15 16:39:00

其余规则与参考项目一致：品种去掉报价后缀（BTCUSDT → BTC）；开/加/减/平仓标题与数量或比例行；可选均价、杠杆、止盈止损、盈亏、余额；时间为本地时区 2006.01.02 15:04:05。

Error handling

情况	处理
JSON 无效、缺 `action`	Ack 丢掉，不重试
Source 不存在/禁用、无匹配规则、条件未过	Ack，记日志，不进 DLQ
`Process` 返回 error（查库失败等）	`x-retry-count` +1 后重新投递本队列；超过 `max_retry` 则投入 `dead_letter_queue`（未配置则丢弃并 Ack）
连接/channel 断开	5s 后重连，重新 `consumeOnce`

启动时声明 durable exchange（若配置了）、durable 队列、绑定、durable DLQ。消费 autoAck=false，QoS prefetch=1。consumer tag 用订阅 name。

渠道发送仍由 engine.Router 异步重试（1s / 5s / 30s）。MQ 在 Process 接受路由后即 Ack，不等待渠道结果。

不移植参考项目按钉钉目标的部分成功重试。

Testing

格式化：开/平/加/减仓文案；period 有则出现、空则省略；均价行；策略倍数后的数量
均价追踪：OPEN/ADD 更新，REDUCE/CLOSE 展示原均价；同 signalId 不重复应用
策略覆盖：数量倍数、杠杆覆盖；<=0 的倍数视为 1
订阅 → Notify：假 Process，断言 event、data.formatted、原始字段与 period
重试/DLQ：超过 max_retry 才进 DLQ；坏 JSON 直接 Ack
HTTP /notify：抽 Process 后现有解析与匹配行为不变

不写连真实 broker 的测试。声明/绑定逻辑用可注入的 channel 假对象，或抽纯函数测重试计数与 DLQ 判定。

Boot

cmd/server/main.go 在 HTTP server 启动后、等信号退出前，对每条有效订阅 go subscriber.Run(ctx)。收到 SIGINT/SIGTERM 时 cancel 该 ctx，再 Shutdown HTTP。

运营侧需事先创建 Source（name 与订阅 source 一致）、Template、Rule（event 为 trade.open 等）、Channel。本模块不自动建这些记录。

Out of scope later

第二种 formatter（有新 exchange 再加）
均价写入 Redis
管理 API 热更新订阅

8.2 KiB Raw Blame History Unescape Escape