当前位置：首页 > PHP

php 消息队列实现

2026-02-13 16:58:40PHP

消息队列的基本概念

消息队列是一种异步通信机制，允许应用程序通过发送和接收消息来解耦生产者和消费者。在 PHP 中，常见的消息队列实现包括 Redis、RabbitMQ、Kafka 和 Beanstalkd。

使用 Redis 实现消息队列

Redis 的 LPUSH 和 BRPOP 命令可以模拟简单的队列功能。

安装 Redis 扩展：

pecl install redis

生产者代码示例：

<?php
$redis = new Redis();
$redis->connect('127.0.0.1', 6379);
$redis->lPush('task_queue', 'Task data');
?>

消费者代码示例：

<?php
$redis = new Redis();
$redis->connect('127.0.0.1', 6379);
while (true) {
    $task = $redis->brPop('task_queue', 0);
    echo "Processing: " . $task[1] . "\n";
}
?>

使用 RabbitMQ 实现消息队列

RabbitMQ 是一个功能完善的消息队列系统，支持复杂的路由和消息确认机制。

安装 AMQP 扩展：

pecl install amqp

生产者代码示例：

<?php
$connection = new AMQPConnection(['host' => '127.0.0.1']);
$connection->connect();
$channel = new AMQPChannel($connection);
$queue = new AMQPQueue($channel);
$queue->setName('task_queue');
$queue->declareQueue();
$queue->publish('Task data');
?>

消费者代码示例：

<?php
$connection = new AMQPConnection(['host' => '127.0.0.1']);
$connection->connect();
$channel = new AMQPChannel($connection);
$queue = new AMQPQueue($channel);
$queue->setName('task_queue');
$queue->declareQueue();
$queue->consume(function ($message) {
    echo "Processing: " . $message->getBody() . "\n";
    return true; // 确认消息
});
?>

使用 Beanstalkd 实现消息队列

Beanstalkd 是一个轻量级的消息队列系统，适合简单的任务分发。

安装 Pheanstalk 库：

composer require pda/pheanstalk

生产者代码示例：

<?php
require 'vendor/autoload.php';
$queue = new Pheanstalk\Pheanstalk('127.0.0.1');
$queue->put('Task data');
?>

消费者代码示例：

php 消息队列实现

<?php
require 'vendor/autoload.php';
$queue = new Pheanstalk\Pheanstalk('127.0.0.1');
while (true) {
    $job = $queue->watch('default')->reserve();
    echo "Processing: " . $job->getData() . "\n";
    $queue->delete($job);
}
?>