datahub/docs/OrderConsumer业务流程.md

# OrderConsumer 业务流程图

## 概览

OrderConsumer 是订单消息的消费者，负责从 `orders.queue` 接收消息，解析并持久化订单数据到数据库。它实现了自动重试机制，失败的消息会自动重试，超过重试次数的消息会被发送到错误队列等待人工处理。

---

## 完整流程图

```mermaid
flowchart TD
    Start([消息到达 orders.queue]) --> CheckDebug{是否开启调试延迟?}
    CheckDebug -->|是| Sleep[休眠 N 秒]
    CheckDebug -->|否| GetRetry[获取重试次数]
    Sleep --> GetRetry

    GetRetry --> ParseRetry{检查 application_headers}

    ParseRetry -->|不存在| FirstTime[首次处理 retryCount=0]
    ParseRetry -->|存在| ReadXDeath[读取 x-death count]
    ReadXDeath --> RetryCount[retryCount = count]

    FirstTime --> LogRetry[记录重试次数]
    RetryCount --> LogRetry

    LogRetry --> TryBlock[开始处理消息]

    TryBlock --> CreateParser[创建 Parser]
    CreateParser --> ExtractMeta[提取 metadata]
    ExtractMeta --> EntityMatch[匹配实体对象]
    EntityMatch --> EntityMap[转换 raw_data]
    EntityMap --> BeginTx[开始事务]

    BeginTx --> LoopSave[保存实体]
    LoopSave --> Commit[提交事务]
    Commit --> ReturnACK[返回 ACK]
    ReturnACK --> End([处理完成])

    TryBlock -->|异常| CatchError[捕获异常]
    CatchError --> Rollback[回滚事务]
    Rollback --> LogError[记录错误]
    LogError --> CheckMaxRetry{超过重试次数?}

    CheckMaxRetry -->|是| ExceededLog[超过重试次数]
    CheckMaxRetry -->|否| NotExceededLog[未超过次数]

    ExceededLog --> SendError[发送到错误队列]
    SendError --> BuildErrorMsg[构建错误消息]
    BuildErrorMsg --> ProduceError[发送到 errors.exchange]
    ProduceError --> TrySend{发送成功?}

    TrySend -->|成功| SuccessLog[记录成功日志]
    TrySend -->|失败| FailLog[记录失败日志]

    SuccessLog --> AckAfterError[返回 ACK]
    FailLog --> AckAfterError
    AckAfterError --> EndError([已发送到错误队列])

    NotExceededLog --> ReturnNACK[返回 NACK]
    ReturnNACK --> ToDLX([进入 DLX])
    ToDLX --> ToRetryQueue[进入重试队列]
    ToRetryQueue --> WaitTTL[等待 5 秒]
    WaitTTL --> BackToMain[TTL到期回流]
    BackToMain --> BackToQueue[回到 orders.queue]
    BackToQueue --> Start

    style Start fill:#e1f5e1
    style End fill:#e1f5e1
    style EndError fill:#ffe1e1
    style ReturnACK fill:#90EE90
    style AckAfterError fill:#FFB6C1
    style ReturnNACK fill:#FFD700
    style SendError fill:#FF6B6B
    style Commit fill:#4CAF50
    style Rollback fill:#F44336
```

---

## 消息流转状态图

```mermaid
stateDiagram-v2
    [*] --> OrdersQueue: 新消息到达

    OrdersQueue --> Processing: Consumer 消费

    Processing --> Success: 处理成功
    Processing --> FirstFailure: 首次失败 count=0
    Processing --> SecondFailure: 第2次失败 count=1
    Processing --> ThirdFailure: 第3次失败 count=2
    Processing --> MaxRetryExceeded: 超过重试次数

    Success --> [*]: ACK

    FirstFailure --> DLX: NACK requeue=false
    SecondFailure --> DLX: NACK requeue=false
    ThirdFailure --> DLX: NACK requeue=false

    DLX --> RetryQueue: routing_key retry

    RetryQueue --> RetryQueue: 等待 TTL 5秒
    RetryQueue --> MainExchange: TTL到期自动死信

    MainExchange --> OrdersQueue: routing_key order.retry

    MaxRetryExceeded --> ErrorsExchange: sendToErrorQueue
    ErrorsExchange --> ErrorsQueue: routing_key error
    ErrorsQueue --> [*]: 等待人工处理
```

---

## 重试计数机制

```mermaid
graph LR
    M1[消息 首次处理] --> H1{有 headers?}
    H1 -->|否| C0[count = 0]

    C0 -->|NACK| M2[消息 第1次重试]
    M2 --> H2{有 headers?}
    H2 -->|是| XD2[读取 x-death]
    XD2 --> C1[count = 1]

    C1 -->|NACK| M3[消息 第2次重试]
    M3 --> H3{有 headers?}
    H3 -->|是| XD3[读取 x-death]
    XD3 --> C2[count = 2]

    C2 -->|NACK| M4[消息 第3次重试]
    M4 --> H4{有 headers?}
    H4 -->|是| XD4[读取 x-death]
    XD4 --> C3[count = 3]

    C3 --> ERROR[发送到错误队列]

    style C0 fill:#90EE90
    style C1 fill:#FFD700
    style C2 fill:#FFA500
    style C3 fill:#FF6B6B
    style ERROR fill:#DC143C,color:#fff
```

---

## 关键代码逻辑

### 1. getRetryCount() - 获取重试次数

```php
protected function getRetryCount(AMQPMessage $message): int
{
    // ⚠️ 关键：首次消息没有 application_headers，必须先检查
    if (!$message->has('application_headers')) {
        return 0;  // 首次处理
    }

    $headers = $message->get('application_headers');
    if (!$headers) {
        return 0;
    }

    $headerData = $headers->getNativeData();
    $xDeath = $headerData['x-death'] ?? [];

    if (empty($xDeath)) {
        return 0;
    }

    // RabbitMQ 自动维护的重试次数
    return $xDeath[0]['count'] ?? 0;
}
```

**x-death 结构示例**:
```php
[
    'x-death' => [
        0 => [
            'count' => 2,                    // 重试次数
            'reason' => 'rejected',          // 死信原因
            'queue' => 'orders.queue',       // 原队列
            'time' => 1701234567,           // 时间戳
            'exchange' => 'dlx.orders',      // DLX 名称
            'routing-keys' => ['retry']
        ]
    ]
]
```

### 2. sendToErrorQueue() - 发送到错误队列

```php
protected function sendToErrorQueue(AMQPMessage $message, Throwable $error): void
{
    $originalData = json_decode($message->getBody(), true);

    // 构建详细的错误消息
    $errorMessage = [
        'error_id' => uniqid('err_', true),
        'original_message' => $originalData,
        'error' => [
            'type' => get_class($error),
            'message' => $error->getMessage(),
            'trace' => $error->getTraceAsString(),
            'timestamp' => date('c'),
        ],
        'metadata' => [
            'platform' => $originalData['platform'] ?? 'unknown',
            'platform_id' => $originalData['meta']['platform_id'] ?? null,
            'company_id' => $originalData['meta']['company_id'] ?? null,
            'store_id' => $originalData['meta']['store_id'] ?? null,
            'data_type' => $originalData['data_type'] ?? 'unknown',
            'failed_at' => date('c'),
            'retry_count' => $this->getRetryCount($message),
        ],
    ];

    // 使用 errors.exchange 发送（不是默认 exchange）
    $producer = ApplicationContext::getContainer()->get(Producer::class);
    $producer->produce(
        new class($errorMessage) extends \Hyperf\Amqp\Message\ProducerMessage {
            protected string $exchange = 'errors.exchange';      // ✅ 通用错误交换机
            protected string|array $routingKey = 'error';
            protected string $poolName = 'default_consumer';
            protected array $properties = [
                'delivery_mode' => 2,  // 持久化
            ];

            public function __construct(array $data) {
                $this->payload = $data;
            }
        }
    );
}
```

---

## 时序图：完整的重试流程

```mermaid
sequenceDiagram
    participant P as Producer
    participant MQ as orders.queue
    participant C as OrderConsumer
    participant DLX as dlx.orders
    participant RQ as orders.retry.queue
    participant ME as main.exchange
    participant EX as errors.exchange
    participant EQ as errors.queue
    participant DB as Database

    Note over P,MQ: 1. 首次消息
    P->>MQ: 发送订单消息
    MQ->>C: 消费消息 (count=0)
    C->>C: getRetryCount() → 0
    C->>DB: 尝试保存数据
    DB-->>C: 失败 (异常)
    C->>C: count(0) < MAX(3)
    C->>MQ: NACK(requeue=false)
    MQ->>DLX: 死信到 DLX
    Note over DLX: x-death count=1
    DLX->>RQ: routing_key: retry

    Note over RQ: 2. 第1次重试
    RQ->>RQ: 等待 TTL 5秒
    RQ->>ME: TTL到期，自动死信
    ME->>MQ: routing_key: order.retry
    MQ->>C: 消费消息 (count=1)
    C->>C: getRetryCount() → 1
    C->>DB: 尝试保存数据
    DB-->>C: 失败 (异常)
    C->>C: count(1) < MAX(3)
    C->>MQ: NACK(requeue=false)
    MQ->>DLX: 死信到 DLX
    Note over DLX: x-death count=2
    DLX->>RQ: routing_key: retry

    Note over RQ: 3. 第2次重试
    RQ->>RQ: 等待 TTL 5秒
    RQ->>ME: TTL到期，自动死信
    ME->>MQ: routing_key: order.retry
    MQ->>C: 消费消息 (count=2)
    C->>C: getRetryCount() → 2
    C->>DB: 尝试保存数据
    DB-->>C: 失败 (异常)
    C->>C: count(2) < MAX(3)
    C->>MQ: NACK(requeue=false)
    MQ->>DLX: 死信到 DLX
    Note over DLX: x-death count=3
    DLX->>RQ: routing_key: retry

    Note over RQ,EQ: 4. 超过重试次数
    RQ->>RQ: 等待 TTL 5秒
    RQ->>ME: TTL到期，自动死信
    ME->>MQ: routing_key: order.retry
    MQ->>C: 消费消息 (count=3)
    C->>C: getRetryCount() → 3
    C->>DB: 尝试保存数据
    DB-->>C: 失败 (异常)
    C->>C: count(3) >= MAX(3) ✓
    C->>EX: sendToErrorQueue()
    EX->>EQ: 消息进入错误队列
    C->>MQ: ACK (防止再次重试)
    Note over EQ: 等待人工处理
```

---

## 配置参数

### 环境变量 (.env)

```bash
# 最大重试次数（默认3次）
AMQP_MAX_RETRIES=3

# 调试延迟（秒），便于在 mq:status 中观察消息流转
# 生产环境设置为 0，开发环境可设置为 2
AMQP_CONSUMER_DEBUG_DELAY=0
```

### Consumer 配置

```php
#[Consumer(
    exchange: "main.exchange",
    routingKey: "order.#",
    queue: "orders.queue",
    pool: "default_consumer",
    nums: 1,
    enable: true
)]
class OrderConsumer extends ConsumerMessage
{
    // ⚠️ 关键：必须设置为 false，否则会无限循环重试
    protected bool $requeue = false;

    protected ?array $qos = [
        'prefetch_size' => 0,
        'prefetch_count' => 1,  // 开发环境建议设置为1，生产环境可设置为10-100
        'global' => false,
    ];
}
```

### 队列配置 (rabbitmq.sh)

**orders.queue**:
- `x-dead-letter-exchange`: `dlx.orders` - 失败时发送到的 DLX
- `x-dead-letter-routing-key`: `retry` - DLX 路由键
- `x-message-ttl`: 86400000 (24小时) - 消息最长存活时间

**orders.retry.queue**:
- `x-message-ttl`: 5000 (5秒) - 重试延迟时间
- `x-dead-letter-exchange`: `main.exchange` - TTL到期后回流的 exchange
- `x-dead-letter-routing-key`: `order.retry` - 回流时的 routing key

**errors.queue**:
- `x-message-ttl`: 604800000 (7天) - 错误消息保留时间

---

## 监控和调试

### 查看队列状态

```bash
php bin/hyperf.php app:mq:status
```

输出示例:
```
=== Business Queues ===
+-----------+---------+----------+---------+-----------+
| Metric    | orders  | products | refunds | inventory |
+-----------+---------+----------+---------+-----------+
| Messages  | 0       | 0        | 0       | 0         |
| Consumers | 1       | 0        | 0       | 0         |
| Status    | ✓ Empty | ✓ Empty  | ✓ Empty | ✓ Empty   |
+-----------+---------+----------+---------+-----------+

=== Dead Letter Queues (Retry Queues) ===
+-----------+--------------+----------------+---------------+-----------------+
| Metric    | orders.retry | products.retry | refunds.retry | inventory.retry |
+-----------+--------------+----------------+---------------+-----------------+
| Messages  | 0            | 0              | 0             | 0               |
| Consumers | 0            | 0              | 0             | 0               |
| Status    | ✓ Empty      | ✓ Empty        | ✓ Empty       | ✓ Empty         |
+-----------+--------------+----------------+---------------+-----------------+

=== Shared Queues ===
+-----------+----------+
| Metric    | errors   |
+-----------+----------+
| Messages  | 5        |
| Consumers | 0        |
| Status    | ✓ Active |
+-----------+----------+
```

### 开启调试模式观察流程

```bash
# 1. 设置调试延迟
echo "AMQP_CONSUMER_DEBUG_DELAY=2" >> .env

# 2. 启动消费者
php bin/hyperf.php start

# 3. 发送测试消息
php bin/hyperf.php app:mq-push:tmall

# 4. 观察日志输出
tail -f runtime/logs/hyperf.log
```

你将看到：
```
>>> No application_headers, first time processing
Retry count: 0/3
=== Error Caught ===
Error: Cannot find parser for platform 'tmall' and entity 'order'
>>> Retry not exceeded, sending to DLX (NACK)

[5秒后]
>>> Extracted count from x-death: 1
Retry count: 1/3
>>> Retry not exceeded, sending to DLX (NACK)

[5秒后]
>>> Extracted count from x-death: 2
Retry count: 2/3
>>> Retry not exceeded, sending to DLX (NACK)

[5秒后]
>>> Extracted count from x-death: 3
Retry count: 3/3
>>> MAX RETRIES EXCEEDED! Sending to error queue...
>>> Successfully sent to error queue!
>>> Returning ACK to prevent further retries
```

---

## 常见问题排查

### 问题1: 消息无限循环，队列始终显示空

**原因**: `requeue` 没有设置为 `false`

**解决**:
```php
protected bool $requeue = false;  // ⚠️ 必须设置
```

### 问题2: OutOfBoundsException - No "application_headers"

**原因**: 首次处理的消息没有 `application_headers` 属性

**解决**: 使用 `has()` 检查
```php
if (!$message->has('application_headers')) {
    return 0;
}
```

### 问题3: 消息不进入错误队列

**原因**:
1. Consumer 没有 `errors.exchange` 的写权限
2. 使用了默认 exchange (空字符串) 而非 `errors.exchange`

**解决**:
1. 确保 rabbitmq.sh 中 consumer 权限包含 `errors.exchange`
2. 使用 `errors.exchange` 而非空 exchange:
```php
protected string $exchange = 'errors.exchange';  // ✅ 正确
protected string $exchange = '';                 // ❌ 错误
```

### 问题4: 重试次数始终是 0

**原因**:
1. 重试队列没有正确配置 `x-dead-letter-routing-key`
2. 消息没有回流到主队列

**解决**: 确保 rabbitmq.sh 中重试队列配置：
```bash
--arguments '{"x-message-ttl":5000,"x-dead-letter-exchange":"main.exchange","x-dead-letter-routing-key":"order.retry"}'
```

---

## 相关文档

- [RabbitMQ.md](RabbitMQ.md) - RabbitMQ 架构和配置详解
- [HyperfAmqp.md](HyperfAmqp.md) - Hyperf AMQP 使用指南和重试机制详解
- [EntityParseFactory完整使用指南.md](EntityParseFactory完整使用指南.md) - 实体解析工厂使用



## 注意事项

EntityParse 为数据处理的核心业务，会影响后续连续聚合物化视图的结果

backend/app/Platform/[Platform]/EntityParse/Order.php
backend/app/Platform/Tmall/EntityParse/Order.php