ray604638018/cursor

Fork 0

Files

ray zhou f71a5c59af ok

2026-05-29 11:21:40 +08:00

7.7 KiB

Raw Blame History

name, overview, todos, isProject

name

overview

todos

isProject

异步用户统计 MQ

wallet_user_stat 通过 slot/foundation RabbitMQ 异步更新；废弃 WalletStatModel 分片统计表；commit 后发 MQ，command 消费落库，Redis biz_id 幂等。

id	content	status
mq-entity-keys	事件 Entity/常量、MQ 命名；foundation RabbitMQConfig 工厂方法	completed

id	content	status
mq-producer-consumer-service	WalletUserStatMQService（Producer）与 WalletUserStatConsumeService	completed

id	content	status
command-update-wallet-user-stat	UpdateWalletUserStat command 使用 RabbitMQConsumer::consume	completed

id	content	status
remove-wallet-stat-model	移除 WalletStatModel 全部引用，读统计改 WalletUserStatModel	completed

id	content	status
wire-wallet-logic	注册/充值/提现完成 commit 后 publish；RegisterService 去掉同步 stat	completed

id	content	status
doc-and-verify	doc 留痕 + Docker 冒烟（foundation MQ + 幂等）	completed

false

wallet_user_stat 异步统计更新方案（修订）

变更要点（相对上一版）

废弃 WalletStatModel（分片 wallet_stat_XX）：删除所有同步 inc / insertData / 读统计逻辑，不再并行双写。
MQ 统一走 slot/foundation src/MQ，不再为本需求扩展 RabbitMqService（wager/console 等历史消费者可暂保留旧封装）。

为什么异步

维度	说明
定位	`wallet_user_stat` 为运营/展示统计，非真账
性能	资金事务 commit 后异步写 `rh_wallet`，缩短主路径
规范	主账本 DB 先提交，再发 MQ；消费幂等

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`	host/port/user/password/vhost
`RabbitMQProducer`	`getInstance($config, $exchange, $queue, 'direct', $route)->sendMessage(array\|string)`
`RabbitMQConsumer`	`consume($callback, $autoAck=true)`、`decodeMessage($message)`

配置从现有 .env 组装（与 RabbitMqService 相同变量：MQ_HOST、MQ_PORT、MQ_USER、MQ_PASSWORD、MQ_VHOST）。

本仓新增 WalletUserStatMqSupport（或写在 WalletUserStatMQService 内私有方法）：

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 常量，direct，routingKey = queue 名）：

EXCHANGE_WALLET_USER_STAT = 'wallet_user_stat'
QUEUE_WALLET_USER_STAT = 'wallet_user_stat'

消息体

app/entity/mq/WalletUserStatEventEntity.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 已有 apply* / insertInitRow。

Service 分层

类	职责
`WalletUserStatMQService`	`publish(WalletUserStatEventEntity)` → `RabbitMQProducer::sendMessage($entity->activeData())`
`WalletUserStatConsumeService`	校验、Redis 幂等 `wallet:user_stat:dedupe:{event}:{biz_id}`、调用 Model

禁止在 WalletUserStatMQService 内再包一层无意义的 RabbitMqService 转发。

Command

app/command/UpdateWalletUserStat.php（结构参考 UpdateWagerTask，但换 foundation Consumer）：

$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 幂等 + 日志补数）

启动：

docker exec -w /app/www/ray/slot-wallet php82 php webman updateWalletUserStat

移除 WalletStatModel

文件	改动
`WalletLogic.php`	删除 `WalletStatModel` use 及 `inc`/`insertData`；commit 后 `WalletUserStatMQService::publish`
`RegisterService.php`	删除 `$walletStatModel` 与 `insertData`；注册 commit 后发 `init` 事件
`WalletLogic::getWallet`（`needStat==1`）	改读 `WalletUserStatModel::findByUidCurrency`：`r` ← `total_deposit_amount`，`tw` ← `total_withdraw_amount`
`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）

验证

composer 已含 slot/foundation（lock 中已有 MQ 类）
启动 consumer + 走注册/充值/提现
查 rh_wallet.wallet_user_stat
重复 biz_id 验证 Redis 幂等
确认 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 与启动命令

7.7 KiB Raw Blame History Unescape Escape