PHP实现RabbitMQ批量发送消息以提高吞吐量的代码示例-php教程-PHP中文网

直接循环调用basic_publish更慢，因每次调用均触发序列化、网络i/o和broker队列锁竞争；需用confirm_select+批量publish+wait_for_confirms实现可靠批量发送。

php实现rabbitmq批量发送消息以提高吞吐量的代码示例

为什么直接循环调用`basic_publish`反而更慢？

很多人以为“批量发送”就是写个for循环反复调用$channel->basic_publish()，结果发现吞吐量不升反降，甚至比单条还慢。根本原因在于：每次调用都触发一次AMQP协议帧的序列化、网络I/O和Broker端的队列锁竞争。RabbitMQ本身不提供原生的“批量publish API”，所谓“批量”必须靠客户端缓冲+单次网络往返来模拟。

用`confirm_select` + 批量`basic_publish` + `wait_for_confirms`实现可靠批量

这是PHP中兼顾吞吐与可靠性的主流做法：先启用发布确认模式，把多条消息连续发出去（不等待响应），再一次性阻塞等待全部确认。注意，这并非“一条TCP包发多条消息”，而是利用AMQP的流水线机制减少往返延迟。

实操要点：

$channel->confirm_select()必须在发送前调用，且每个$channel只能启用一次
每条basic_publish仍需传入完整AMQPMessage对象，不能合并成一个消息体
$channel->wait_for_confirms(10)的超时值建议设为秒级（如10），避免无限挂起
若返回false，说明有消息未被Broker确认，需结合get_unconfirmed_count()做重试或日志记录

$connection = new AMQPConnection(['host' => 'localhost']);
$channel = $connection->channel();
$channel->confirm_select(); // 启用确认模式

$messages = [
    new AMQPMessage('order_123'),
    new AMQPMessage('order_456'),
    new AMQPMessage('order_789'),
];

foreach ($messages as $msg) {
    $channel->basic_publish($msg, '', 'orders');
}

if (!$channel->wait_for_confirms(10)) {
    error_log('Some messages failed to confirm');
}