3. Redis List 实现消息队列：LPUSH + BRPOP 原理、生产者消费者模型、代码实战（Python redis-py）

3.1 基于 List 的消息队列模型

Redis 的 List 数据类型天然支持队列操作：生产者使用 LPUSH 从左侧插入消息，消费者使用 BRPOP 从右侧阻塞弹出消息。这一组合构成了最简洁的“左进右出”消息队列，具备以下核心特征：

• 先进先出（FIFO）：LPUSH 将消息压入队列头部，BRPOP 从尾部取出，保证消息按到达顺序被消费。
• 阻塞等待：BRPOP 在队列为空时阻塞连接，直到有新消息到达或超时，避免空轮询浪费 CPU。
• 原子操作：LPUSH 和 BRPOP 均为单命令原子操作，无需额外加锁。

3.2 LPUSH + BRPOP 工作原理

下图展示了 LPUSH + BRPOP 在单队列中的完整数据流：

3.3 生产者消费者模型

在典型的生产者-消费者模式中，多个生产者向同一队列写入消息，多个消费者从队列中竞争消费。Redis List 的 LPUSH + BRPOP 天然支持这种模型：

角色	操作	说明
生产者	LPUSH queue_name message	将消息序列化为字符串（如 JSON），推入队列左端
消费者	BRPOP queue_name timeout	从队列右端阻塞弹出，返回 [队列名, 消息] 元组
多消费者	每个消费者独立调用 BRPOP	消息被其中一个消费者取走，实现负载均衡
异常处理	捕获超时或连接异常	超时返回 None，需重试或重新连接

3.4 Python redis-py 代码实战

以下代码使用 redis-py 库实现完整的生产者-消费者模型：

3.4.1 生产者代码

import redis
import json
import time
import uuid

# 连接 Redis
r = redis.Redis(host='localhost', port=6379, db=0)

def produce_message(queue_name, data):
    """生产消息并推入队列"""
    message = {
        'msg_id': str(uuid.uuid4()),
        'timestamp': time.time(),
        'payload': data
    }
    # LPUSH 将消息推入队列左端
    r.lpush(queue_name, json.dumps(message))
    print(f"[生产者] 已推送消息: {message['msg_id']}")
    return message['msg_id']

# 模拟生产 5 条消息
if __name__ == '__main__':
    for i in range(5):
        produce_message('task_queue', {'task_num': i, 'action': 'process'})
        time.sleep(0.5)

3.4.2 消费者代码

import redis
import json

r = redis.Redis(host='localhost', port=6379, db=0)

def consume_message(queue_name, timeout=0):
    """阻塞消费消息，timeout=0 表示永久阻塞"""
    result = r.brpop(queue_name, timeout=timeout)
    if result is None:
        # 超时返回 None
        print("[消费者] 等待超时，未获取到消息")
        return None
    # result 格式: (queue_name, message_bytes)
    queue, msg_bytes = result
    message = json.loads(msg_bytes)
    print(f"[消费者] 已消费消息: {message['msg_id']}, 任务: {message['payload']}")
    return message

# 持续消费
if __name__ == '__main__':
    while True:
        try:
            msg = consume_message('task_queue', timeout=30)
            if msg:
                # 处理业务逻辑
                print(f"  处理任务: {msg['payload']['action']}")
        except Exception as e:
            print(f"[消费者] 异常: {e}")
            # 重连逻辑
            r = redis.Redis(host='localhost', port=6379, db=0)

3.5 关键参数与避坑指南

参数/场景	推荐值	说明
timeout	30~60 秒	避免永久阻塞导致连接闲置被服务端断开；0 表示永久阻塞
消息序列化	JSON	跨语言兼容，包含元数据便于回溯
消费者数量	≤ 队列写入 QPS	过多消费者会导致大量空阻塞，浪费连接资源
连接池	使用 redis.ConnectionPool	避免每次操作创建新连接，提升性能

3.6 多消费者负载均衡示意

下图展示 2 个消费者从同一队列竞争消费的流程：

每个消费者独立调用 BRPOP，Redis 保证每条消息只被一个消费者取走，实现天然的负载均衡。当消费者处理能力不均时，处理快的消费者会消费更多消息。

3.7 本章小结

• LPUSH + BRPOP 是 Redis List 实现消息队列的核心组合，简单可靠。
• 生产者将消息序列化后 LPUSH 入队，消费者通过 BRPOP 阻塞获取，支持多消费者竞争消费。
• 需关注消息丢失、队列堆积、超时处理等生产环境问题。
• Python redis-py 代码实现清晰，适合快速搭建轻量级消息队列。