PHP消息队列怎么用_PHP中RabbitMQ或Kafka消息队列集成使用-php教程-PHP中文网

PHP消息队列怎么用_PHP中RabbitMQ或Kafka消息队列集成使用

看不見的法師

发布： 2025-10-10 23:23:01

原创

715人浏览过

答案：PHP中使用RabbitMQ和Kafka实现消息队列，RabbitMQ通过php-amqplib库支持异步任务如邮件发送，Kafka借助rdkafka扩展适用于高吞吐日志处理；建议RabbitMQ用于业务解耦，Kafka用于数据流分析，可结合使用并保障消息可靠性。

php消息队列怎么用_php中rabbitmq或kafka消息队列集成使用

在PHP项目中使用消息队列，主要是为了实现系统解耦、异步处理和流量削峰。RabbitMQ 和 Kafka 是两种主流的消息中间件，适用于不同场景。下面分别介绍它们在 PHP 中的集成方法。

RabbitMQ 在 PHP 中的使用

RabbitMQ 是基于 AMQP 协议的轻量级消息队列，适合任务分发、邮件发送、日志处理等场景。PHP 可通过 php-amqplib 库与 RabbitMQ 交互。

安装依赖：

使用 Composer 安装 php-amqplib：

composer require php-amqplib/php-amqplib

立即学习“PHP免费学习笔记（深入）”；

发送消息（Producer）：

创建一个生产者脚本，将消息发送到队列：

require_once 'vendor/autoload.php';
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;

$connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
$channel = $connection->channel();

$channel->queue_declare('task_queue', false, true, false, false);

$message = new AMQPMessage('Hello World!', [
    'delivery_mode' => 2, // 消息持久化
]);

$channel->basic_publish($message, '', 'task_queue');

echo " [x] Sent 'Hello World!'\n";

$channel->close();
$connection->close();

登录后复制

接收消息（Consumer）：

消费者脚本持续监听队列并处理消息：

require_once 'vendor/autoload.php';
use PhpAmqpLib\Connection\AMQPStreamConnection;

$connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
$channel = $connection->channel();

$channel->queue_declare('task_queue', false, true, false, false);

echo " [*] Waiting for messages. To exit press CTRL+C\n";

$callback = function ($msg) {
    echo " [x] Received ", $msg->body, "\n";
    // 模拟耗时任务
    sleep(2);
    echo " [x] Done\n";
    $msg->ack(); // 手动确认
};

$channel->basic_consume('task_queue', '', false, false, false, false, $callback);

while ($channel->is_consuming()) {
    $channel->wait();
}

$channel->close();
$connection->close();

登录后复制

Kafka 在 PHP 中的使用

Kafka 更适合高吞吐、大数据流处理场景，如日志收集、行为追踪。PHP 使用 Kafka 需借助 rdkafka 扩展，底层基于 librdkafka。

序列猴子开放平台

具有长序列、多模态、单模型、大数据等特点的超大规模语言模型

查看详情

安装 rdkafka 扩展：

需先安装 C 库，再安装 PHP 扩展：

Ubuntu: sudo pecl install rdkafka
配置 php.ini 添加：extension=rdkafka.so

发送消息（Producer）：

<?php
$rk = new RdKafka\Producer();
$rk->setLogLevel(LOG_DEBUG);
$rk->addBrokers("127.0.0.1");

$topic = $rk->newTopic("test");

for ($i = 0; $i < 10; $i++) {
    $topic->produce(RD_KAFKA_PARTITION_UA, 0, "Message $i");
    $rk->poll(0);
}

while ($rk->getOutQLen() > 0) {
    $rk->poll(50);
}
?>

登录后复制

接收消息（Consumer）：

<?php
$rk = new RdKafka\Consumer();
$rk->setLogLevel(LOG_DEBUG);
$rk->addBrokers("127.0.0.1");

$topic = $rk->newTopic("test");
$topic->consumeStart(0, RD_KAFKA_OFFSET_BEGINNING);

while (true) {
    $msg = $topic->consume(0, 1000);
    if ($msg === null) continue;
    if ($msg->err) {
        echo "Error: " . $msg->errstr(), "\n";
        break;
    } else {
        echo $msg->payload, "\n";
    }
}
?>

登录后复制