Symfony Messenger调度大量消息时内存泄漏问题排查与方案咨询

调度数千万条消息时内存占用持续增长，原因不明。以下是简化后的消息调度代码：

$total = 10000000;
$iterations = \ceil($total / 10000);

for ($i = 0; $i <= $iterations; ++$i) {
    $start = \microtime(true);
    $ids = \range(0, 10000);

    foreach ($ids as $id) {
        $this->messageBus->dispatch(new Message((int) $id));
    }

    $end = \microtime(true) - $start;
    $memory = memory_get_usage() / 1024 / 1024;
    $output->writeln("Finished iteration $i in {$end}s. Memory usage: {$memory}MB");
}

运行后内存持续上升，示例输出如下：

Finished iteration 0 in 0.88457703590393s. Memory usage: 65.843116760254MB
Finished iteration 1 in 0.8521249294281s. Memory usage: 75.336936950684MB
Finished iteration 2 in 0.8602819442749s. Memory usage: 81.830673217773MB
Finished iteration 3 in 0.87028503417969s. Memory usage: 94.406578063965MB
Finished iteration 4 in 0.86899781227112s. Memory usage: 101.90029144287MB
Finished iteration 5 in 0.85589599609375s. Memory usage: 107.39400482178MB
Finished iteration 6 in 0.86256003379822s. Memory usage: 126.88771820068MB
...etc...

已排查要点：

• Message仅包含一个int类型id；
• 问题出在消息调度阶段，与消息处理无关；
• 禁用TraceableMessageBus后问题仍存在（内存占用更高）；
• 使用RabbitMQ传输；
• 采用默认消息总线配置；
• 仅保留send_message中间件时，内存每次迭代增长3MB（原10MB），配置如下：

framework:
    messenger:
        buses:
            messenger.bus.default:
                default_middleware: false
                middleware:
                    - messenger.middleware.send_message

• 尝试禁用SendMessageMiddleware中的logger，内存泄漏仍未解决。

请问该内存泄漏的原因是什么？若属于可接受开销，批量调度大量消息时应采用哪些最佳实践避免服务器崩溃？

内存增长原因 #

1. RabbitMQ客户端资源累积：底层AMQP客户端（如php-amqplib）在单条消息调度时，会产生临时协议帧、消息元数据缓存等对象，部分资源无法被PHP垃圾回收器及时回收；即使复用信道，客户端内部的缓存结构也会随消息量逐步膨胀。
2. 消息总线上下文留存：仅保留send_message中间件时，消息总线仍会创建临时上下文对象、消息包装器，这类对象可能因循环引用或GC延迟，无法在每次迭代后被完全清理，导致内存持续堆积。
3. PHP GC机制局限性：PHP垃圾回收器对循环引用的处理存在延迟，当大量短生命周期对象（如Message实例、传输上下文）被快速创建时，回收不及时会造成内存占用逐步上升。

批量调度最佳实践 #

• 分批次+手动触发GC：在每次迭代结束后调用gc_collect_cycles()强制触发垃圾回收，同时可适当缩小批次大小（如调整为5000条/批），减少内存瞬时压力：

foreach ($ids as $id) {
    $this->messageBus->dispatch(new Message((int) $id));
}
// 手动触发垃圾回收
gc_collect_cycles();

• 复用消息实例：避免每次创建新的Message对象，复用同一实例并修改id值，减少对象创建带来的内存开销：

$message = new Message(0);
foreach ($ids as $id) {
    $message->setId((int) $id);
    $this->messageBus->dispatch($message);
}

• 使用RabbitMQ批量发布API：绕过Messenger单条dispatch逻辑，直接调用AMQP客户端的批量发布方法（如php-amqplib的batch_publish），一次性发送多条消息，大幅降低内存占用：

$batch = new AMQPBatchExchange($channel);
foreach ($ids as $id) {
    $amqpMsg = new AMQPMessage(json_encode(['id' => $id]));
    $batch->publish($amqpMsg, 'your_exchange_name');
}
$batch->publishBatch();

• 进程重启机制：将大规模调度拆分为多个独立进程，每处理一定批次后自动重启进程。可通过Supervisor管理调度进程，设置进程生命周期，避免内存无限累积。
• 调整PHP运行参数：临时调高memory_limit确保调度不中途崩溃，同时开启zend.enable_gc=1并设置gc_probability=1，提高垃圾回收触发频率。

内容的提问来源于stack exchange，提问作者Johnson55