front/docs/订单批量处理与异步队列方案0707.md

# 订单批量处理与异步队列方案

> **文档定位**：梳理 OMS 后端现有订单批量相关 API、执行模式与可复用资产，并给出统一异步队列改造建议。  
> **对照仓库**：`C:\Users\wilso\Desktop\oms0414-back`（`SimpleAdmin.sln`）  
> **前端对照**：`E:\oms0414\src\views\order\myOrders\index.vue`  
> **关联文档**：`docs/oms系统设计-050801.md`（订单章）、`docs/rbac系统设计-060801.md`（权限）  
> **编写日期**：2026-07-07  
> **技术决策**：**选用 RabbitMQ** 作为订单批量异步队列方案（v2.0）

---

## 1　导读

### 1.1 这份文档解决什么问题？

1. **盘点**当前后端所有「批量处理订单」相关 HTTP 接口（含原生批量与前端循环伪批量）。
2. **说明**各接口入参/出参、Service 实现位置、同步/异步执行方式。
3. **指出**现有异步组件（`t_order_batch_job`、`SmallBagTrackOrderNotifier`）及 **`t_order_task` 已合并移除** 的演进结论。
4. **建议**按业务类型划分异步队列、统一批量 Job API 草案与落地优先级。
5. **定义**基于 **RabbitMQ** 的交换机/队列拓扑、消息契约、Consumer 与 Fake 框架集成方式（**已定稿**）。

### 1.2 技术决策

| 项 | 决策 |
| --- | --- |
| 消息中间件 | **RabbitMQ** |
| .NET 接入 | 优先复用 Fake 生态 **`Fake.RabbitMQ`**（连接池/通道池）；订单 Job 使用 **Work Queue 模式** 自建 Publisher/Consumer |
| 与 `Fake.EventBus.RabbitMQ` 关系 | 事件总线用于**跨模块集成事件**（如 Job 完成通知）；**订单子任务投递不走 EventBus**，走专用 Work Queue |
| 状态持久化 | MySQL **`t_order_batch_job` / `t_order_batch_job_item`** 为权威进度；MQ 仅负责调度与重试 |
| 宿主部署 | `SimpleAdmin.Api` 进程内注册 Consumer（`IHostedService`）；后期可拆独立 Worker 项目 |

### 1.3 现状结论（摘要）

| 维度 | 现状 |
| --- | --- |
| API 暴露 | `OrderService`（partial class）上 `[HttpPost]` / `[HttpGet]`，Fake 动态 API，**无手写 Controller** |
| 原生批量 API | 异常件 3 个 + 小包 `batch-create` + Excel 导入 + 条件导出，共 **5 类** |
| 伪批量 | My Orders 页 **8 个单票 API**，前端 `for` 循环逐条调用 |
| 队列基础设施 | **RabbitMQ 已接入**（`Fake.EventBus.RabbitMQ` + `OrderBatchJobPublisher`） |
| 轻量异步 | `SmallBagTrackOrderNotifier` 已逐步由 MQ `track-upsert-small-bag` 替代 |
| 批量任务表 | **`t_order_batch_job` / `t_order_batch_job_item`** 为 Excel 导入及批量操作唯一进度源；**已删除** `t_order_task` / `t_order_task_fail` |
| 权限 | 批量任务页 `Order.BatchJobs.*`；My Orders 批量操作走 Job API |

---

## 2　接口总览

```mermaid
flowchart LR
    subgraph api [SimpleAdmin.Api]
        HTTP["/order/batch/jobs"]
        OS[OrderService]
    end

    subgraph db [MySQL]
        JOB[t_order_batch_job]
        ITEM[t_order_batch_job_item]
    end

    subgraph rmq [RabbitMQ]
        EX["exchange: oms.order.batch"]
        QD["queue: order.batch.integration"]
        QA["queue: order.batch.account"]
        QB["queue: order.batch.state"]
        QC["queue: order.batch.exception"]
        QE["queue: order.batch.document"]
        QF["queue: order.batch.create"]
        DLX["DLX: oms.order.batch.dlx"]
    end

    subgraph consumers [Consumers IHostedService]
        CD[IntegrationConsumer]
        CA[AccountConsumer]
        CB[StateConsumer]
        CC[ExceptionConsumer]
        CE[DocumentConsumer]
        CF[CreateConsumer]
    end

    HTTP --> JOB
    HTTP --> EX
    EX --> QD & QA & QB & QC & QE & QF
    QD & QA & QB & QC & QE & QF --> consumers
    consumers --> ITEM
    consumers --> OS
    QD & QA & QB & QC & QE & QF -.失败.-> DLX
```

### 2.1 分类一览（改造后目标）

```mermaid
flowchart TB
    subgraph native [原生批量 API]
        EP["/order/mark-problematic<br/>/order/mark-claimed<br/>/order/handle-status/update"]
        PCB["/order/parcel/batch-create"]
        UP["/order/upload → batch job"]
        BJ["/order/batch/jobs/*"]
        EX["/order/export"]
    end

    subgraph pseudo [伪批量 - 前端循环单票 API]
        PAY["/order/pay"]
        PD["/order/pre-declare"]
        INT["/order/intercept / cancel-intercept"]
        DEL["/order/delete"]
        PRT["/order/print"]
        TRK["/order/track-upsert-small-bag"]
        DS["/order/downstream-forward-create-one"]
    end

    subgraph svc [OrderService 同步处理]
        OS[OrderService partial]
    end

    subgraph async_hint [异步组件]
        BJDB[t_order_batch_job]
        STN[SmallBagTrackOrderNotifier / MQ]
    end

    native --> OS
    pseudo --> OS
    UP --> BJDB
    BJ --> BJDB
```

### 2.2 原生批量 API 清单

| 路由 | 方法 | Service 方法 | 文件 | 批量语义 | 上限 | 执行模式 |
| --- | --- | --- | --- | --- | --- | --- |
| `/order/mark-problematic` | POST | `MarkProblematicAsync` | `OrderService.Exception.cs` | `List<int> OrderIds` | 200 | 同步 foreach，部分成功 |
| `/order/mark-claimed` | POST | `MarkClaimedAsync` | 同上 | `List<int> OrderIds` | 200 | 同上 |
| `/order/handle-status/update` | POST | `UpdateHandleStatusAsync` | 同上 | `List<int> OrderIds` | 200 | 同上（仅 `processed`） |
| `/order/parcel/batch-create` | POST | `BatchCreateParcelAsync` | `OrderService.cs` | `List<OrderParcelCreateInput>` | 2000 | 同步 for，逐条明细 |
| `/order/upload` | POST | `UploadAsync` → `ImportOrdersAsync` | `OrderService.Helpers.cs` | Excel 多行 | 2000 | MQ 异步入队或同步 ProcessAsync |
| `/order/batch/jobs` | GET | `GetBatchJobListAsync` | `OrderService.BatchJob.cs` | 批量任务列表 | — | 只读 |
| `/order/batch/jobs/{id}` | GET | `GetBatchJobAsync` | 同上 | 任务进度 | — | 只读 |
| `/order/batch/jobs/{id}/items` | GET | `GetBatchJobItemsAsync` | 同上 | 子项明细 | — | 只读 |
| `/order/batch/jobs/{id}/retry-failed` | POST | `RetryBatchJobFailedAsync` | 同上 | 重试失败子项 | — | 写 |
| `/order/batch/jobs/{id}/cancel` | POST | `CancelBatchJobPendingAsync` | 同上 | 取消待处理 | — | 写 |
| `/order/export` | POST | `ExportAsync` | `OrderService.cs` | 筛选条件全量 | — | 同步生成 xlsx |

### 2.3 伪批量 API 清单（单票入参，前端循环）

| 路由 | 方法 | Service 方法 | 文件 | 业务特点 | 前端权限码（参考） |
| --- | --- | --- | --- | --- | --- |
| `/order/pay` | POST | `PayAsync` | `OrderService.cs` | **扣客户余额**，需事务/幂等 | `Order.MyOrders.Pay` |
| `/order/pre-declare` | POST | `PreDeclareAsync` | 同上 | 改订单状态 | `Order.MyOrders.PreDeclare` |
| `/order/intercept` | POST | `InterceptAsync` | 同上 | `IsIntercept=1` | `Order.MyOrders.Intercept` |
| `/order/cancel-intercept` | POST | `CancelInterceptAsync` | 同上 | `IsIntercept=0` | `Order.MyOrders.CancelIntercept` |
| `/order/delete` | POST | `DeleteAsync` | 同上 | 软删 + 物理删 | `Order.MyOrders.Delete` |
| `/order/print` | POST | `PrintAsync` | `OrderService.Helpers.cs` | 生成打印 HTML | `Order.MyOrders.Print` / `BatchPrint` |
| `/order/track-upsert-small-bag` | POST | `TrackUpsertSmallBagAsync` | `OrderService.ParcelBilling.cs` | HTTP 调轨迹服务 | `Order.MyOrders.CreateTrack` |
| `/order/downstream-forward-create-one` | POST | `DownstreamForwardCreateOneAsync` | `OrderService.DownstreamForward.cs` | HTTP 调下游 itfx | `Order.MyOrders.ForwardDownstream` |

**公共入参** `OrderActionInput`：

```csharp
public class OrderActionInput
{
    [Required] public int Id { get; set; }
    [StringLength(64)] public string? RequestId { get; set; }
}
```

**辅助只读**：

| 路由 | 方法 | 说明 |
| --- | --- | --- |
| `/order/downstream-create-one-body` | GET | 预览下游请求体，不实际转发 |

### 2.4 单票编辑类 API（详情抽屉，非列表批量）

| 路由 | 方法 | Service 方法 | 文件 |
| --- | --- | --- | --- |
| `/order/update` | POST | `UpdateOrderDetailAsync` | `OrderService.OrderEdit.cs` |
| `/order/weight/update` | POST | `UpdateOrderWeightsAsync` | 同上 |
| `/order/fee/recalculate-receivable` | POST | `RecalculateReceivableFeesAsync` | 同上 |
| `/order/fee/recalculate-payable` | POST | `RecalculatePayableFeesAsync` | 同上 |
| `/order/fee/add` | POST | `AddOrderFeeAsync` | 同上 |
| `/order/parcel/create` | POST | `CreateParcelAsync` | `OrderService.cs` / `ParcelBilling.cs` |

### 2.5 非 OrderIds 的「批量数据处理」

| 路由 | 方法 | 说明 |
| --- | --- | --- |
| `/order/price-check` | POST | 多产品 × 多仓库运费试算，同步嵌套循环 |
| `/order/list` | POST | 分页列表 |
| `/order/status-counts` | GET | 状态 Tab 计数 |

---

## 3　DTO 与出参模型

### 3.1 异常件批量（推荐复用为队列 Job 汇总出参）

**文件**：`Oms.Application/Dtos/Orders/OrderExceptionDtos.cs`

```csharp
// 统一批量出参
public class OrderBatchActionResultOutput
{
    public int SuccessCount { get; set; }
    public List<OrderBatchActionFailureOutput> Failures { get; set; } = [];
}

public class OrderBatchActionFailureOutput
{
    public int OrderId { get; set; }
    public string? ServiceNo { get; set; }
    public string Reason { get; set; } = string.Empty;
}

// 问题件
public class OrderMarkProblematicInput
{
    public List<int> OrderIds { get; set; } = [];
    public int? IssueTemplateId { get; set; }
    public string? IssueType { get; set; }
    public string? Title { get; set; }
    public string? Content { get; set; }
    public string? Remark { get; set; }
}

// 理赔件
public class OrderMarkClaimedInput
{
    public List<int> OrderIds { get; set; } = [];
    public string? ClaimType { get; set; }
    public decimal? ClaimAmount { get; set; }
    public string? ClaimCurrency { get; set; }
    public string? ClaimReason { get; set; }
    public string? Remark { get; set; }
}

// 标记已处理
public class OrderHandleStatusUpdateInput
{
    public List<int> OrderIds { get; set; } = [];
    public string? HandleStatus { get; set; }
    public string? Remark { get; set; }
}
```

**内部约束**：`MaxExceptionBatchSize = 200`（`OrderService.Exception.cs`）。

### 3.2 小包批量创建

**文件**：`Oms.Application/Dtos/Orders/OrderParcelBatchCreateInput.cs` 等

- **入参**：`OrderParcelBatchCreateInput` 继承 `List<OrderParcelCreateInput>`，单条结构与 `/order/parcel/create` 一致。
- **出参**：`OrderParcelBatchCreateOutput { BatchNo, SuccessCount, FailCount, Items[] }`
- **上限**：`ParcelBatchMaxCount = 2000`（`OrderService.ParcelBilling.cs`）

### 3.3 Excel 导入

| DTO | 字段 |
| --- | --- |
| `OrderUploadInput` | `File` (IFormFile), `TemplateId` |
| `OrderUploadOutput` | `JobId`, `Async` |
| `OrderBatchJobOutput` | 任务汇总（含 `templateId` / `taskName` 导入元数据） |
| `OrderBatchJobItemOutput` | 子项明细（含 `clientNo`、`rowNumber`、`resultMessage`） |

### 3.4 条件导出

| DTO | 字段 |
| --- | --- |
| `OrderExportInput` | `status`, `keyword`, `createTimeStart`, `createTimeEnd` |
| `OrderExportOutput` | `downloadUrl`, `expireAt`（24h 过期） |

---

## 4　现有执行模式详解

### 4.1 异常件批量（已实现部分成功）

**调用链**：

```
MarkProblematicAsync / MarkClaimedAsync / UpdateHandleStatusAsync
  → NormalizeOrderIds（去重、非空、≤200）
  → ExecuteProblematicBatchAsync / ExecuteExceptionBatchAsync
  → 逐条 try/catch，成功 SuccessCount++，失败写入 Failures
```

**问题件单条逻辑**：`OrderExceptionWorkflowService.TryMarkProblematicAsync`（`Issues/OrderExceptionWorkflowService.cs`），含通知（`OrderIssueNotificationService`）。

### 4.2 小包批量创建

```
BatchCreateParcelAsync
  → 生成 BatchNo
  → for each OrderParcelCreateInput
      → CreateParcelCoreAsync（预报、计费、落库）
      → 成功则 EnqueueSmallBagTrackUpsertAsync（轨迹）
  → 返回逐条 Items 明细
```

### 4.3 Excel 导入（统一批量任务）

```
UploadAsync → ImportOrdersAsync
  → BuildImportBatchJobInputAsync
  → Insert t_order_batch_job（Action=import-row）+ t_order_batch_job_item（每行）
  → MQ 开启：Publish 逐行消息 → OrderBatchJobProcessor → CreateParcelCoreAsync
  → MQ 关闭：CreatePayloadJobAndProcessSyncAsync 同步 ProcessAsync
  → 返回 { jobId, async }
```

**进度查询**：`GET /order/batch/jobs/{id}` + `/items`；失败行含 `rowNumber`、`clientNo`、`failReason`，原始 Excel 行在 `PayloadJson.rowDataJson`。

**已移除**：`t_order_task`、`t_order_task_fail`、`/order/task/list`、`/order/task/detail`。

### 4.4 轨迹轻量异步（现有唯一 fire-and-forget）

**文件**：`Oms.Application/Integrations/SmallBagTrack/SmallBagTrackOrderNotifier.cs`

```csharp
public void Enqueue(SmallBagTrackUpsertRequest request)
{
    _ = Task.Run(async () =>
    {
        using var scope = scopeFactory.CreateScope();
        var client = scope.ServiceProvider.GetRequiredService<ISmallBagTrackOrderClient>();
        await client.UpsertAsync(request);
    });
}
```

**性质**：单票、内存队列、无持久化、无重试、进程重启即丢失。

### 4.5 前端伪批量行为

**文件**：`src/views/order/myOrders/index.vue`

- 勾选多行后，对 `pay` / `pre-declare` / `intercept` 等 **逐条 POST** `OrderActionInput`。
- 支付等操作：**任一失败即中断**，无部分成功汇总（与异常件 API 不同）。
- 批量打印 / 轨迹 / 下游：循环调用后前端汇总结果弹窗。

---

## 5　现有基础设施盘点

| 组件 | 是否存在 | 说明 |
| --- | --- | --- |
| RabbitMQ Work Queue | ✅ 已接入 | `Fake.EventBus.RabbitMQ` + `OrderBatchJobPublisher` |
| `Fake.RabbitMQ` | ✅ | 连接池/通道池 |
| 订单专用 Consumer | ✅ | `OrderBatchJobEventHandler` / `OrderBatchJobProcessor` |
| `BackgroundService` | ✅ | `OrderBatchDlxMonitorHostedService` 等 |
| `t_order_batch_job` | ✅ | 批量任务权威进度（含 Excel 导入） |
| `t_order_task` | ❌ 已删除 | 已合并至 `t_order_batch_job` |
| `SmallBagTrackOrderNotifier` | ✅ | 逐步由 MQ 替代 |

### 5.1 批量任务表（已实现）

**表**：`t_order_batch_job` / `t_order_batch_job_item`  
**处理器**：`OrderBatchJobProcessor`、`OrderBatchJobItemFeedback`

| 字段/能力 | 说明 |
| --- | --- |
| `Action` | `import-row` / `pay` / `print` 等 |
| `Status` | `pending` / `running` / `completed` / `failed` / `cancelled` |
| `PayloadJson` | 导入任务含 `taskName`、`templateId` |
| 子项 `PayloadJson` | 导入行含 `rowNumber`、`rowDataJson` |
| `ClientNo` | 客户单号（执行后回填） |
| `RetryCount` | 系统重试次数，耗尽标记 failed |

> **历史**：原 `t_order_task` / `t_order_task_fail` 已于 2026-07 合并至本组表并删表，详见 `sql/migrate_drop_order_task_merge_batch.sql`。

---

## 6　RabbitMQ 队列分组（与业务 Action 映射）

按 **业务风险 + 耗时 + 外部依赖** 划分 **6 条业务队列** + **1 条死信队列**：

| 队列组 | RabbitMQ Queue | Routing Key | 包含 Action | 并发建议 | Prefetch |
| --- | --- | --- | --- | --- | --- |
| **D. 外部集成** | `oms.order.batch.integration` | `integration.*` | `track-upsert-small-bag`, `downstream-forward-create-one` | 中（限流防打爆下游） | 5 |
| **F. 订单创建** | `oms.order.batch.create` | `create.*` | `parcel-create`, `import-row` | 中 | 10 |
| **A. 账务** | `oms.order.batch.account` | `account.*` | `pay` | **低**（按 CustomerId 串行） | 1 |
| **B. 状态变更** | `oms.order.batch.state` | `state.*` | `pre-declare`, `intercept`, `cancel-intercept`, `delete`, `handle-status-update` | 高 | 20 |
| **C. 异常工作流** | `oms.order.batch.exception` | `exception.*` | `mark-problematic`, `mark-claimed` | 中 | 10 |
| **E. 文档生成** | `oms.order.batch.document` | `document.*` | `print`, `export` | 低 | 3 |
| **死信** | `oms.order.batch.dlx` | — | 超过最大重试的子任务 | — | — |

**Exchange**：`oms.order.batch`（类型 **topic**，持久化 `durable=true`）。

### 6.1 落地顺序建议

1. **D（外部集成）** — ✅ 轨迹、下游预报已走 MQ。
2. **F（创建/导入）** — ✅ Excel 导入统一 `import-row` 批量任务；`parcel/batch-create` 仍为同步 HTTP。
3. **A（支付）** — ✅ 批量支付已入队。
4. **B / C / E** — ✅ 预报、拦截、异常件、打印等已接入 Job API。

---

## 7　统一批量 Job API 草案

### 7.1 路由设计

| 路由 | 方法 | 说明 |
| --- | --- | --- |
| `/order/batch/jobs` | POST | 创建批量任务，写 DB 并 Publish 到 RabbitMQ，立即返回 `jobId` |
| `/order/batch/jobs/{id}` | GET | 查询进度与汇总结果 |
| `/order/batch/jobs/{id}/items` | GET | 逐条明细（成功/失败/进行中） |
| `/order/batch/jobs/{id}/cancel` | POST | 取消未执行子任务（可选） |
| `/order/batch/jobs/{id}/retry-failed` | POST | 将 failed 子任务重新 Publish（可选） |

### 7.2 入参草案

```csharp
public class OrderBatchJobCreateInput
{
    /// <summary>
    /// pay | pre-declare | intercept | cancel-intercept | delete |
    /// print | track-upsert-small-bag | downstream-forward-create-one |
    /// mark-problematic | mark-claimed | handle-status-update | export | ...
    /// </summary>
    public string Action { get; set; } = string.Empty;

    public List<int> OrderIds { get; set; } = [];

    public JsonElement? Payload { get; set; }
    public string? RequestId { get; set; }
}
```

### 7.3 出参草案

```csharp
public class OrderBatchJobOutput
{
    public long JobId { get; set; }
    public string Action { get; set; } = string.Empty;
    public string Status { get; set; } = string.Empty;
    public int Total { get; set; }
    public int SuccessCount { get; set; }
    public int FailCount { get; set; }
    public DateTime? StartedAt { get; set; }
    public DateTime? FinishedAt { get; set; }
}
```

单条明细复用 `OrderBatchActionFailureOutput`，扩展 `Status`：`pending` / `running` / `success` / `failed`。

---

## 8　RabbitMQ 实施方案（已定稿）

### 8.1 为何选 RabbitMQ

| 能力 | 说明 |
| --- | --- |
| 持久化队列 | 进程重启不丢任务（配合 `delivery_mode=2`） |
| 死信 + 延迟重试 | 业务失败与系统失败分离治理 |
| 多队列隔离 | 支付、外部 HTTP、导入互不阻塞 |
| 水平扩展 | 多 API/Worker 实例共享 Consumer |
| 与项目基线一致 | `开发复刻说明.md` 已约定 RabbitMQ；Fake 提供 `Fake.RabbitMQ` 连接池 |

**不采用 Hangfire / DB 轮询的原因**：订单批量含大量外部 HTTP 与长耗时导入，需要独立队列削峰与跨实例消费；DB 轮询难以优雅做按客户串行与 DLX。

### 8.2 NuGet 与模块依赖

在 `SimpleAdmin.Api` / `Oms.Application` 增加：

```xml
<PackageReference Include="Fake.RabbitMQ" Version="$(FakePackageVersion)" />
<!-- 可选：Job 完成集成事件 -->
<PackageReference Include="Fake.EventBus.RabbitMQ" Version="$(FakePackageVersion)" />
```

模块依赖（`OmsApplicationModule` 或新建 `OmsRabbitMqModule`）：

```csharp
[DependsOn(typeof(FakeRabbitMqModule))]
public class OmsRabbitMqModule : FakeModule { }
```

### 8.3 配置（appsettings / Consul）

```json
{
  "RabbitMQ": {
    "Connections": {
      "Default": {
        "HostName": "rabbitmq",
        "Port": 5672,
        "UserName": "oms",
        "Password": "***",
        "VirtualHost": "/oms"
      }
    }
  }
}
```

队列声明在应用启动时由 `OrderBatchQueueInitializer`（`IHostedService`）执行：创建 Exchange、6 条业务 Queue、DLX、绑定关系。

### 8.4 消息契约

**队列消息体** `OrderBatchJobMessage`（JSON，Newtonsoft 驼峰）：

```csharp
public class OrderBatchJobMessage
{
  public long JobId { get; set; }
  public long JobItemId { get; set; }
  public string Action { get; set; } = string.Empty;      // pay, delete, track-upsert-small-bag, ...
  public int OrderId { get; set; }
  public int? CustomerId { get; set; }                    // A 组串行键
  public string? PayloadJson { get; set; }               // Action 专用参数
  public int OperatorUserId { get; set; }
  public string MessageId { get; set; } = string.Empty;  // 幂等键
  public int RetryCount { get; set; }
}
```

**发布流程**：

```
POST /order/batch/jobs
  → 写 t_order_batch_job + items（status=pending）
  → 按 Action 映射 routingKey，逐条 Publish 到 oms.order.batch
  → 立即返回 { jobId }
```

**消费流程**：

```
Consumer 收到消息
  → 查 item 是否已成功（幂等跳过）
  → item.status=running
  → IOrderBatchActionExecutor.ExecuteAsync(message)
  → 成功：item.status=success，更新 job 计数
  → 业务失败：item.status=failed + reason，ACK
  → 系统失败：NACK requeue；RetryCount>=3 → DLX
  → job 全部终态后 status=completed
```

### 8.5 与现有组件的替换关系

| 现有 | 改造后 |
| --- | --- |
| `SmallBagTrackOrderNotifier`（`Task.Run`） | Publish 到 `oms.order.batch.integration`，routingKey=`integration.track-upsert` |
| `ImportOrdersAsync` 同步 foreach | 上传存文件/OSS → 写 job → 每行一条消息到 `oms.order.batch.create` |
| 前端 My Orders 循环 `pay` 等 | 统一 `POST /order/batch/jobs`，轮询 `GET /order/batch/jobs/{id}` |
| 异常件 API `mark-problematic` 等 | **兼容期保留同步 API**；新增 `async=true` 或独立 Job API 走 MQ |

### 8.6 重试与死信策略

| 场景 | 策略 |
| --- | --- |
| 下游 HTTP 超时/5xx | NACK + requeue，指数退避（`x-message-ttl` + DLX 延迟队列，或 Consumer 内 `Task.Delay` 后重发） |
| 业务校验失败（已支付、状态不允许） | 记 failed，**ACK 不重试** |
| 消息重复消费 | `MessageId` + DB 唯一约束，已 success 则直接 ACK |
| DLX 堆积 | 告警 + 管理端「重新投递」接口 |

建议队列参数：

```
x-dead-letter-exchange: oms.order.batch.dlx
x-dead-letter-routing-key: dead
```

### 8.7 部署拓扑

```mermaid
flowchart TB
    subgraph prod [生产环境]
        API1[SimpleAdmin.Api 实例 1]
        API2[SimpleAdmin.Api 实例 2]
        RMQ[RabbitMQ Cluster]
        MySQL[(MySQL)]
    end

    API1 & API2 -->|Publish/Consume| RMQ
    API1 & API2 --> MySQL
    FE[oms0414 前端] -->|创建 Job / 轮询进度| API1
```

- **第一期**：Consumer 与 API 同进程（`IHostedService`），减少运维复杂度。
- **第二期**：拆 `Oms.OrderWorker` 独立控制台项目，仅引用 `Oms.Application` + Consumers。

### 8.8 本地开发

`docker-compose` 增加 RabbitMQ 服务（管理台 15672）：

```yaml
rabbitmq:
  image: rabbitmq:3.13-management
  ports:
    - "5672:5672"
    - "15672:15672"
  environment:
    RABBITMQ_DEFAULT_USER: oms
    RABBITMQ_DEFAULT_PASS: oms123
    RABBITMQ_DEFAULT_VHOST: /oms
```

### 8.9 Consumer 执行原则

1. **复用现有 Service**：`IOrderBatchActionExecutor` 按 `action` 分发，内部调用抽离后的单条逻辑。
2. **幂等**：`MessageId = {jobId}:{orderId}:{action}`；DB 唯一键 `(JobId, OrderId, Action)`；支付额外校验 `IsPaid`。
3. **部分成功**：业务失败 ACK 并写 `failed`；系统失败 NACK requeue，超次进 DLX。
4. **并发**：A 组 `prefetch=1` + `customerId` 串行；D 组全局限流。
5. **可观测**：日志带 `JobId/OrderId/Action`；可选 `OrderBatchJobCompletedEto` 走 `Fake.EventBus.RabbitMQ`。

---

## 9　关键文件索引（规划新增）

| 文件（规划路径） | 职责 |
| --- | --- |
| `Oms.Application/Messaging/OrderBatchJobMessage.cs` | MQ 消息契约 |
| `Oms.Application/Messaging/IOrderBatchJobPublisher.cs` | Publish 抽象 |
| `Oms.Application/Messaging/OrderBatchQueueInitializer.cs` | 启动声明 Exchange/Queue |
| `Oms.Application/Workers/OrderBatch/*Consumer.cs` | 6 组 RabbitMQ Consumer |
| `Oms.Application/Services/Orders/Batch/IOrderBatchActionExecutor.cs` | Action 执行分发 |
| `Oms.Application/Services/Orders/OrderService.cs` | pay/delete/export/batch-create/upload 等主 API |
| `Oms.Application/Services/Orders/OrderService.Exception.cs` | 原生 `OrderIds` 批量异常件 API |
| `Oms.Application/Services/Orders/OrderService.ParcelBilling.cs` | 小包创建、轨迹同步 |
| `Oms.Application/Services/Orders/OrderService.DownstreamForward.cs` | 下游预报转发 |
| `Oms.Application/Services/Orders/OrderService.Helpers.cs` | Excel 导入、打印、任务查询 |
| `Oms.Application/Integrations/SmallBagTrack/SmallBagTrackOrderNotifier.cs` | 待替换为 MQ Publisher |
| `Oms.Domain/OrdersAggregate/Entities/OrderBatchJob.cs` | Job 实体（规划） |
| `src/views/order/myOrders/index.vue` | 前端伪批量循环（待改 Job API） |

---

## 10　与主设计文档的关系

- 订单状态机、费用、模板字段详见 **`docs/oms系统设计-050801.md` 第 4 章**。
- 客户隔离、`CustomerId` 解析见 **OMS 设计 §1.4** 与 **RBAC 设计文档**。
- Fake 框架 RabbitMQ 能力见 **`oms0414-back/SPEC.md` §9.3**（`Fake.RabbitMQ`、`Fake.EventBus.RabbitMQ`）。

---

## 11　后续工作清单（RabbitMQ 实施）

### 阶段一：基础设施（D 组） — ✅ 已完成

- [x] RabbitMQ + `Fake.EventBus.RabbitMQ`
- [x] `t_order_batch_job` / `t_order_batch_job_item` 表
- [x] `OrderBatchJobPublisher` + `OrderBatchJobProcessor`
- [x] 轨迹、下游预报 MQ 消费

### 阶段二：创建与导入（F 组） — ✅ 已完成

- [x] `/order/upload` → `import-row` 批量任务（MQ 异步 / 无 MQ 同步）
- [x] **删除** `t_order_task` / `t_order_task_fail` 及 `/order/task/*`
- [ ] `parcel/batch-create` 可选改为异步 Job（当前仍为同步 HTTP）

### 阶段三：统一 Job API + 前端 — ✅ 已完成

- [x] `/order/batch/jobs` 系列 API
- [x] 前端 My Orders / 批量任务页 Job API + 进度轮询
- [x] B/C/E/A 组 Action 与 `IOrderBatchActionExecutor`

### 阶段四：治理与测试 — 进行中

- [x] DLX 监控、`retry-failed` / `cancel` 管理接口
- [x] `OrderBatchJobProcessorTests`
- [ ] 可选：拆 `Oms.OrderWorker` 独立进程

---

*文档版本：v2.1 | 技术决策：RabbitMQ | 2026-07-22（同步：OrderTask 已合并至 BatchJob）*