ThinkPHP 集成 Redis 队列：从入门到实战技术

ThinkPHP 集成 Redis 队列：从入门到实战技术

一、引言

在分布式系统架构中，异步处理、服务解耦和流量削峰是提升系统性能的核心需求。Redis 作为高性能内存数据库，凭借其丰富的数据结构（如 List、Stream、Sorted Set）和轻量级特性，成为实现队列功能的理想选择。本文将结合 ThinkPHP 框架的特性，详细阐述如何通过 Redis 队列构建高可用、可扩展的异步处理系统，涵盖基础概念、环境配置、实战案例及最佳实践。

二、Redis 队列核心概念解析

2.1 为何选择 Redis 队列？

Redis 队列的核心优势体现在三方面：

极致性能：基于内存操作，单节点支持万级 QPS，满足高并发场景下的实时响应需求。

轻量部署：无需像 Kafka/RabbitMQ 等中间件的复杂配置，可直接通过 PHP 扩展集成，适合中小规模业务快速落地。

结构灵活：提供多种数据结构适配不同业务场景：

◦ FIFO 队列（List）：基于左进右出（LPUSH/RPOP）实现简单异步任务，如订单状态更新。

◦ 优先级队列（Sorted Set）：通过分值（Score）控制任务执行顺序，适用于高优先级订单加急处理。

◦ 持久化队列（Stream）：支持消息持久化、分组消费和确认机制，适合微服务架构下的可靠消息传递。

2.2 核心数据结构对比

数据结构

特性

典型场景

Redis 核心命令

ThinkPHP 操作示例

List

先进先出，简单高效

短信发送、日志异步写入

lpush/rpop, brpop

$redis->lpush(‘queue:log’, json_encode($log))

Stream

持久化、分组消费

分布式任务调度、消息重试

xadd, xgroup, xreadgroup

$redis->xadd(‘stream:task’, ‘*’, $fields)

Sorted Set

优先级 / 延迟处理

优惠券过期提醒、超时订单取消

zadd, zrange, zrem

$redis->zadd(‘delay:order’, time()+60, $oid)

三、开发环境搭建与配置

3.1 依赖安装

3.1.1 PHP Redis 扩展安装

# 方式一：通过 PECL 安装 phpredis（推荐）

pecl install redis

# 方式二：通过 Composer 安装 Predis（适用于集群环境）

composer require predis/predis

3.1.2 ThinkPHP 配置调整

修改 config/redis.php，配置 Redis 连接参数：

return [

‘default’ => [

‘type’ => ‘redis’,

‘host’ => env(‘REDIS.HOST’, ‘127.0.0.1’), // 支持环境变量注入

‘port’ => env(‘REDIS.PORT’, 6379),

‘password’ => env(‘REDIS.PASS’, ”),

‘select’ => 0, // 数据库索引（0-15）

‘timeout’ => 5, // 连接超时时间（秒）

‘persistent’ => true, // 开启长连接（生产环境建议启用）

],

// 集群配置示例（适用于高可用场景）

‘cluster’ => [

‘type’ => ‘redis’,

‘mode’ => ‘cluster’,

‘nodes’ => [

[‘host’ => ‘node1.com’, ‘port’ => 6380],

[‘host’ => ‘node2.com’, ‘port’ => 6381],

],

‘password’ => ‘cluster_pass’,

‘timeout’ => 3,

]

];

四、基于 List 的基础队列实战

4.1 队列操作核心代码

4.1.1 入队操作（左压栈）

use think\facade\Cache;

$redis = Cache::store(‘redis’)->handler();

// 存储 JSON 格式任务数据（推荐方式）

$task = [

‘task_id’ => uniqid(),

‘type’ => ‘order_process’,

‘data’ => [‘order_id’ => ‘20231205001’, ‘amount’ => 299.99]

];

$redis->lpush(‘queue:default’, json_encode($task));

4.1.2 出队操作（阻塞式右弹出）

// 消费者脚本专用（阻塞等待任务，避免空轮询）

$result = $redis->brpop(‘queue:default’, 10); // 10 秒超时

if ($result) {

[$queueName, $taskJson] = $result;

$task = json_decode($taskJson, true);

// 执行业务逻辑

$this->handleTask($task);

}

4.2 订单异步处理案例

4.2.1 前端下单接口（控制器）

// app/controller/Order.php

public function submitOrder() {

$orderData = $this->request->post();

// 验证订单数据…

// 入队异步处理

$redis = Cache::store(‘redis’)->handler();

$redis->lpush(‘queue:order’, json_encode([

‘order_id’ => $orderData[‘order_id’],

‘product_id’ => $orderData[‘product_id’],

‘quantity’ => $orderData[‘quantity’]

]));

return json([‘code’ => 200, ‘msg’ => ‘下单成功，系统正在处理’]);

}

4.2.2 后台消费者脚本（scripts/order_consumer.php）

<?php

require __DIR__ . ‘/../../thinkphp/base.php’;

$redis = app(\think\cache\driver\Redis::class)->handler();

while (true) {

$result = $redis->brpop(‘queue:order’, 10);

if (!$result) continue;

$task = json_decode($result[1], true);

try {

// 模拟库存扣减（实际需调用服务）

$this->deductStock($task[‘product_id’], $task[‘quantity’]);

// 模拟物流通知

$this->sendLogisticsNotice($task[‘order_id’]);

echo “[“.date(‘Y-m-d H:i:s’).”] 任务完成：{$task[‘order_id’]}\n”;

} catch (\Exception $e) {

// 重试机制（最多 3 次）

$this->retryTask($task, $e, 3);

}

}

4.2.3 启动消费者服务

# 前台运行（便于调试）

php scripts/order_consumer.php

# 后台守护进程运行

nohup php scripts/order_consumer.php > order.log 2>&1 &

五、基于 Stream 的高级队列应用

5.1 Stream 队列核心特性

持久化存储：消息默认持久化到磁盘，支持重启后继续处理未完成任务。

分组消费：多个消费者组成消费组（Consumer Group），实现任务负载均衡（如多个 worker 节点共同处理订单）。

消息确认机制：通过 XACK 命令标记消息已处理，避免重复执行或数据丢失。

5.2 分布式任务处理示例

5.2.1 创建 Stream 并生产消息

// 生产端：添加带重试次数的任务

$redis->xadd(‘stream:task’, ‘*’, [

‘task_type’ => ‘payment_notify’,

‘order_id’ => ‘20231206001’,

‘retry’ => 0, // 初始重试次数

‘create_at’ => time()

]);

5.2.2 初始化消费者组

$redis->xgroup(‘CREATE’, ‘stream:task’, ‘group_workers’, ‘$’, true);

// 如需消费历史消息，将 ‘$’ 替换为 ‘0-0’

5.2.3 消费组节点处理逻辑

// 消费者节点 1（worker1.php）

$messages = $redis->xreadgroup(

‘GROUP’, ‘group_workers’, ‘worker_1’,

‘STREAMS’, ‘stream:task’, ‘>’ // 获取未确认的消息

);

if ($messages) {

foreach ($messages[0][1] as $msgId => $fields) {

try {

$this->handlePaymentNotify($fields[‘order_id’]);

$redis->xack(‘stream:task’, ‘group_workers’, $msgId); // 确认消息

echo “Worker1 处理：{$fields[‘order_id’]}\n”;

} catch (\Exception $e) {

if ((int)$fields[‘retry’] < 3) {

// 增加重试次数并重新入队

$fields[‘retry’] = (int)$fields[‘retry’] + 1;

$redis->xadd(‘stream:task’, ‘*’, $fields);

} else {

// 记录死信队列

$redis->xadd(‘stream:deadletter’, ‘*’, $fields);

}

}

}

}

六、生产环境最佳实践

6.1 消息序列化规范

强制使用 JSON 格式：

// 推荐做法

$redis->lpush(‘queue’, json_encode($data, JSON_UNESCAPED_UNICODE));

// 禁止使用 PHP 原生序列化

// $redis->lpush(‘queue’, serialize($data));

数据校验：消费端需对反序列化后的数据进行字段校验，避免因格式错误导致服务异常。

6.2 持久化与高可用配置

6.2.1 Redis 持久化策略

AOF 模式：推荐配置 appendfsync everysec，兼顾性能与数据安全性（最多丢失 1 秒数据）。

RDB 备份：定期生成 RDB 快照用于灾难恢复，建议配合云存储（如 S3）实现异地备份。

6.2.2 集群方案

Redis Cluster：适用于超大规模数据，支持自动分片和故障转移。

Sentinel 哨兵模式：监控主从节点状态，自动完成主从切换，配置示例：

// ThinkPHP 哨兵模式配置

‘sentinel’ => [

‘type’ => ‘redis’,

‘mode’ => ‘sentinel’,

‘master’ => ‘mymaster’,

‘sentinels’ => [

[‘host’ => ‘sentinel1.com’, ‘port’ => 26379],

[‘host’ => ‘sentinel2.com’, ‘port’ => 26379],

],

‘password’ => ‘sentinel_pass’,

]