商品报价与 Mongo 读模型：架构、MQ 链路与高可用演进

这是一份双层文档：前半部分供领导在 5 分钟内理解业务、风险和改造顺序；后半部分供研发、测试和运维沿入口、字段、日志和代码证据继续下钻。

一张图看懂当前业务

flowchart LR
    classDef source fill:#fff2cc,stroke:#b8860b,color:#222;
    classDef dgj fill:#d9eaff,stroke:#3977b8,color:#111;
    classDef event fill:#ffd6d6,stroke:#bb3030,color:#111;
    classDef fact fill:#dff4df,stroke:#3c8a3c,color:#111;
    classDef projection fill:#ffe3bf,stroke:#c66a00,color:#111;
    classDef downstream fill:#eadcff,stroke:#7546a8,color:#111;

    subgraph CHANGE["业务变化来源"]
        PAGE["商品管理页面<br/>新增、编辑、删除、启停、导入"]:::source
        ITEM["ItemCenter<br/>价格、限采、销售限价"]:::source
        OPS["OpsCenter / GLB OMS<br/>货主销售区域"]:::source
        INV["库存事实<br/>inventory_real_time"]:::fact
    end

    subgraph EXEC["DGJ2 写入入口"]
        HTTP["6 个 HTTP<br/>Inventory / Invlocation"]:::dgj
        MQ["3 类 MQ 事件<br/>ItemCenterNotify / OpsCenterNotify"]:::event
        JOB["1 个后置 Job<br/>ItemCenterJob"]:::dgj
        CLI["2 个主同步 CLI<br/>商品快照 / 库存增量"]:::dgj
        FIX["6 个修复入口<br/>索引、价格、限采、描述"]:::dgj
        WRITER["统一 Mongo 写入层<br/>MaterielCacheStationSer"]:::dgj
    end

    subgraph DATA["事实与投影"]
        FACTS[("MySQL / ItemCenter<br/>商品、库存、价格与规则事实")]:::fact
        MONGO[("Mongo<br/>dgj_goods_sid_{sid}<br/>报价商品读模型")]:::projection
        DERIVED[("Elasticsearch / Redis<br/>搜索投影与差集基线")]:::projection
    end

    subgraph QUOTE["报价读取链"]
        BASE["Basedata<br/>站点商品"]:::downstream
        SALE["Sale Service<br/>报价聚合"]:::downstream
        VIN["VinAudit<br/>VIN / 车型候选"]:::downstream
        PURCHASE["Purchase<br/>在途数量"]:::downstream
        PRICE["Price API<br/>客户价 / 指导价"]:::downstream
        UI["商品报价页面<br/>基础信息 + 价格 + 库存"]:::source
    end

    PAGE --> HTTP
    ITEM --> MQ
    OPS --> MQ
    INV --> CLI
    HTTP --> FACTS
    MQ -->|"价格 / 限采先落消息表"| JOB
    MQ -->|"区域 / 限价直接同步"| WRITER
    JOB --> FACTS
    CLI --> FACTS
    FIX --> WRITER
    FACTS --> HTTP
    FACTS --> CLI
    HTTP --> WRITER
    JOB --> WRITER
    CLI --> WRITER
    WRITER ==>|"按 SID 写商品快照"| MONGO
    WRITER -. "同步或重建" .-> DERIVED
    MONGO --> BASE --> SALE --> UI
    VIN --> SALE
    PURCHASE --> SALE
    PRICE --> SALE

四个汇报结论

1. 商品报价是 DGJ2、Sale、Basedata、VinAudit、Purchase、ItemCenter 和价格接口共同组成的跨微服务聚合链，不是单体页面查询。
2. dgj_goods_sid_{sid} 是为报价读取构建的 Mongo 商品快照，不是最终事实源；MySQL、ItemCenter 和库存事实决定正确结果。
3. 已从当前代码确认 6 个 HTTP、3 类 MQ、1 个后置 Job、2 个主同步 CLI、6 个修复入口会直接或间接影响商品 Mongo。
4. 高可用建设应按“先可观测、再可靠、再解耦、最后治理历史脚本”推进，而不是继续增加无审计的单点修复入口。

当前梳理形成的业务能力

报价同步主链

sequenceDiagram
    participant UI as "商品报价页面"
    participant DGJ as "DGJ2 /sale/Offer/searchList"
    participant Sale as "Sale Service /offer/search/list"
    participant Vin as "VinAudit"
    participant Base as "Basedata"
    participant Mongo as "Mongo dgj_goods_sid_{sid}"
    participant Inv as "Inventory Center"
    participant Pur as "Purchase Service"
    participant Price as "Price API"

    UI->>DGJ: VIN + customer_id + category_id + page
    DGJ->>Sale: 转发并补充登录态
    Sale->>Vin: 查询 VIN / 车型候选 SKU
    Sale->>Base: 查询 SID 商品与状态
    Base->>Mongo: 读取基础字段与状态投影
    Sale->>Inv: 查询可用/优良品库存
    Sale->>Pur: 查询在途和未交数量
    Sale->>Price: 查询客户价、指导价、现金价
    Mongo-->>Base: 商品读模型
    Vin-->>Sale: 车型候选
    Base-->>Sale: 站点商品
    Inv-->>Sale: 库存
    Pur-->>Sale: 在途
    Price-->>Sale: 价格金额
    Sale-->>DGJ: 商品 + 库存 + 在途 + 价格 + 标签
    DGJ-->>UI: 报价列表

POST /sale/Offer/searchList 本身不写商品 Mongo。它读取已经维护好的商品快照，并聚合多系统结果；因此 Mongo 正确不代表最终价格、库存或在途一定正确。

三条 MQ 的业务差异

flowchart TB
    classDef source fill:#fff2cc,stroke:#b8860b,color:#222;
    classDef dgj fill:#d9eaff,stroke:#3977b8,color:#111;
    classDef mq fill:#ffd6d6,stroke:#bb3030,color:#111;
    classDef db fill:#dff4df,stroke:#3c8a3c,color:#111;
    classDef projection fill:#ffe3bf,stroke:#c66a00,color:#111;

    subgraph P["MQ-01｜价格生效 / 失效"]
        direction LR
        P1["ItemCenter<br/>价格配置变化"]:::source --> P2(("RabbitMQ")):::mq
        P2 --> P3["ItemCenterNotify<br/>保存 SID + 消息类型"]:::dgj
        P3 --> P4[("t_bs_mq_message<br/>WAIT")]:::db
        P4 --> P5["ItemCenterJob<br/>领取并合并 SID"]:::dgj
        P5 --> P6["findPriceSkuList<br/>当前有价 SKU"]:::source
        P6 --> P7["Mongo / ES / Redis<br/>noPricingFlag"]:::projection
    end

    subgraph R["MQ-02｜货主销售区域变化"]
        direction LR
        R1["OpsCenter / GLB OMS<br/>区域配置变化"]:::source --> R2(("RabbitMQ")):::mq
        R2 --> R3["OpsCenterNotify<br/>Consumer 内直接处理"]:::dgj
        R3 --> R4["AdminModel<br/>归一化主站 SID"]:::dgj
        R4 --> R5["findPriceSkuList<br/>重新计算适用 SKU"]:::source
        R5 --> R6["Mongo / ES / Redis<br/>noPricingFlag"]:::projection
    end

    subgraph L["MQ-03｜指导价 / 销售限价变化"]
        direction LR
        L1["OpsManager → ItemCenter<br/>限价规则变化"]:::source --> L2(("RabbitMQ")):::mq
        L2 --> L3["ItemCenterNotify<br/>Consumer 内直接处理"]:::dgj
        L3 --> L4[("MySQL<br/>销售限价关系")]:::db
        L4 --> L5["Mongo / ES<br/>is_hot / is_limit_price"]:::projection
    end


    P7 ~~~ R1
    R6 ~~~ L1

区域变化最终更新 noPricingFlag，是因为区域属于价格适用条件：DGJ2 重新拉取 ItemCenter 当前有价 SKU，仍有价置 0，已无价置 1；实际价格金额仍由报价价格链获取。

关键字段与页面影响

标识不可混用：sid 路由服务站和 Mongo collection；id/invId 是 DGJ2 商品键；skuId 是 ItemCenter 跨系统商品键。

入口清单与计数口径

MQ 价格事件与 ItemCenterJob 是同一业务链的前后半段，不应解释成两次独立业务变化。

多入口请求链路

当前风险和优先级

建议演进路线

flowchart LR
    A["阶段一：可观测<br/>requestId / eventId<br/>结构化日志<br/>一致性巡检"]
    B["阶段二：可靠<br/>落表后 ACK<br/>原子领取<br/>幂等版本与重放"]
    C["阶段三：解耦<br/>基础商品先返回<br/>价格/库存独立加载<br/>提交时强校验"]
    D["阶段四：治理<br/>修复脚本 dry-run<br/>审批与审计<br/>淘汰临时入口"]
    A --> B --> C --> D

建议成功标准：

• 每个 MQ 事件可用 eventId + sid + skuId 追踪到 Mongo、ES、Redis 的最终状态。
• WAIT、ING、ERROR、部分失败有监控和 owner，超时 ING 可回收。
• 任一投影失败可单独重放，不必重新发送整条业务消息。
• 基础商品链成功时，价格或库存单项故障不再导致整页不可用。
• 人工修复具备预估影响、操作审计、读回验证和幂等重跑能力。

需要跨团队确认的决策

15 分钟团队分享建议

推荐用“货主区域变化后商品突然不报价”串联讲解：区域事件 → 主站 SID 归一化 → ItemCenter 有价快照 → Mongo/ES/Redis noPricingFlag 差集 → Basedata → Sale 聚合 → 页面；异常时按相同顺序逐段找证据。

日志证据矩阵

当前多数 Mongo 写方法没有统一上报 matchedCount/modifiedCount；日志“完成”不能替代 Mongo/ES/Redis 读回。

环节数据变更台账

代码与详细技术底稿

仓库：/Users/zhoujiangbin/code/docker-dev-env/www/dgj2.0

详细文档入口：doc/ai/mongo/README.md

关键代码：

• HTTP：application/controllers/basedata/Inventory.php、Invlocation.php
• MQ：application/controllers/tasks/ItemCenterNotify.php、OpsCenterNotify.php
• 后置任务：application/controllers/tasks/ItemCenterJob.php
• 主同步：application/controllers/tasks/Goods.php
• 公共 Mongo 写入：application/Services/Materiels/MaterielCacheStationSer.php
• 价格/区域快照：application/Services/Notify/MaterielNotifySer.php
• 限采差集：application/Services/Materiels/BuyLimitSer.php
• 修复入口：GoodsCache.php、Import.php、GoodsDescUpdate.php

当前证据边界

• A 级代码证据：入口、方法、集合、字段、锁和水位逻辑已从当前代码确认。
• B 级运行证据：部分 ItemCenter MQ 日志和消息表状态已有样例。
• C 级待补证：生产 cron/Kubernetes 表达式、Job 实例数、部分上游 Producer 源码、Mongo 复制延迟基线。
• 本文的演进阶段和成功标准是建议方案，不代表已经上线。