Redis 技术笔记

九、实际运行中的注意事项

9.1 任务丢失风险

9.1.1 问题场景

python

result = redis_client.brpop("task:queue", timeout=5)
_, task_id = result
# ← 如果 Worker 在这里崩溃，task_id 已从队列消失但未执行
run_task(int(task_id))

问题：BRPOP 是"弹出即删除"，取出后 Worker 崩溃，任务永久丢失。

9.1.2 涉及函数：`brpop`

python

redis_client.brpop(keys, timeout=5)

参数	类型	必填	说明
`keys`	`str` 或 `list[str]`	是	要监听的队列名。传单个字符串监听 1 个队列，传列表监听多个队列（按列表顺序决定优先级）
`timeout`	`int`	否	最大阻塞秒数。`0` = 永久阻塞。默认 `0`。推荐设为 `5`，确保能响应 `Ctrl+C`

返回值：tuple[str, str] | None

返回值	含义
`("task:queue", "6")`	元组：`[0]` = 命中的队列名，`[1]` = 弹出的元素值
`None`	超时无元素

"弹出即删除"的含义：元素从 Redis 列表中移除的那一刻就不再存在。BRPOP 返回给 Worker 后，Redis 里已经没有这条数据了。如果 Worker 进程在收到返回值后、执行 run_task() 前崩溃，这条任务就永久丢失。

9.1.3 解决方案：RPOPLPUSH 模式

核心思路：弹出时不直接删除，而是原子地"移动"到另一个"处理中"队列。Worker 完成后再从"处理中"队列删除。崩溃后"处理中"队列里残留的任务可以被恢复。

python

# 弹出并放入"处理中"队列
task_id = redis_client.rpoplpush("task:queue", "task:processing")

# 处理完成后从"处理中"队列删除
redis_client.lrem("task:processing", 1, task_id)

# Worker 启动时扫描"处理中"队列，重新推回 task:queue
def recover_stuck_tasks():
    stuck = redis_client.lrange("task:processing", 0, -1)
    for task_id in stuck:
        redis_client.lpush("task:queue", task_id)
        redis_client.lrem("task:processing", 1, task_id)

9.1.4 涉及函数参数详解

rpoplpush(src, dst)

参数	类型	必填	说明
`src`	`str`	是	源队列名，从右端弹出元素
`dst`	`str`	是	目标队列名，推入左端

返回值：str | None

返回值	含义
`"6"`	被移动的元素值
`None`	源队列为空

原子性：弹出和推入在一条 Redis 命令内完成，不可能出现"弹出了但没推入"的中间状态。

数据流向：

task:queue                              task:processing
[8, 7, 6]    rpoplpush    [6] ←── 新推入
                            ↓
[8, 7]                    [6]    Worker 处理中...
                            ↓
[8, 7]                    []     lrem 删除 "6"

lrem(key, count, value)

参数	类型	必填	说明
`key`	`str`	是	列表名
`count`	`int`	是	删除数量和方向。`> 0` 从头到尾删 count 个；`< 0` 从尾到头删 count 个；`= 0` 删除所有匹配的
`value`	`str`	是	要删除的元素值（匹配列表中等于此值的元素）

返回值：int（实际删除的元素数量）

为什么 count=1：每个任务 ID 在 "处理中"队列只出现一次，删除 1 个即可。

为什么 count 的正负方向重要：

列表: [6, 7, 6, 8, 6]

lrem(key, 2, "6")   → 从头到尾删 2 个 "6"
                     删除位置 0 和 2
                     结果: [7, 8, 6]

lrem(key, -2, "6")  → 从尾到头删 2 个 "6"
                     删除位置 4 和 2
                     结果: [6, 7, 8]

lrem(key, 0, "6")   → 删除所有 "6"
                     结果: [7, 8]

lrange(key, start, stop)

参数	类型	必填	说明
`key`	`str`	是	列表名
`start`	`int`	是	起始索引（0 = 第一个元素，-1 = 最后一个元素）
`stop`	`int`	是	结束索引（包含该位置）

返回值：list[str]（不弹出，只读取，列表内容不变）

lrange("task:processing", 0, -1) 的含义：

列表: [6, 7, 8]
       ↑      ↑
     start=0  stop=-1

返回: ["6", "7", "8"]    ← 读取全部元素，列表不变

常用索引：

start	stop	含义
`0`	`-1`	全部元素
`0`	`0`	只取第一个
`-3`	`-1`	取最后 3 个
`0`	`2`	取前 3 个

lpush(key, *values)

参数	类型	必填	说明
`key`	`str`	是	列表名
`*values`	`str`	是	一个或多个要推入的值，按参数顺序逐个推入左端

返回值：int（推入后的列表长度）

推入顺序：

python

redis_client.lpush("task:queue", "6", "7", "8")
# 推入过程:
#   推入 "6" → [6]
#   推入 "7" → [7, 6]
#   推入 "8" → [8, 7, 6]
# 最终列表: [8, 7, 6]
# 返回: 3

注意：参数顺序和最终列表顺序是反的。先传的 "6" 在列表最右端，后传的 "8" 在最左端。

9.1.5 恢复流程图

Worker 启动
    │
    ▼
recover_stuck_tasks()
    │
    ├── lrange("task:processing", 0, -1)
    │     返回: ["6", "7"]     ← 上次崩溃残留的任务
    │
    ├── 对每个 task_id:
    │     ├── lpush("task:queue", task_id)    ← 推回主队列
    │     └── lrem("task:processing", 1, task_id)  ← 从处理中删除
    │
    ▼
process_queue()    ← 开始正常消费

9.2 队列积压监控

9.2.1 监控代码

python

# 检查队列长度
queue_len = redis_client.llen("task:queue")
if queue_len > 50:
    logger.warning("任务队列积压: %s 个待处理", queue_len)

9.2.2 涉及函数：`llen`

python

redis_client.llen(key)

参数	类型	必填	说明
`key`	`str`	是	列表名

返回值：int（列表长度，即元素个数）

时间复杂度：O(1)，直接读取列表头部的长度字段，不遍历。

本项目中的监控阈值：

队列长度	含义	建议动作
0-5	正常	无
5-20	轻微积压	日志记录
20-50	中等积压	告警通知
> 50	严重积压	考虑启动更多 Worker

9.2.3 多队列监控

python

# 同时监控 VIP 和普通队列
vip_len = redis_client.llen("task:queue:vip")
normal_len = redis_client.llen("task:queue:normal")

if vip_len > 10:
    logger.warning("VIP 队列积压: %s 个", vip_len)
if normal_len > 50:
    logger.warning("普通队列积压: %s 个", normal_len)

9.2.4 队列内容查看（不弹出）

python

# 查看队列中有哪些任务（不影响消费）
queued_tasks = redis_client.lrange("task:queue", 0, -1)
# 返回: ["6", "7", "8"]    ← 从右到左是弹出的先后顺序

# 只看前 10 个（最近推入的 10 个）
recent_tasks = redis_client.lrange("task:queue", 0, 9)
# 返回: ["8", "7"]    ← 列表只有 2 个元素时

lrange 不影响队列：和 brpop 不同，lrange 只读取不删除，不会干扰 Worker 消费。

9.3 优先级队列的饥饿问题

9.3.1 问题描述

VIP 队列持续有任务 → 普通队列永远不被消费 → 饥饿

时间 0s:  VIP 队列不断有任务推入
时间 1h:  普通队列的任务等了 1 小时还没被处理 ❌

9.3.2 解决方案代码

python

def process_queue() -> None:
    normal_count = 0
    while True:
        # 每处理 3 个 VIP 任务后，强制取 1 个普通任务
        if normal_count >= 3:
            result = redis_client.brpop(["task:queue:normal"], timeout=1)
            normal_count = 0
        else:
            result = redis_client.brpop(
                ["task:queue:vip", "task:queue:normal"], timeout=5
            )

        if result:
            queue_name, task_id = result
            if queue_name == "task:queue:vip":
                normal_count += 1
            run_task(int(task_id))

9.3.3 涉及函数参数详解

brpop(keys, timeout) — 两种调用方式

python

# 方式 1：传列表（多队列优先级）
result = redis_client.brpop(
    ["task:queue:vip", "task:queue:normal"],
    timeout=5
)

参数	值	说明
`keys`	`["task:queue:vip", "task:queue:normal"]`	先查 vip，vip 为空再查 normal
`timeout`	`5`	最多等 5 秒，超时返回 None

python

# 方式 2：传单字符串（只查一个队列）
result = redis_client.brpop(["task:queue:normal"], timeout=1)

参数	值	说明
`keys`	`["task:queue:normal"]`	只查 normal 队列，不查 vip
`timeout`	`1`	只等 1 秒，快速跳过，不长时间阻塞

为什么强制消费时 timeout=1 而不是 5：强制消费普通队列时，如果 normal 队列刚好也空了，不需要长时间等待，1 秒足够。如果没有普通任务，立即回到正常的多队列优先级循环。

9.3.4 返回值处理

python

if result:
    queue_name, task_id = result
    if queue_name == "task:queue:vip":
        normal_count += 1
    run_task(int(task_id))

变量	值示例	说明
`result`	`("task:queue:vip", "101")`	元组：队列名 + 任务 ID
`queue_name`	`"task:queue:vip"`	用于判断任务来源
`task_id`	`"101"`	Redis 存的是字符串，需要 `int()` 转换
`normal_count`	`0` → `1` → `2` → `3`	连续消费 VIP 的计数，达到 3 后强制消费普通

9.3.5 消费顺序图

VIP 持续到达时的消费顺序:

  VIP 101 → VIP 102 → VIP 103 → 普通 201 → VIP 104 → VIP 105 → VIP 106 → 普通 202 → ...
  ↑───────── 3个VIP ──────────↑  ↑强制普通↑  ↑───────── 3个VIP ──────────↑  ↑强制普通↑

  normal_count:  1    2    3    reset=0     1    2    3    reset=0

9.4 BRPOP 的性能特征

9.4.1 时间复杂度

操作	时间复杂度	说明
弹出	O(1)	列表右端弹出，直接操作尾部指针，常数时间
阻塞等待	O(1)	不占 CPU，由 Redis 事件通知唤醒
多 key 检查	O(N)	N = key 数量，通常 ≤ 2-3，可忽略
多 Worker 等待	O(1)	等待链表头部操作，不随 Worker 数增长

9.4.2 `brpop` 的 timeout 参数对性能的影响

timeout 值	CPU 开销	退出延迟	适用场景
`0`（永久阻塞）	最低	无法 `Ctrl+C`	不推荐
`1`	较高（每秒循环）	1 秒	需要快速退出的测试环境
`5`	低	5 秒	推荐，平衡性和响应性
`30`	最低	30 秒	长期稳定运行的生产环境

timeout=5 时的循环开销：每 5 秒一次 while True 循环，CPU 占用可忽略。即使 100 个 Worker，每秒总共 20 次 BRPOP 调用，对 Redis 毫无压力。

9.4.3 `brpop` 多 key 检查的开销

python

# 2 个 key
redis_client.brpop(["task:queue:vip", "task:queue:normal"], timeout=5)

Redis 内部执行：

1. 查找 "task:queue:vip"   → O(1) 哈希表查找
2. 检查列表是否非空          → O(1) 读取长度字段
3. 如果 vip 非空 → 弹出 → 返回
4. 如果 vip 为空 → 查找 "task:queue:normal" → O(1)
5. 检查 normal 是否非空      → O(1)
6. 如果 normal 非空 → 弹出 → 返回
7. 都空 → 加入两个等待链表  → O(1)

总开销：2-3 次 O(1) 操作，约 100 纳秒。即使 10 个 key 也只是 10 次 O(1)，可忽略。

9.4.4 单实例 Redis 队列吞吐量

场景	QPS	瓶颈
单 Worker BRPOP + LPUSH	10 万+	Redis 不是瓶颈，Worker 处理速度是瓶颈
10 Worker BRPOP + LPUSH	10 万+	Redis 仍不是瓶颈，网络带宽可能是
100 Worker BRPOP + LPUSH	10 万+	Redis 单线程开始成为瓶颈，需考虑集群
仅 LPUSH（API 推入）	10 万+	推入比弹出更快，无阻塞开销

结论：瓶颈永远在 Worker 处理速度（Seedance API 调用 4 分钟/任务），而非 Redis 队列操作（100 纳秒/操作）。

9.5 连接池配置

9.5.1 BRPOP 占用连接的问题

python

# BRPOP 阻塞期间，这个连接被独占
result = redis_client.brpop("task:queue", timeout=5)
# ← 5 秒内不能用同一个连接执行其他命令

原因：redis-py 的 Redis 对象内部维护一个 ConnectionPool。brpop 调用时从池中借一个连接，阻塞期间该连接被独占。如果池中只有 1 个连接，其他 Redis 命令（如 set、get、lpush）必须等待 BRPOP 返回后才能执行。

9.5.2 涉及配置：`ConnectionPool`

python

import redis

redis_client = redis.Redis(
    host="localhost",
    port=6379,
    decode_responses=True,
    max_connections=10,
)

参数	类型	默认值	说明
`host`	`str`	`"localhost"`	Redis 服务器地址
`port`	`int`	`6379`	Redis 端口
`decode_responses`	`bool`	`False`	`True` 时返回 `str` 而非 `bytes`；`False` 时返回 `bytes`
`max_connections`	`int`	`2**31`（实际受系统限制）	连接池最大连接数。超过时抛 `ConnectionError`

decode_responses=True 的影响：

python

# decode_responses=False（默认）
result = redis_client.brpop("task:queue", timeout=5)
# result = (b"task:queue", b"6")    ← bytes
# 需要 result[1].decode("utf-8") 才能用

# decode_responses=True
result = redis_client.brpop("task:queue", timeout=5)
# result = ("task:queue", "6")     ← str
# 直接可用

9.5.3 连接数计算

Worker 进程内部的连接使用:

  连接 1: brpop("task:queue", timeout=5)     ← 阻塞中，独占
  连接 2: set("task:6:progress", "50%")       ← 写进度
  连接 3: lpush("review:queue", task_id)      ← 推入审核队列

  max_connections=10 足够

多 Worker 进程的连接数：

Worker 数	每个 Worker 连接数	总连接数	Redis max_clients
1	3-5	5	默认 10000，足够
10	3-5	50	足够
50	3-5	250	足够
200	3-5	1000	接近上限，需调大

9.5.4 连接池耗尽的错误

python

# 如果 max_connections=1 且连接被 BRPOP 占用
redis_client.brpop("task:queue", timeout=60)   # 连接被占 60 秒
redis_client.set("key", "value")
# → redis.exceptions.ConnectionError: Too many connections

解决方案：max_connections 至少设为 Worker 内部并发使用的连接数 + 2（1 个给 BRPOP，其余给进度写入等）。

9.6 Worker 优雅退出

9.6.1 timeout 与信号处理的关系

python

def process_queue() -> None:
    while True:
        try:
            result = redis_client.brpop("task:queue", timeout=5)
            if result:
                run_task(int(result[1]))
        except KeyboardInterrupt:
            logger.info("Worker 收到退出信号，停止")
            break

timeout=5 在退出场景的作用：

时间 0s:  Worker 执行 brpop(timeout=5)，开始阻塞
时间 2s:  用户按 Ctrl+C → 发送 SIGINT 信号
时间 2s:  Python 收到 SIGINT → 抛出 KeyboardInterrupt
          但 brpop 还在阻塞中，异常被 redis-py 内部捕获
时间 5s:  brpop 超时返回 None → while 循环回到顶部
时间 5s:  KeyboardInterrupt 在 try-except 中被捕获 → break

如果 timeout=0（永久阻塞）：

时间 0s:  Worker 执行 brpop(timeout=0)，永久阻塞
时间 2s:  用户按 Ctrl+C → SIGINT
时间 2s:  KeyboardInterrupt 被抛出
          redis-py 尝试中断阻塞命令
          但 Redis 服务端不知道客户端要退出
          连接可能卡住 → 需要 kill -9

9.6.2 推荐的退出方案

python

import signal

shutdown = False

def handle_signal(signum, frame):
    global shutdown
    shutdown = True

signal.signal(signal.SIGINT, handle_signal)
signal.signal(signal.SIGTERM, handle_signal)

def process_queue() -> None:
    while not shutdown:
        try:
            result = redis_client.brpop("task:queue", timeout=5)
            if result:
                run_task(int(result[1]))
        except Exception as e:
            logger.exception("Worker 异常: %s", e)
            time.sleep(1)

    logger.info("Worker 优雅退出")

signal.signal(signum, handler) 参数：

参数	类型	说明
`signum`	`int`	信号编号。`signal.SIGINT` = 2（Ctrl+C），`signal.SIGTERM` = 15（kill 命令默认）
`handler`	`callable`	信号处理函数，接收 `(signum, frame)` 两个参数

handler 的参数：

参数	说明
`signum`	触发的信号编号（2 或 15）
`frame`	当前的 Python 栈帧对象（通常不用）

优雅退出流程：

Ctrl+C / kill
    │
    ▼
handle_signal() → shutdown = True
    │
    ▼
brpop 最多等 5 秒超时返回 None
    │
    ▼
while not shutdown → 条件为 False → 退出循环
    │
    ▼
Worker 优雅退出（不会在任务处理中途被杀）

九、实际运行中的注意事项 ​

9.1 任务丢失风险 ​

9.1.1 问题场景 ​

9.1.2 涉及函数：brpop ​

9.1.3 解决方案：RPOPLPUSH 模式 ​

9.1.4 涉及函数参数详解 ​

9.1.5 恢复流程图 ​

9.2 队列积压监控 ​

9.2.1 监控代码 ​

9.2.2 涉及函数：llen ​

9.2.3 多队列监控 ​

9.2.4 队列内容查看（不弹出） ​

9.3 优先级队列的饥饿问题 ​

9.3.1 问题描述 ​

9.3.2 解决方案代码 ​

9.3.3 涉及函数参数详解 ​

9.3.4 返回值处理 ​

9.3.5 消费顺序图 ​

9.4 BRPOP 的性能特征 ​

9.4.1 时间复杂度 ​

9.4.2 brpop 的 timeout 参数对性能的影响 ​

9.4.3 brpop 多 key 检查的开销 ​

9.4.4 单实例 Redis 队列吞吐量 ​

9.5 连接池配置 ​

9.5.1 BRPOP 占用连接的问题 ​

9.5.2 涉及配置：ConnectionPool ​

9.5.3 连接数计算 ​

9.5.4 连接池耗尽的错误 ​

9.6 Worker 优雅退出 ​

9.6.1 timeout 与信号处理的关系 ​

9.6.2 推荐的退出方案 ​

九、实际运行中的注意事项

9.1 任务丢失风险

9.1.1 问题场景

9.1.2 涉及函数：`brpop`

9.1.3 解决方案：RPOPLPUSH 模式

9.1.4 涉及函数参数详解

9.1.5 恢复流程图

9.2 队列积压监控

9.2.1 监控代码

9.2.2 涉及函数：`llen`

9.2.3 多队列监控

9.2.4 队列内容查看（不弹出）

9.3 优先级队列的饥饿问题

9.3.1 问题描述

9.3.2 解决方案代码

9.3.3 涉及函数参数详解

9.3.4 返回值处理

9.3.5 消费顺序图

9.4 BRPOP 的性能特征

9.4.1 时间复杂度

9.4.2 `brpop` 的 timeout 参数对性能的影响

9.4.3 `brpop` 多 key 检查的开销

9.4.4 单实例 Redis 队列吞吐量

9.5 连接池配置

9.5.1 BRPOP 占用连接的问题

9.5.2 涉及配置：`ConnectionPool`

9.5.3 连接数计算

9.5.4 连接池耗尽的错误

9.6 Worker 优雅退出

9.6.1 timeout 与信号处理的关系

9.6.2 推荐的退出方案