当前位置：首页 > PHP

php 消息队列实现

2026-01-16 11:27:23PHP

PHP 消息队列实现方法

使用 Redis 实现消息队列

Redis 的 LPUSH 和 BRPOP 命令可以轻松实现消息队列功能。生产者通过 LPUSH 将消息推入队列，消费者通过 BRPOP 阻塞等待并消费消息。

// 生产者代码
$redis = new Redis();
$redis->connect('127.0.0.1', 6379);
$redis->lPush('message_queue', json_encode(['data' => 'test message']));

// 消费者代码
while (true) {
    $message = $redis->brPop('message_queue', 0);
    $data = json_decode($message[1], true);
    // 处理消息逻辑
}

使用 RabbitMQ 实现消息队列

RabbitMQ 是一个功能强大的消息队列系统，支持多种消息协议。PHP 可以通过 php-amqplib 库与 RabbitMQ 交互。

require_once __DIR__ . '/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);
$msg = new AMQPMessage('Hello World!');
$channel->basic_publish($msg, '', 'task_queue');

// 消费者代码
$callback = function ($msg) {
    echo 'Received: ', $msg->body, "\n";
    $msg->ack();
};
$channel->basic_consume('task_queue', '', false, false, false, false, $callback);
while ($channel->is_consuming()) {
    $channel->wait();
}

使用 Beanstalkd 实现消息队列

Beanstalkd 是一个轻量级、高性能的消息队列服务，适合简单的任务队列需求。

// 生产者代码
$pheanstalk = new Pheanstalk\Pheanstalk('127.0.0.1');
$pheanstalk->useTube('test')->put('job data');

// 消费者代码
$job = $pheanstalk->watch('test')->ignore('default')->reserve();
$jobData = $job->getData();
$pheanstalk->delete($job);

使用数据库实现消息队列

对于简单的应用场景，可以使用数据库表作为消息队列。创建一个包含状态字段的表，生产者插入消息，消费者轮询并处理。

// 创建消息表
CREATE TABLE message_queue (
    id INT AUTO_INCREMENT PRIMARY KEY,
    message TEXT NOT NULL,
    status ENUM('pending', 'processed') DEFAULT 'pending',
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

// 生产者代码
$pdo = new PDO('mysql:host=localhost;dbname=test', 'username', 'password');
$stmt = $pdo->prepare("INSERT INTO message_queue (message) VALUES (?)");
$stmt->execute(['test message']);

// 消费者代码
$stmt = $pdo->prepare("SELECT * FROM message_queue WHERE status = 'pending' LIMIT 1 FOR UPDATE");
$stmt->execute();
$message = $stmt->fetch(PDO::FETCH_ASSOC);
if ($message) {
    // 处理消息
    $update = $pdo->prepare("UPDATE message_queue SET status = 'processed' WHERE id = ?");
    $update->execute([$message['id']]);
}