当前位置：首页 > PHP

php实现redis消息队列

2026-02-13 19:24:35PHP

Redis 消息队列实现方法

PHP 中可以使用 Redis 实现消息队列，Redis 提供了多种数据结构来支持队列操作，如列表（List）、发布/订阅（Pub/Sub）和有序集合（Sorted Set）。以下是几种常见的实现方式：

使用 Redis List 实现队列

Redis 的 List 数据结构支持队列的基本操作（FIFO），可以通过 LPUSH 和 RPOP 实现生产者-消费者模式。

生产者代码示例：

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

$queueName = 'message_queue';
$message = json_encode(['data' => 'Hello, Redis Queue!']);

// 将消息推入队列头部
$redis->lPush($queueName, $message);
echo "Message pushed to queue.\n";
?>

消费者代码示例：

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

$queueName = 'message_queue';

// 从队列尾部取出消息
while (true) {
    $message = $redis->rPop($queueName);
    if ($message) {
        $data = json_decode($message, true);
        echo "Processing message: " . $data['data'] . "\n";
    }
    sleep(1); // 避免空轮询
}
?>

使用 Redis Pub/Sub 实现发布订阅

Redis 的发布订阅模式适用于广播消息的场景，多个消费者可以订阅同一个频道。

发布者代码示例：

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

$channel = 'message_channel';
$message = json_encode(['data' => 'Hello, Redis Pub/Sub!']);

$redis->publish($channel, $message);
echo "Message published to channel.\n";
?>

订阅者代码示例：

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

$channel = 'message_channel';

$redis->subscribe([$channel], function ($redis, $channel, $message) {
    $data = json_decode($message, true);
    echo "Received message from $channel: " . $data['data'] . "\n";
});
?>

使用 Redis Sorted Set 实现延迟队列

有序集合（Sorted Set）可以用于实现延迟队列，通过设置消息的分数（时间戳）来控制消息的消费时间。

生产者代码示例：

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

$queueName = 'delayed_queue';
$message = json_encode(['data' => 'Delayed message!']);
$delay = time() + 10; // 10秒后消费

$redis->zAdd($queueName, $delay, $message);
echo "Delayed message added to queue.\n";
?>

消费者代码示例：

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

$queueName = 'delayed_queue';

while (true) {
    $now = time();
    // 获取分数（时间戳）小于当前时间的消息
    $messages = $redis->zRangeByScore($queueName, 0, $now);
    if (!empty($messages)) {
        foreach ($messages as $message) {
            $data = json_decode($message, true);
            echo "Processing delayed message: " . $data['data'] . "\n";
            $redis->zRem($queueName, $message); // 移除已处理消息
        }
    }
    sleep(1); // 避免空轮询
}
?>