PHP项目Redis列表如何实现队列

本文目录导读：

1. 核心原理
2. 基础实现（不使用任何框架）
3. 进阶实现：使用阻塞式弹出（BRPOP）解决轮询问题
4. 实现可靠队列：防止任务丢失（关键）
5. 生产环境中的最佳实践
6. 总结对比

在 PHP 项目中使用 Redis 列表实现队列是一个非常经典且高效的做法，Redis 的列表（List）数据类型天然具有双端操作的特性，非常适合用来构建轻量级的消息队列。

以下是实现的详细步骤和最佳实践。

核心原理

• 生产者：使用 LPUSH 命令将任务（消息）从列表的左侧推入。
• 消费者：使用 RPOP 命令从列表的右侧取出并移除任务。

这种方式遵循了 FIFO（先进先出）原则，保证了消息的处理顺序。

基础实现（不使用任何框架）

假设你已经连接了 Redis（这里使用 predis 库作为示例，你也可以使用 phpredis 扩展）。

生产者（推送任务）

<?php
require 'vendor/autoload.php';
use Predis\Client as RedisClient;
// 连接 Redis
$redis = new RedisClient([
    'scheme' => 'tcp',
    'host'   => '127.0.0.1',
    'port'   => 6379,
]);
$queueKey = 'my_queue';
// 模拟任务数据
$task1 = json_encode(['user_id' => 1, 'action' => 'send_email', 'data' => 'hello']);
$task2 = json_encode(['user_id' => 2, 'action' => 'process_image', 'data' => 'img_01.jpg']);
// LPUSH：将任务推入列表左侧
$redis->lpush($queueKey, [$task1, $task2]); // 可以一次性推入多个
echo "任务已发送\n";

消费者（处理任务）

<?php
require 'vendor/autoload.php';
use Predis\Client as RedisClient;
$redis = new RedisClient([
    'scheme' => 'tcp',
    'host'   => '127.0.0.1',
    'port'   => 6379,
]);
$queueKey = 'my_queue';
// 无限循环，持续监听队列
while (true) {
    // RPOP：从列表右侧弹出一个任务（非阻塞）
    $task = $redis->rpop($queueKey);
    if ($task) {
        $data = json_decode($task, true);
        echo "处理任务：\n";
        print_r($data);
        // 这里处理你的业务逻辑，比如发送邮件、处理图片等
        // processTask($data);
        sleep(1); // 模拟耗时操作
    } else {
        // 队列为空时，等待一段时间再检查，避免 CPU 空转
        echo "队列为空，等待1秒...\n";
        sleep(1);
    }
}

基础实现的局限性

1. 轮询开销：消费者需要 sleep 来避免空耗 CPU，但这会导致任务处理的延迟。
2. 任务丢失风险：如果消费者在处理任务时崩溃，任务就会丢失（因为已经 RPOP 了）。

进阶实现：使用阻塞式弹出（BRPOP）解决轮询问题

Redis 提供了 BRPOP 命令，它可以实现阻塞弹出，如果列表为空，连接会一直阻塞直到有新数据到来或超时。

消费者（使用 BRPOP）

<?php
require 'vendor/autoload.php';
use Predis\Client as RedisClient;
$redis = new RedisClient([
    'scheme' => 'tcp',
    'host'   => '127.0.0.1',
    'port'   => 6379,
]);
$queueKey = 'my_queue';
while (true) {
    // BRPOP 返回一个数组：['key_name', 'value']
    // 第二个参数是超时时间（秒），0 表示无限等待
    $result = $redis->brpop($queueKey, 0);
    // 如果正常返回，$result 不为 null，且包含 [队列名, 消息内容]
    if ($result) {
        $task = json_decode($result[1], true);
        echo "收到任务：\n";
        print_r($task);
        // 处理任务
        // processTask($task);
    }
}

优点：

• 实时性高：消息一旦到达，消费者立即被唤醒处理。
• 节省资源：消费者进程在等待时几乎不占用 CPU。

实现可靠队列：防止任务丢失（关键）

上面的实现中,如果消费者在 RPOP/BRPOP 之后、处理完成之前崩溃，任务就丢了，解决方案是使用 RPOPLPUSH 或 BRPOPLPUSH。

核心思想：

1. 从主队列 queue 中弹出任务。
2. 立即将任务推到备份队列 queue_backup 中。
3. 处理任务。
4. 处理完毕后,从备份队列中移除任务。

如果消费者崩溃,备份队列中会留下未完成的任务，重启或监控脚本可以重新处理。

消费者（使用 BRPOPLPUSH）

<?php
require 'vendor/autoload.php';
use Predis\Client as RedisClient;
$redis = new RedisClient([
    'scheme' => 'tcp',
    'host'   => '127.0.0.1',
    'port'   => 6379,
]);
$mainQueue = 'queue';
$backupQueue = 'queue_backup';
while (true) {
    // BRPOPLPUSH：从 mainQueue 右侧弹出，推入 backupQueue 左侧
    $task = $redis->brpoplpush($mainQueue, $backupQueue, 0);
    if ($task) {
        $data = json_decode($task, true);
        echo "处理任务...\n";
        try {
            // 执行业务逻辑
            // processTask($data);
            // 处理成功：从备份队列中移除任务
            $redis->lrem($backupQueue, -1, $task); // 移除最后一张（即本任务）
            echo "任务完成，已从备份队列移除\n";
        } catch (Exception $e) {
            // 处理失败：将任务放回主队列重试，或记录到死信队列
            echo "任务失败，放回主队列: " . $e->getMessage() . "\n";
            $redis->lpush($mainQueue, $task); // 放回队尾或其他位置
            // 注意：需要从 backupQueue 中移除失败的，否则会重复
            $redis->lrem($backupQueue, -1, $task);
        }
    }
}

注意：PHP 的 predis 库 brpoplpush 方法在较新版本可能不支持超时参数 0，如果遇到问题可以用 rpoplpush 配合 sleep 轮询。

生产环境中的最佳实践

1. 消息序列化：不要直接存储复杂对象，使用 json_encode / json_decode 或 serialize / unserialize，JSON 更通用、跨语言兼容。
2. 死信队列：对于处理多次仍失败的消息，放入一个专门的队列（如 queue_dead），由人工介入检查，避免阻塞后续任务。
3. 延迟队列：如果需要延迟执行任务，可以使用 Redis 的有序集合（ZSet）或专门的库（如 rabbitmq）。
4. 多消费者：可以启动多个消费者实例，Redis 的 RPOP / BRPOP 是原子操作，天然支持单条消息只被一个消费者消费（适用于负载均衡）。
5. 数据量控制：建议控制单个消息体大小，避免过大占用内存（例如日志场景不要直接将大日志文件内容放进去）。
6. 超时设置：生产者的连接、消费者的 BRPOP 超时时间要根据业务合理设置。

总结对比

对于大多数 PHP 项目（如 ThinkPHP、Laravel），Redis 列表实现队列已经足够轻量和高效，如果需要更复杂的功能（如延迟任务、死信交换机、消息确认），建议考虑使用 RabbitMQ 或 Apache Kafka 等专门的消息中间件。