利用 Redis 队列实现高并发下的任务处理（PHP/Python示例）

林见深

发布时间：2026-10-15| 4人浏览过

原创

在多服务器部署或高并发场景下，直接对数据库进行行锁可能导致性能瓶颈。本文将演示如何利用 Redis 原子操作的 List 结构作为轻量级分布式队列，结合 PHP 和 Python 代码示例，说明如何通过 LPUSH、LPOP 及 BLPOP 命令实现任务的生产与消费，从而在不增加数据库压力的情况下平衡负载。

为什么选择 Redis 做队列

Redis 是一个高性能的 Key-Value 存储系统，其数据主要驻留在内存中，读写速度极快。相比于 Memcached，Redis 支持更丰富的数据结构，包括列表（List）、集合（Set）、有序集合（ZSet）等，并支持持久化（AOF/RDB）和主从复制。

在处理秒杀活动、邮件发送、日志记录等高并发写请求时，将瞬时压力转化为异步队列处理是常见方案。Redis 的所有操作都是原子性的，这使其天然适合用于解决分布式环境下的计数、状态同步及队列分发问题，避免复杂的数据库锁机制带来的冲突和开销。

基于 PHP 的简单队列实现

以下示例展示了使用 PHP Redis 扩展连接本地 Redis 服务，并进行循环入队和出队的操作。适用于单机测试或理解基本逻辑。

第1步：生产者 - 入队操作

目的：模拟多个生产源向名为 key1 的队列中追加任务数据。
原理：使用 LPUSH 命令将元素插入到列表头部，保持先进先出（FIFO）的顺序（若使用 RPUSH 则方向相反）。

<?php
$redis = new Redis();
$redis->connect('127.0.0.1', 6379);
while(True){ 
    try{
        // 生成带时间戳的唯一任务值
        $value = 'value_'.date('Y-m-d H:i:s'); 
        // 将任务压入列表左侧
        $redis->LPUSH('key1', $value); 
        sleep(rand()%3); // 模拟处理间隔
        echo $value."\n";
    }catch(Exception $e){
        echo $e->getMessage()."\n";
    }
}
?>

第2步：消费者 - 出队操作

目的：从队列中获取任务并执行，直至队列为空后暂停等待。
注意：在生产环境中，建议使用长连接 pconnect 以减少握手开销。

<?php
$redis = new Redis();
$redis->pconnect('127.0.0.1', 6379);
while(True){ 
    try{
        // 从列表右侧弹出一个元素
        echo $redis->LPOP('key1')."\n";
    }catch(Exception $e){
        echo $e->getMessage()."\n";
    }
    sleep(rand()%3);
}?>

优化消费端：使用 BLPOP 防止空转

上述 PHP 出队示例中，如果队列为空，LPOP 会立即返回 false，导致循环频繁轮询，消耗 CPU 资源。在实际的分布式部署（如多台应用服务器共用一个 Redis 实例）中，推荐使用 BLPOP（Block List Pop），它能在队列为空时阻塞等待指定时间。

第3步：改进的消费者代码

入口：独立运行的消费脚本（如 list_pop.php）。
动作：调用 blPop 并设置超时时间，确保只有在有新消息或超时后才继续。

<?php
// 假设已有一个获取 Redis 实例的单例方法
$redis = getRedisInstance(); 
$redis->pconnect('Redis服务器IP', 6379);

while(true) {
    try { 
        // 阻塞等待 10 秒，若有数据则立即返回 ['key', 'value']
        var_export( $redis->blPop('list1', 10) ); 
    } catch(Exception $e) {
        // 处理连接异常
    }
}
?>

Python 实现对照示例

对于 Python 环境，可使用 redis-py 库实现类似功能。关键在于生产和消费必须操作同一个 List key。

第4步：Python 生产者（write.py）

#!/usr/bin/env python
import time
from redis import Redis

# 连接到本地 Redis
redis_client = Redis(host='127.0.0.1', port=6379)

while True: 
    now = time.strftime("%Y/%m/%d %H:%M:%S")
    # lpush 将新数据加入列表左侧
    redis_client.lpush('test_queue', now) 
    time.sleep(1)

第5步：Python 消费者（read.py）

此处展示非阻塞版本的读取逻辑。实际项目中建议同样改用 bpop 系列方法以避免忙等。

#!/usr/bin/env python
import sys
from redis import Redis

redis_client = Redis(host='127.0.0.1', port=6379)

while True: 
    # rpop 从列表右侧弹出数据
    res = redis_client.rpop('test_queue') 
    if res is None: 
        pass # 队列为空，短暂休眠避免CPU空转
    else: 
        print(str(res))
    time.sleep(0.5) # 添加适当延迟

分布式部署注意事项

在真实的多服务器架构中（例如前端通过 F5 负载均衡，后端有多台应用服务器），应将 Redis 单独部署或在独立的服务器集群上运行，与应用服务器解耦。

网络配置：确保所有应用服务器能通过网络访问 Redis 服务器的 IP 和端口。
异常处理：队列操作中应包含完善的 Try-Catch 块，特别是处理网络断开后的重连逻辑。
数据一致性：虽然 Redis 操作是原子的，但若涉及“取出任务-处理任务-保存结果”的业务流程，需确保业务逻辑层面的事务安全，防止任务丢失或重复处理（可通过加锁或幂等设计解决）。

总结

利用 Redis 的 List 结构是实现简易分布式队列的高效方案。PHP 的 LPUSH/LPOP 和 Python 的 lpush/rpop 提供了基础接口，而 BLPOP 则是提升消费者效率的关键。通过这种模式，可以有效缓解数据库压力，平滑流量峰值。以上就是 Redis 队列操作详细内容的介绍，更多关于 PHP 与 Redis 并发处理技术的资料请关注本站其它相关文章！