ray604638018/cursor

Fork 0

Files

ray zhou 2dd9f17da9 ok

2026-06-29 14:51:55 +08:00

5.7 KiB

Raw Blame History

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 在 connect() 里写死了：

$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 当前调用：

$rabbitMqService->getMessage([$this, 'deal']);

已是 manual ack（no_ack = false，deal() 末尾 $message->ack()），只差把 prefetch 改成 1。

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 — getMessage($callback, $qos=false)，$qos=true 时 basic_qos(null, 1, false)
slot_hub/plugin/slot/hub/WorkerBusiness.php — 消费侧 getMessage(..., true)

1. 改 `RabbitMqService`

文件：slot_console/app/service/RabbitMqService.php

connect 阶段：去掉 setBasicQos(null, 10, null)（发布端不需要 prefetch；消费端在 getMessage 里单独设置更明确）。
getMessage：增加第二/along slot_pwa 风格：

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

$rabbitMqService->getMessage([$this, 'deal'], true);

仅此一处消费 console_bus 的入口需要改；其它只 sendMessage 的调用不受影响。

3. Supervisor 多进程部署（你已选此方式）

每个 worker 仍是独立 PHP 进程 + 独立 AMQP connection，订阅同一队列 console_bus（MQKeyManagerService::QUEUE_CONSOLE_BUS）。RabbitMQ 默认 round-robin：有 3 个 consumer 就 3 路并行，且配合 prefetch=1 后每个 consumer 同时只「持有」1 条未 ack 消息。

示例（进程数按 CPU/吞吐调整）：

[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 一致）：

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 独占队列，与多进程目标相反）。

多进程额外注意（非本次必改，但上线前心里有数）

同 uid 消息顺序：多 consumer 后，同一用户的两条 bus 消息可能乱序执行；若某类 event 强依赖顺序，需在 Logic 层加锁/幂等（例如按 uid 分布式锁），不能单靠 prefetch。
失败重试：目前仅 TYPE_FREE_CREDIT_INIT 在 catch 里 nack(true) 重入队；其它类型异常后仍 ack()，多进程不会放大这个问题，但重试策略需业务上接受。
验证：RabbitMQ 管理台 → Queues → console_bus → Consumers，应看到 N 个 consumer；压测时每个 consumer 的 Unacked 应 ≤ 1。

改动范围

必改：RabbitMqService.php、EventBus.php
不改：Supervisor 配置可在运维侧按 numprocs 调整，不必动 PHP 代码
不测 RabbitMQ 真连接的单测可保持现状；若有 mock 消费测试，传入第二参数 true 即可

5.7 KiB Raw Blame History Unescape Escape