php实现redis消息队列
Redis消息队列的基本概念
Redis作为内存数据库，支持多种数据结构如List、Pub/Sub、Stream等，适合实现轻量级消息队列。PHP可通过phpredis或predis扩展与Redis交互。
使用List结构实现队列
Redis的List结构支持LPUSH/RPOP等操作，适合实现FIFO队列。生产者将消息推入列表头部，消费者从尾部取出消息。
// 生产者示例
$redis = new Redis();
$redis->connect('127.0.0.1', 6379);
$redis->lPush('task_queue', json_encode(['task_id' => 123, 'data' => 'example']));
// 消费者示例
while (true) {
$task = $redis->rPop('task_queue');
if ($task) {
$data = json_decode($task, true);
// 处理任务逻辑
}
sleep(1); // 避免空轮询
}使用Stream结构实现高级队列
Redis 5.0+的Stream支持消费者组、消息确认等高级特性，更接近专业消息队列。
// 生产者
$redis->xAdd('order_stream', '*', ['user_id' => 42, 'item' => 'book']);
// 消费者组
$redis->xGroup('CREATE', 'order_stream', 'order_consumers', '$', true);
while (true) {
$messages = $redis->xReadGroup(
'order_consumers', 'worker1', ['order_stream' => '>'], 1
);
if ($messages) {
foreach ($messages['order_stream'] as $id => $message) {
// 处理消息后确认
$redis->xAck('order_stream', 'order_consumers', [$id]);
}
}
}消息可靠性保障
- 持久化处理：启用Redis的AOF或RDB持久化，防止服务器重启丢失消息
- 失败重试：消费者处理失败时将消息重新放入队列
if (!processMessage($task)) { $redis->lPush('task_queue', $task); } - 死信队列：超过重试次数的消息转入专门队列
$failCount = $redis->incr("fail_count:$taskId"); if ($failCount > 3) { $redis->lPush('dead_letter_queue', $task); }
性能优化建议
- 批量操作：使用
pipeline减少网络往返$redis->pipeline()->lPush(...)->lPush(...)->exec(); - 连接池：避免频繁创建连接
- 监控：通过
INFO命令监控队列长度和消费速度
与专业队列系统对比
- 优势：部署简单、延迟低（毫秒级）、支持多种数据结构
- 局限：无严格的消息顺序保证、集群模式下功能受限
- 适用场景：轻量级任务、短时消息存储、开发测试环境
对于生产环境高要求场景，建议评估RabbitMQ或Kafka等专业消息中间件。
