在使用 Redis 做消息队列时，最常见的简单方案是直接对列表执行 LPOP 或 RPOP。生产者不断把消息推入列表，消费者用轮询方式反复取出。这个方案代码量小，但有两个明显问题：一是轮询会浪费 CPU 并且增加客户端与 Redis 的往返次数；二是一旦消费者取出消息后崩溃，这条消息就从列表中消失了，没有被任何其他消费者处理。如果业务允许丢消息，这个方案还勉强可用，但对于订单、通知、任务分派等场景，消息丢失意味着数据不一致。要同时解决阻塞等待和消息可靠性，Redis 提供了一个专门命令 BRPOPLPUSH。

BRPOPLPUSH 的命令语义与阻塞行为

BRPOPLPUSH 可以看作 RPOPLPUSH 的阻塞版本。它的完整语法是 BRPOPLPUSH source destination timeout，其中 source 是源列表，destination 是备份列表，timeout 是等待超时时间，单位为秒。该命令执行时，Redis 会尝试从 source 的尾部弹出最后一个元素，如果 source 不为空，则立即将该元素插入 destination 的头部，并把这个元素返回给客户端。如果 source 为空，命令不会立刻返回 nil，而是阻塞当前客户端，直到有新的元素进入 source 或者等待时间超过 timeout。

原子性是值得强调的一点。整个弹出、写入、返回的过程在 Redis 服务端一次性完成，中间不会插入其他客户端的命令。相比之下，如果手动组合 RPOP 和 LPUSH，中间存在一个窗口：A 客户端 RPOP 拿到消息后，还没来得及 LPUSH 到备份队列就崩溃了，消息就会丢失。BRPOPLPUSH 把两步合并成一步，因此能避免这种非原子操作造成的问题。

从返回值看，命令正常处理消息时返回字符串元素；超时未获取到任何元素时返回 nil。需要注意的是，超时返回 nil 与列表为空时 RPOPLPUSH 返回 nil 的语义不同，因为 BRPOPLPUSH 是在超时后才返回 nil，客户端可以根据返回值是否为 nil 来决定是继续下一次阻塞等待还是先处理心跳或检查条件。另一个细节是 timeout 设置为 0 表示无限阻塞，如果客户端希望一直等待消息，可以采用这个值，但要配合连接池和心跳机制，避免连接被中间设备断开。

下面先用 redis-cli 演示一次基本操作。开两个终端，一个模拟消费者，另一个模拟生产者。

# 消费者：从队列 task_queue 阻塞取出并备份到 task_queue_processing
127.0.0.1:6379> BRPOPLPUSH task_queue task_queue_processing 0

此时消费者处于阻塞状态。生产者执行 LPUSH 后，消费者会立即返回元素。

# 生产者：向 task_queue 左侧推入一条消息
127.0.0.1:6379> LPUSH task_queue "process_order_1001"
(integer) 1

消费者端会返回 process_order_1001，同时该元素已经从 task_queue 弹出，并被原子地写入了 task_queue_processing。此时可以查看两个列表确认：

127.0.0.1:6379> LRANGE task_queue 0 -1
(empty array)
127.0.0.1:6379> LRANGE task_queue_processing 0 -1
1) "process_order_1001"

用备份队列实现至少一次投递

可靠队列模式的核心思想是：消费者每次取消息时都同时把消息放入一个私有或共享的备份队列，只有当消息被业务逻辑处理成功后，才从备份队列中删除它。BRPOPLPUSH 的 destination 参数就是用来承担这个备份角色的。常见的命名方式是把源队列称为 task_queue，把备份队列称为 task_queue_processing 或 task_queue_inflight。备份队列中的消息代表正在处理中但尚未确认完成的那些消息。

为什么需要 ACK 删除？因为消息进入备份队列后，如果不主动删除，备份队列会无限增长。消费者处理成功后的最后一步应当是执行 LREM 从备份队列中移除这条消息。LREM 默认从头部开始移除指定数量的匹配元素。由于备份队列中可能有多条内容相同的消息，我们的业务消息最好带有唯一 ID，删除时精确匹配该 ID。这样可以避免误删其他同内容的消息。

下面的 Python 示例展示了完整的消费循环：消费者使用 BRPOPLPUSH 阻塞取出消息，打印或调用业务处理函数，成功后在 else 分支中执行 LREM 删除备份。如果处理函数抛出异常，消息会保留在备份队列中，等待恢复程序重新入队或人工处理。

import redis
client = redis.Redis(host='127.0.0.1', port=6379, db=0)
SOURCE_QUEUE = 'task_queue'
BACKUP_QUEUE = 'task_queue_processing'
def handle_message(message):
    # 实际业务处理逻辑，这里仅打印
    print(f"processing: {message}")
while True:
    # 阻塞等待消息，超时设为 5 秒，便于周期性执行心跳或日志
    message = client.brpoplpush(SOURCE_QUEUE, BACKUP_QUEUE, timeout=5)
    if message is None:
        print("waiting...")
        continue
    try:
        handle_message(message)
    except Exception as exc:
        # 处理失败，保留备份队列中的消息，后续由恢复程序处理
        print(f"handle error: {exc}")
    else:
        # 处理成功，删除备份队列中的消息，确认 ACK
        client.lrem(BACKUP_QUEUE, 0, message)

这个循环中有几个容易忽略的点。一是 LREM 的 count 参数，这里设置为 0 表示删除所有与 message 相等的元素。如果消息可能重复，更好的做法是为每条消息生成唯一 ID，并在备份队列中存储带有 ID 的 JSON 或序列化字符串，ACK 时按 ID 精确删除。二是 BRPOPLPUSH 的超时设定为 5 秒，而不是 0。虽然 0 可以无限等待，但客户端可能在 Redis 连接空闲时被代理或防火墙断开；周期性地超时返回并重新发起命令，反而有利于保持连接活跃和及时响应关闭信号。

当消费者崩溃后，备份队列中的消息不会被自动处理。我们需要一个恢复器来扫描备份队列，把滞留过久的消息重新放回源队列。滞留时间的阈值通常根据业务允许的最大处理时长来设定，比如 10 分钟。恢复器可以是一个定时任务，先读取备份队列中的所有元素及其入队时间，然后把超过阈值的消息 LPUSH 回源队列，并从备份队列中移除。下面给出一个简化的恢复函数：

import time
def recover_stale_messages(client, source, backup, max_age_seconds=600):
    # 备份队列中保存字符串，格式为 id:timestamp
    messages = client.lrange(backup, 0, -1)
    now = time.time()
    for raw in messages:
        try:
            msg_id, ts = raw.decode().split(':')
            if now - float(ts) > max_age_seconds:
                # 重新放回源队列，并从备份队列移除
                client.lpush(source, raw)
                client.lrem(backup, 0, raw)
        except ValueError:
            # 格式不符的消息跳过或记录日志
            continue

阻塞式可靠队列的性能与替代方案

BRPOPLPUSH 的时间复杂度为 O(1)，因为它只操作列表的头部和尾部。即便在源列表为空时，Redis 也不会通过忙循环来等待，而是把客户端放入一个阻塞等待队列中，当有新元素写入源列表时再唤醒这些客户端。因此，一个 Redis 实例可以同时服务成千上万个阻塞客户端，而不会占满 CPU。不过要注意，Redis 是单线程处理命令的，执行 LREM 删除备份队列中的大元素时可能造成短暂的阻塞，因为 LREM 需要遍历列表。如果业务消息体很大，建议备份队列只存消息 ID 或较短的元数据，实际内容放入其他存储，以减少 LREM 的遍历成本。

这种模式给出的可靠性是至少一次投递。也就是说，消息可能会被重复处理，但不会丢失。重复的来源主要是消费者处理成功后，还没来得及执行 LREM 就崩溃了，恢复器之后会把该消息重新入队，导致第二个消费者再次处理同一消息。如果业务不能容忍重复，就需要在业务侧实现幂等，比如根据消息唯一 ID 去重，或者用数据库事务和状态字段来控制。这也是所有可靠队列模式共同面对的问题，并非 BRPOPLPUSH 特有的缺陷。

与 Redis 5.0 以后提供的 Streams 消费组相比，BRPOPLPUSH 的可靠队列显得更轻量，也更直观。Streams 的 XREADGROUP 命令原生支持消费者组、ACK 和 Pending 列表，功能更完善，但学习成本和数据结构的复杂度更高。如果项目已经在使用 Redis 5.0 以上版本，并且需要水平扩展消费者、消息重放、消费组管理，Streams 是更合适的选择。如果只需要一个简单、可靠、容易调试的队列，且 Redis 版本较旧或不想引入 Streams 概念，BRPOPLPUSH 仍然是一个有效的方案。下面用对比表说明两种方式的差异：

最后要注意一个常见误区：不要因为备份队列名为 processing 就消费它。处理中的消息只能由恢复器检查，不能由普通消费者再次 BRPOPLPUSH 取出，否则会破坏等待条件和重复投递逻辑。此外，如果多个消费者共享同一个备份队列，恢复器的 LREM 删除必须精确匹配唯一消息，否则可能误删其他消费者正在处理的消息。更稳妥的做法是让每个消费者使用自己的备份队列，比如 task_queue_processing:consumer_1，这样备份队列之间天然隔离，恢复程序也能按消费者进行监控。

综合来看，Redis BRPOPLPUSH 提供了一种简单可靠的阻塞队列模式。它适合消息量中等、消费者数量相对固定、对重复消息可以通过幂等处理来容忍的场景。通过合理的备份队列命名、LREM 确认删除和定时恢复策略，可以构建一个很小的代码库却具备基本消息不丢失能力的任务分派系统。