Redis List实现简单消息队列的BRPOP阻塞读取

Redis List 是构建轻量级消息队列的极佳数据结构，其中的 BRPOP 命令提供了“阻塞式读取”的能力。相比于轮询，这种方式能极大降低 CPU 消耗。以下将直接演示如何利用这一特性实现生产者-消费者模型。

第一阶段：命令行实操体验阻塞效果

在编写代码之前，先通过命令行直观理解 BRPOP 的工作原理。

1. 打开 终端 A，输入 以下命令连接 Redis 并进入阻塞状态：

   redis-cli
   BRPOP my_queue 0

   此处 my_queue 是列表名，0 表示超时时间为 0 秒（即永久阻塞，直到有数据到来）。此时终端 A 会处于“挂起”等待状态，无任何返回。

2. 保持 终端 A 的等待状态，打开 一个新的终端 B，输入 以下命令发送一条消息：

   redis-cli
   RPUSH my_queue "task_001"

3. 观察 终端 A，会发现它立即从挂起状态恢复，并输出了刚才写入的数据：

   1) "my_queue"
   2) "task_001"

   这说明 BRPOP 成功“监听”到了数据的插入。

第二阶段：编写生产者代码

生产者负责将任务推送到列表的右侧（尾部）。这里使用 Python 进行演示。

1. 安装 Redis 的 Python 客户端库：

   pip install redis

2. 创建 文件 producer.py，编写 以下代码：

   import redis
   import time
   
   # 连接 Redis
   r = redis.Redis(host='localhost', port=6379, db=0)
   
   def send_task(queue_name, task_content):
       # 将任务推送到列表右侧
       r.rpush(queue_name, task_content)
       print(f"[生产者] 已发送任务: {task_content}")
   
   if __name__ == "__main__":
       # 模拟发送 5 个任务
       for i in range(1, 6):
           send_task("my_queue", f"data_payload_{i}")
           time.sleep(1)  # 间隔 1 秒

3. 运行 该脚本，它将向 my_queue 中放入 5 条数据：

   python producer.py

第三阶段：编写消费者代码

消费者是核心部分，使用 BRPOP 从列表左侧（头部）阻塞式拉取数据。

1. 创建 文件 consumer.py，编写 以下代码：

   import redis
   
   # 连接 Redis
   r = redis.Redis(host='localhost', port=6379, db=0)
   
   def process_task(task_data):
       # 模拟业务处理逻辑
       print(f"[消费者] 正在处理: {task_data} ...")
       # 假设处理耗时 0.5 秒
       time.sleep(0.5) 
       print(f"[消费者] 完成处理: {task_data}")
   
   def listen_for_tasks(queue_name, timeout=0):
       print(f"[消费者] 启动监听，队列: {queue_name}...")
       while True:
           # brpop 返回一个元组：(队列名, 数据内容)
           # timeout=0 表示无限期阻塞
           result = r.brpop(queue_name, timeout=timeout)
   
           if result:
               _, data = result
               # Redis 返回的是字节类型，需 decode
               process_task(data.decode('utf-8'))
           else:
               print("[消费者] 等待超时（若设置了超时时间）")
   
   import time # 需要在文件头部引入 time 模块
   
   if __name__ == "__main__":
       listen_for_tasks("my_queue")

2. 运行 该脚本：

   python consumer.py

3. 观察 输出。如果队列中已有第一步产生的数据，消费者会立即处理；如果队列为空，程序会停在这一行 result = r.brpop(...) 等待，不消耗 CPU。

第四阶段：理解关键机制

为了更好地利用 BRPOP，需要掌握其两个核心特性：超时控制与多消费者竞争。

1. 超时控制

BRPOP 的第二个参数控制阻塞时长。

• 设置 为 0：表示一直等，直到有数据或连接断开。适用于必须处理完所有任务的后台守护进程。
• 设置 为 5：表示等 5 秒，如果还没数据就返回 None。适用于需要定期做其他检查（如心跳检测）的逻辑。

2. 多消费者负载均衡

当有多个消费者同时执行 BRPOP 监听同一个队列时，Redis 会遵循“先阻塞先获取”的公平分发原则（FIFO）。这天然实现了简单的负载均衡。

以下流程图描述了多消费者同时阻塞时的竞争与获取逻辑：

第五阶段：普通 POP 与 阻塞 POP 对比

为了明确为什么选择 BRPOP 而非 RPOP，请参考下表：

第六阶段：异常处理建议

在实际生产环境中，必须考虑消费者在处理任务时崩溃的情况。

1. 确认 问题：如果消费者拿到了任务（BRPOP 返回数据，Redis 已将数据移除），但在 process_task 中程序崩溃，该任务就永久丢失了。

2. 解决 方案：使用 RPOPLPUSH 命令或其阻塞版本 BRPOPLPUSH。

   • 该命令在从源列表弹出数据的同时，将数据推入到一个“备份列表”中。
   • 只有当业务逻辑处理完毕后，再手动从备份列表中删除该任务。
   • 如果程序崩溃，备份列表中仍有数据，可以通过另一套脚本进行重试或告警。

使用 BRPOPLPUSH 的核心代码修改如下：

# 备份队列名
BACKUP_QUEUE = "my_queue_backup"

# 在消费者循环中
while True:
    # 从 my_queue 取出放入 my_queue_backup
    result = r.brpoplpush("my_queue", BACKUP_QUEUE, timeout=0)

    if result:
        data = result.decode('utf-8')
        try:
            process_task(data)
            # 处理成功，从备份队列中移除
            r.lrem(BACKUP_QUEUE, 0, result) 
        except Exception as e:
            print(f"处理失败，任务保留在备份队列: {e}")
            # 这里可以增加告警逻辑