---
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 与启动命令 |