511 lines
14 KiB
Markdown
511 lines
14 KiB
Markdown
# 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 |