129 lines
5.7 KiB
Markdown
129 lines
5.7 KiB
Markdown
<!-- 78ee2e02-349b-4b66-a20c-03e7ee129a45 -->
|
||
---
|
||
todos:
|
||
- id: "rabbitmq-prefetch"
|
||
content: "RabbitMqService：connect 去掉 prefetch=10；getMessage 增加 singleMessagePrefetch 参数并在 true 时 basic_qos(1)"
|
||
status: pending
|
||
- id: "eventbus-call"
|
||
content: "EventBus::execute 调用 getMessage([$this, 'deal'], true)"
|
||
status: pending
|
||
- id: "consumer-tag"
|
||
content: "（可选）consumer_tag 加 getmypid() 便于管理台区分多进程"
|
||
status: pending
|
||
- id: "verify-supervisor"
|
||
content: "Supervisor numprocs 配置 + RabbitMQ 管理台确认 N consumer 且 Unacked≤1"
|
||
status: pending
|
||
isProject: false
|
||
---
|
||
# EventBus 多进程「每进程一次只处理一条」方案
|
||
|
||
## 现状与问题
|
||
|
||
[`slot_console/app/service/RabbitMqService.php`](slot_console/app/service/RabbitMqService.php) 在 `connect()` 里写死了：
|
||
|
||
```php
|
||
$this->setExhcange($type)->setBasicQos(null, 10, null)->setQueue()->queueBind();
|
||
```
|
||
|
||
`prefetchCount = 10` 表示：**每个 consumer 最多预取 10 条未 ack 的消息**。Supervisor 开 N 个 `php webman event:bus` 时，理论上最多有 `N × 10` 条消息被「占住」但未必在处理（尤其处理慢时，其余 9 条会堵在该进程本地）。
|
||
|
||
[`EventBus.php`](slot_console/app/command/EventBus.php) 当前调用：
|
||
|
||
```php
|
||
$rabbitMqService->getMessage([$this, 'deal']);
|
||
```
|
||
|
||
已是 **manual ack**（`no_ack = false`，`deal()` 末尾 `$message->ack()`），只差把 prefetch 改成 1。
|
||
|
||
```mermaid
|
||
sequenceDiagram
|
||
participant Q as console_bus
|
||
participant P1 as Process1
|
||
participant P2 as Process2
|
||
Q->>P1: deliver msg1 (prefetch=1)
|
||
Q->>P2: deliver msg2
|
||
Note over P1: 处理 msg1，未 ack 前不再投递
|
||
P1->>Q: ack msg1
|
||
Q->>P1: deliver msg3
|
||
```
|
||
|
||
## 推荐改动（对齐 slot_pwa / slot_hub 既有模式）
|
||
|
||
同仓库已有先例：
|
||
|
||
- [`slot_pwa/app/service/RabbitMqService.php`](slot_pwa/app/service/RabbitMqService.php) — `getMessage($callback, $qos=false)`，`$qos=true` 时 `basic_qos(null, 1, false)`
|
||
- [`slot_hub/plugin/slot/hub/WorkerBusiness.php`](slot_hub/plugin/slot/hub/WorkerBusiness.php) — 消费侧 `getMessage(..., true)`
|
||
|
||
### 1. 改 `RabbitMqService`
|
||
|
||
文件：[`slot_console/app/service/RabbitMqService.php`](slot_console/app/service/RabbitMqService.php)
|
||
|
||
- **connect 阶段**：去掉 `setBasicQos(null, 10, null)`（发布端不需要 prefetch；消费端在 `getMessage` 里单独设置更明确）。
|
||
- **getMessage**：增加第二/along slot_pwa 风格：
|
||
|
||
```php
|
||
public function getMessage(callable $callback, bool $singleMessagePrefetch = false): void
|
||
{
|
||
if ($singleMessagePrefetch) {
|
||
$this->_channel->basic_qos(null, 1, false); // 每 consumer 未 ack 前最多 1 条
|
||
}
|
||
// basic_consume + wait 循环保持不变
|
||
}
|
||
```
|
||
|
||
- **consumer_tag（可选但建议）**：由固定 `$queueName . 'consumer'` 改为带 pid，便于 RabbitMQ 管理台区分多个 Supervisor 进程，例如 `$this->queueName . '_consumer_' . getmypid()`。同一 channel 内唯一即可，多进程各自独立 connection，不会冲突。
|
||
|
||
### 2. 改 `EventBus` 调用
|
||
|
||
文件：[`slot_console/app/command/EventBus.php`](slot_console/app/command/EventBus.php)
|
||
|
||
```php
|
||
$rabbitMqService->getMessage([$this, 'deal'], true);
|
||
```
|
||
|
||
仅此一处消费 `console_bus` 的入口需要改；其它只 `sendMessage` 的调用不受影响。
|
||
|
||
### 3. Supervisor 多进程部署（你已选此方式）
|
||
|
||
每个 worker 仍是独立 PHP 进程 + 独立 AMQP connection，**订阅同一队列** `console_bus`（[`MQKeyManagerService::QUEUE_CONSOLE_BUS`](slot_console/app/service/MQKeyManagerService.php)）。RabbitMQ 默认 round-robin：有 3 个 consumer 就 3 路并行，且配合 prefetch=1 后每个 consumer 同时只「持有」1 条未 ack 消息。
|
||
|
||
示例（进程数按 CPU/吞吐调整）：
|
||
|
||
```ini
|
||
[program:slot_console_event_bus]
|
||
command=docker exec -w /app/www/slot/slot_console php82 php webman event:bus
|
||
process_name=%(program_name)s_%(process_num)02d
|
||
numprocs=3
|
||
autostart=true
|
||
autorestart=true
|
||
```
|
||
|
||
本地联调可开多个终端各跑一条（与 [`slot_console/doc/lucky_reward_deploy.md`](slot_console/doc/lucky_reward_deploy.md) 一致）：
|
||
|
||
```bash
|
||
docker exec -w /app/www/slot/slot_console php82 php webman event:bus
|
||
```
|
||
|
||
## 机制说明（为何这样就够）
|
||
|
||
| 机制 | 作用 |
|
||
| --- | --- |
|
||
| `basic_qos(0, 1, false)` | 该 consumer 未 ack 前，broker 不再向它推第 2 条 |
|
||
| `no_ack = false` + `deal()` 末尾 `ack()` | 处理完成后才释放「占用槽位」 |
|
||
| 多进程各 `basic_consume` 同一 queue | broker 在多个 consumer 间分发 |
|
||
| 每进程独立 connection | 不共享 channel，`getInstance` 单例仅进程内有效 |
|
||
|
||
**不需要** `exclusive = true`（那会变成单 consumer 独占队列，与多进程目标相反）。
|
||
|
||
## 多进程额外注意（非本次必改，但上线前心里有数）
|
||
|
||
1. **同 uid 消息顺序**：多 consumer 后，同一用户的两条 bus 消息可能乱序执行；若某类 event 强依赖顺序，需在 Logic 层加锁/幂等（例如按 `uid` 分布式锁），不能单靠 prefetch。
|
||
2. **失败重试**：目前仅 `TYPE_FREE_CREDIT_INIT` 在 catch 里 `nack(true)` 重入队；其它类型异常后仍 `ack()`，多进程不会放大这个问题，但重试策略需业务上接受。
|
||
3. **验证**：RabbitMQ 管理台 → Queues → `console_bus` → Consumers，应看到 N 个 consumer；压测时每个 consumer 的 **Unacked** 应 ≤ 1。
|
||
|
||
## 改动范围
|
||
|
||
- 必改：[`RabbitMqService.php`](slot_console/app/service/RabbitMqService.php)、[`EventBus.php`](slot_console/app/command/EventBus.php)
|
||
- 不改：Supervisor 配置可在运维侧按 `numprocs` 调整，不必动 PHP 代码
|
||
- 不测 RabbitMQ 真连接的单测可保持现状；若有 mock 消费测试，传入第二参数 `true` 即可
|