185 lines
7.7 KiB
Markdown
185 lines
7.7 KiB
Markdown
---
|
||
name: 异步用户统计 MQ
|
||
overview: wallet_user_stat 通过 slot/foundation RabbitMQ 异步更新；废弃 WalletStatModel 分片统计表；commit 后发 MQ，command 消费落库，Redis biz_id 幂等。
|
||
todos:
|
||
- id: mq-entity-keys
|
||
content: 事件 Entity/常量、MQ 命名；foundation RabbitMQConfig 工厂方法
|
||
status: completed
|
||
- id: mq-producer-consumer-service
|
||
content: WalletUserStatMQService（Producer）与 WalletUserStatConsumeService
|
||
status: completed
|
||
- id: command-update-wallet-user-stat
|
||
content: UpdateWalletUserStat command 使用 RabbitMQConsumer::consume
|
||
status: completed
|
||
- id: remove-wallet-stat-model
|
||
content: 移除 WalletStatModel 全部引用，读统计改 WalletUserStatModel
|
||
status: completed
|
||
- id: wire-wallet-logic
|
||
content: 注册/充值/提现完成 commit 后 publish；RegisterService 去掉同步 stat
|
||
status: completed
|
||
- id: doc-and-verify
|
||
content: doc 留痕 + Docker 冒烟（foundation MQ + 幂等）
|
||
status: completed
|
||
isProject: false
|
||
---
|
||
|
||
# wallet_user_stat 异步统计更新方案（修订）
|
||
|
||
## 变更要点（相对上一版）
|
||
|
||
1. **废弃** [`WalletStatModel`](app/model/multi/WalletStatModel.php)（分片 `wallet_stat_XX`）：删除所有同步 `inc` / `insertData` / 读统计逻辑，**不再并行双写**。
|
||
2. **MQ 统一走** [`slot/foundation` `src/MQ`](vendor/slot/foundation/src/MQ/)，**不再**为本需求扩展 [`RabbitMqService`](app/service/RabbitMqService.php)（wager/console 等历史消费者可暂保留旧封装）。
|
||
|
||
## 为什么异步
|
||
|
||
| 维度 | 说明 |
|
||
|------|------|
|
||
| 定位 | `wallet_user_stat` 为运营/展示统计，非真账 |
|
||
| 性能 | 资金事务 commit 后异步写 `rh_wallet`，缩短主路径 |
|
||
| 规范 | 主账本 DB 先提交，再发 MQ；消费幂等 |
|
||
|
||
```mermaid
|
||
sequenceDiagram
|
||
participant Logic as WalletLogic
|
||
participant ShardDB as wallet_shard
|
||
participant Prod as RabbitMQProducer
|
||
participant Cmd as updateWalletUserStat
|
||
participant Cons as RabbitMQConsumer
|
||
participant Stat as wallet_user_stat
|
||
|
||
Logic->>ShardDB: commit
|
||
Logic->>Prod: sendMessage event
|
||
Cmd->>Cons: consume autoAck
|
||
Cons->>Stat: WalletUserStatModel apply*
|
||
```
|
||
|
||
## Foundation MQ API（vendor 已具备）
|
||
|
||
| 类 | 用途 |
|
||
|----|------|
|
||
| [`RabbitMQConfig`](vendor/slot/foundation/src/MQ/RabbitMqConfig.php) | host/port/user/password/vhost |
|
||
| [`RabbitMQProducer`](vendor/slot/foundation/src/MQ/RabbitMqProducer.php) | `getInstance($config, $exchange, $queue, 'direct', $route)->sendMessage(array\|string)` |
|
||
| [`RabbitMQConsumer`](vendor/slot/foundation/src/MQ/RabbitMqConsumer.php) | `consume($callback, $autoAck=true)`、`decodeMessage($message)` |
|
||
|
||
配置从现有 `.env` 组装（与 [`RabbitMqService`](app/service/RabbitMqService.php) 相同变量：`MQ_HOST`、`MQ_PORT`、`MQ_USER`、`MQ_PASSWORD`、`MQ_VHOST`）。
|
||
|
||
本仓新增 **`WalletUserStatMqSupport`**（或写在 `WalletUserStatMQService` 内私有方法）：
|
||
|
||
```php
|
||
use slot\foundation\MQ\RabbitMQConfig;
|
||
use slot\foundation\MQ\RabbitMQProducer;
|
||
use slot\foundation\MQ\RabbitMQConsumer;
|
||
|
||
private static function config(): RabbitMQConfig
|
||
{
|
||
return new RabbitMQConfig(
|
||
(string) getenv('MQ_HOST'),
|
||
(int) getenv('MQ_PORT'),
|
||
(string) getenv('MQ_USER'),
|
||
(string) getenv('MQ_PASSWORD'),
|
||
(string) getenv('MQ_VHOST'),
|
||
);
|
||
}
|
||
```
|
||
|
||
交换机/队列（[`MQKeyManagerService`](app/service/MQKeyManagerService.php) 常量，direct，routingKey = queue 名）：
|
||
|
||
- `EXCHANGE_WALLET_USER_STAT = 'wallet_user_stat'`
|
||
- `QUEUE_WALLET_USER_STAT = 'wallet_user_stat'`
|
||
|
||
## 消息体
|
||
|
||
[`app/entity/mq/WalletUserStatEventEntity.php`](app/entity/mq/WalletUserStatEventEntity.php) + [`app/constants/WalletUserStatEvent.php`](app/constants/WalletUserStatEvent.php)：
|
||
|
||
| 字段 | 说明 |
|
||
|------|------|
|
||
| `event` | `init` / `deposit` / `withdraw` / `promo_bonus` / `cashback` / `refund` / `chargeback` |
|
||
| `uid`, `currency`, `amount`, `biz_id` | 金额 x1000；`init` 时 amount=0 |
|
||
| `occurred_at` | 可选 |
|
||
|
||
消费端映射 [`WalletUserStatModel`](app/model/WalletUserStatModel.php) 已有 `apply*` / `insertInitRow`。
|
||
|
||
## Service 分层
|
||
|
||
| 类 | 职责 |
|
||
|----|------|
|
||
| [`WalletUserStatMQService`](app/service/wallet/WalletUserStatMQService.php) | `publish(WalletUserStatEventEntity)` → `RabbitMQProducer::sendMessage($entity->activeData())` |
|
||
| [`WalletUserStatConsumeService`](app/service/wallet/WalletUserStatConsumeService.php) | 校验、Redis 幂等 `wallet:user_stat:dedupe:{event}:{biz_id}`、调用 Model |
|
||
|
||
**禁止**在 `WalletUserStatMQService` 内再包一层无意义的 `RabbitMqService` 转发。
|
||
|
||
## Command
|
||
|
||
[`app/command/UpdateWalletUserStat.php`](app/command/UpdateWalletUserStat.php)（结构参考 [`UpdateWagerTask`](app/command/UpdateWagerTask.php)，但换 foundation Consumer）：
|
||
|
||
```php
|
||
$consumer = RabbitMQConsumer::getInstance(
|
||
WalletUserStatMqSupport::config(),
|
||
MQKeyManagerService::EXCHANGE_WALLET_USER_STAT,
|
||
MQKeyManagerService::QUEUE_WALLET_USER_STAT,
|
||
'direct',
|
||
MQKeyManagerService::QUEUE_WALLET_USER_STAT,
|
||
);
|
||
$consumer->consume(function (AMQPMessage $message) use ($consumer) {
|
||
$payload = $consumer->decodeMessage($message);
|
||
(new WalletUserStatConsumeService())->handle(is_array($payload) ? $payload : []);
|
||
}, true, false);
|
||
```
|
||
|
||
- 成功回调后 **auto ack**（foundation 默认行为）
|
||
- 业务异常：`LoggerService::error`；是否 requeue 首版 `false`（与 wager 手工 ack 策略不同，更依赖 Redis 幂等 + 日志补数）
|
||
|
||
启动：
|
||
|
||
```bash
|
||
docker exec -w /app/www/ray/slot-wallet php82 php webman updateWalletUserStat
|
||
```
|
||
|
||
## 移除 WalletStatModel
|
||
|
||
| 文件 | 改动 |
|
||
|------|------|
|
||
| [`WalletLogic.php`](app/api/logic/WalletLogic.php) | 删除 `WalletStatModel` use 及 `inc`/`insertData`；commit 后 `WalletUserStatMQService::publish` |
|
||
| [`RegisterService.php`](app/service/wallet/RegisterService.php) | 删除 `$walletStatModel` 与 `insertData`；注册 commit 后发 `init` 事件 |
|
||
| [`WalletLogic::getWallet`](app/api/logic/WalletLogic.php)（`needStat==1`） | 改读 `WalletUserStatModel::findByUidCurrency`：`r` ← `total_deposit_amount`，`tw` ← `total_withdraw_amount` |
|
||
| [`WalletStatModel.php`](app/model/multi/WalletStatModel.php) | **删除文件**（无引用后） |
|
||
|
||
### 发 MQ 关键路径（post-commit）
|
||
|
||
| 场景 | event | amount |
|
||
|------|-------|--------|
|
||
| 注册 | `init` | 0 |
|
||
| `recharge` / `rechargeSign` | `deposit` | `recharge` |
|
||
| 提现完成 `BIZ_TYPE_WITHDRAW` | `withdraw` | `fee` |
|
||
|
||
`biz_id` 使用请求 DTO 已有 `biz_id`（幂等维度与资金接口一致）。
|
||
|
||
## 不涉及
|
||
|
||
- 不改 `config/thinkorm.php`
|
||
- 不把统计写回资金事务
|
||
- 不改造 wager/console 等仍用 `RabbitMqService` 的旧队列（仅新统计队列用 foundation）
|
||
|
||
## 验证
|
||
|
||
1. `composer` 已含 `slot/foundation`（lock 中已有 MQ 类）
|
||
2. 启动 consumer + 走注册/充值/提现
|
||
3. 查 `rh_wallet.wallet_user_stat`
|
||
4. 重复 `biz_id` 验证 Redis 幂等
|
||
5. 确认 `wallet_stat_XX` 不再被写入；`getWallet` 的 `r`/`tw` 来自新表
|
||
|
||
## 文件一览
|
||
|
||
| 操作 | 路径 |
|
||
|------|------|
|
||
| 新增 | `app/entity/mq/WalletUserStatEventEntity.php` |
|
||
| 新增 | `app/constants/WalletUserStatEvent.php` |
|
||
| 新增 | `app/service/wallet/WalletUserStatMqSupport.php`（可选，集中 Config/实例化） |
|
||
| 新增 | `app/service/wallet/WalletUserStatMQService.php` |
|
||
| 新增 | `app/service/wallet/WalletUserStatConsumeService.php` |
|
||
| 新增 | `app/command/UpdateWalletUserStat.php` |
|
||
| 修改 | `app/service/MQKeyManagerService.php`、`app/service/RedisKeyManagerService.php` |
|
||
| 修改 | `app/api/logic/WalletLogic.php`、`app/service/wallet/RegisterService.php` |
|
||
| 删除 | `app/model/multi/WalletStatModel.php` |
|
||
| 新增/修改 | `doc/` 说明 foundation MQ 与启动命令 |
|