Redis 技术问答：队列、连接与并发专题总集

整合自 Redis-BRPOP深度解析、Redis单线程与多客户端交互、Redis原子性保证机制、Redis连接池详解、Redis推拉模型与源码解析、Worker编排器深度解析日期：2026-08-26

合并来源

本文把 redis/ 目录下围绕队列、连接、原子性、推拉模型和 Worker 编排的专题文档，按“基础语法 → 服务端执行模型 → 连接管理 → 原子性 → 源码与编排 → 数据结构与应用”的顺序重新编排到一篇主文档中，便于统一阅读。

[Redis-BRPOP深度解析.md](file:///Users/binny/workspace-sdd/aiNotes/redis/Redis-BRPOP深度解析.md)
Redis单线程与多客户端交互.md
Redis原子性保证机制.md
Redis连接池详解.md
Redis推拉模型与源码解析.md
Worker编排器深度解析.md

术语说明：文档里的 Worker 默认代表独立执行单元

本文中提到的“多个 Worker”，默认隐含的意思是：多个彼此独立的执行单元。这个执行单元可以是多个进程，也可以是同一个进程里的多个线程。

在 Redis 看来，它并不直接关心 Worker 是线程还是进程。Redis 服务端真正看到的是客户端发起的 TCP 连接，也就是 Redis 内部的 client 结构体。因此：

文档说法	应用侧含义	Redis 服务端看到的对象
3 个 Worker 进程	3 个独立 Python 进程，各自有自己的 `redis_client` 和连接池	通常至少 3 条 TCP 连接，`CLIENT LIST` 至少 3 行
1 个进程里的 3 个 Worker 线程	3 个线程共享同一个 `redis_client`，但并发命令会从连接池借出不同连接	并发时可能看到 3 条 TCP 连接，等待链表也可能有 3 个节点
3 个 Worker 同时 BRPOP	3 个独立执行流同时阻塞等待队列	Redis 等待链表里是 3 个正在阻塞的 TCP 连接

所以后文中如果写“Worker A / Worker B / Worker C”，除非特别说明，重点不是它们到底是线程还是进程，而是它们代表三个独立发起 Redis 命令的执行流。

第一章：BRPOP 基础语法

本章导读

本章围绕“BRPOP 基础语法”展开，先交代适用场景，再说明核心原理，随后给出典型示例，最后补充小结与延伸阅读。

阅读顺序： 结论 → 原理 → 场景 → 代码 → 小结

Q1：BRPOP 和 RPOP 的核心区别是什么？为什么本项目用 BRPOP？

答：

核心区别在于列表为空时的行为：

命令	列表为空时	CPU 占用	延迟
`RPOP`	立即返回 `nil`	轮询模式持续消耗 CPU	取决于 sleep 间隔
`BRPOP`	阻塞等待，直到有元素或超时	等待期间零 CPU 占用	< 1 毫秒

本项目使用 BRPOP 的主要原因是：Worker 没有任务时不需要反复轮询 Redis，而是挂起等待；一旦有任务进入队列，Redis 会立即唤醒等待中的 Worker，让它继续执行后续任务处理逻辑。

需要注意的是，BRPOP 的“阻塞”要区分客户端侧和 Redis 服务端侧：

对当前调用 BRPOP 的 Worker 线程来说，会阻塞在 brpop() 这一行，后面的代码暂时不会继续执行；
对 Redis 服务端来说，不会阻塞 Redis 主线程，Redis 会把这个客户端连接挂到等待链表中，然后继续处理其他客户端命令；
对其他 Worker 或其他 Redis 客户端来说，不受这个阻塞影响，仍然可以继续执行 HSET、LPUSH、GET 等命令。

也就是说，BRPOP 阻塞的是当前客户端线程的后续流程，不是阻塞整个 Redis 服务。

RPOP 轮询方式

python

while True:
    result = redis_client.rpop("task:queue")
    if result:
        process(result)
    else:
        time.sleep(0.1)

这种写法的问题是：队列为空时，Worker 会不断主动查询 Redis。即使每次查询后都 sleep(0.1)，本质上仍然是轮询模式，会产生无效请求，并且任务到达后最多可能等待 0.1 秒才被下一轮轮询发现。

BRPOP 阻塞等待方式

python

def worker_loop():
    while True:
        result = redis_client.brpop("task:queue", timeout=5)

        print("只有 brpop 返回后，才会执行到这里")

        if result is None:
            continue

        queue_name, task_id = result
        process_task(task_id)

上面这段代码中，如果 task:queue 队列为空，程序会停在：

python

result = redis_client.brpop("task:queue", timeout=5)

只有出现以下两种情况之一，代码才会继续往下执行：

队列中有新任务进入，Redis 唤醒当前客户端并返回任务数据；
等待时间超过 timeout=5，Redis 返回 None。

所以，在同一个 Worker 线程里，brpop() 后面的 print()、process_task() 等代码，必须等 brpop() 返回后才会执行。

BRPOP 不会阻塞 Redis 服务端主线程

假设有三个独立执行流：Worker A、Worker B、Worker C。它们可以是三个不同进程，也可以是同一进程里的三个不同线程；关键是它们不是同一个顺序执行流，否则 Worker A 阻塞在 brpop() 时，后面的 hset 和 lpush 不会继续执行。

python

# Worker A：独立线程或独立进程
result = redis_client.brpop("task:queue", timeout=5)

# Worker B：另一个独立线程或独立进程
redis_client.hset("task:6:progress", "stage", "IMAGE")

# Worker C：另一个独立线程或独立进程
redis_client.lpush("task:queue", 6)

当 Worker A 执行 BRPOP 且队列为空时：

对象	是否被阻塞	说明
Worker A 当前线程	是	卡在 `brpop()`，后续代码暂时不执行
Worker B	否	仍然可以正常执行 `hset`
Worker C	否	仍然可以正常执行 `lpush`
Redis 服务端主线程	否	继续处理其他客户端命令
`task:queue` 等待链表	有变化	Worker A 对应的客户端连接被挂入等待链表

当 Worker C 执行：

python

redis_client.lpush("task:queue", 6)

Redis 会在执行 LPUSH 后检查是否有客户端正在等待 task:queue。如果发现 Worker A 正在 BRPOP 等待，就会把任务 6 返回给 Worker A。

此时 Worker A 的代码才会继续执行：

python

queue_name, task_id = result
process_task(task_id)

本章小结

BRPOP 的阻塞是客户端侧的阻塞。它会让当前调用 BRPOP 的线程暂停在这一行，直到有任务或超时返回；但它不会阻塞 Redis 服务端主线程，也不会影响其他客户端继续执行命令。

在任务队列场景中，这正是 BRPOP 的优势：

方式	没有任务时	CPU 消耗	后续代码是否执行	适用场景
`RPOP` 轮询	不断主动查询	高	每次查询后继续循环	简单但低效
`BRPOP` 阻塞等待	客户端挂起等待	低，接近 0	当前线程暂停，返回后继续	Worker 消费队列

所以，“没有任务时挂起，零 CPU；有任务时 < 1ms 唤醒”的准确含义是：

没有任务时，Worker 当前线程会挂起等待，不会反复空转轮询，因此 CPU 消耗接近 0；一旦有任务入队，Redis 会通过已建立的 TCP 连接把结果返回给等待中的客户端，Worker 被唤醒后继续执行后面的任务处理流程。

Q2：`brpop(["task:queue:vip", "task:queue:normal"], timeout=5)` 的优先级规则是什么？如果 VIP 队列持续有任务，普通队列会怎样？

答：

优先级规则：按 key 的声明顺序从左到右检查。只要左边的队列有元素，就不会弹右边的。

BRPOP [vip, normal] 5

VIP: [101, 102]     Normal: [201, 202]
第1次 BRPOP → 取 102（VIP）
第2次 BRPOP → 取 101（VIP）
第3次 BRPOP → 取 201（Normal）   ← VIP 空了才取 Normal
第4次 BRPOP → 取 202（Normal）

VIP 饥饿问题：如果 VIP 队列持续有任务推入，普通队列的任务可能永远不被消费。

解决方案：计数器策略，每处理 3 个 VIP 任务后强制消费 1 个普通任务：

python

normal_count = 0
while True:
    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))

Q3：`redis_client.lpush("task:queue", 101)` 的返回值是什么？为什么不是 101？

答：

返回值是 推入后的列表长度（int 类型），不是你推入的值。

python

result = redis_client.lpush("task:queue", 101)
# 列表内容: [101]     ← 101 被存入列表
# result = 1           ← 返回的是列表长度

result = redis_client.lpush("task:queue", 102)
# 列表内容: [102, 101]
# result = 2           ← 列表现在有 2 个元素

为什么这么设计：你已经知道推入了什么（就是你传的参数），Redis 选择返回更有用的信息——列表当前有多少个元素。

对比其他命令的返回值：

命令	返回值含义	示例
`lpush`	推入后的列表长度	`1`
`brpop`	弹出的元素值 + 队列名	`("task:queue", "101")`
`rpop`	弹出的元素值	`"101"`
`llen`	列表长度	`1`

Q4：为什么 Worker 取出 task_id 后需要 `int(task_id)` 转换？

答：

因为 Redis 内部统一以字符串存储所有值，不管你推入时传的是 int 还是 str。

python

# API 端存入时是 int
redis_client.lpush("task:queue", task.id)  # task.id = 6 (int)

# Redis 内部存储为字符串 "6"

# Worker 端取出时是字符串
_, task_id = redis_client.brpop("task:queue", timeout=5)
# task_id = "6" (str)

# SQLAlchemy 查询需要 int 主键
task = db.get(Task, int(task_id))  # 必须转回 int

如果不转换，直接用字符串 "6" 查数据库主键，SQLAlchemy 会报类型错误。

decode_responses 参数不影响这个行为——不管返回 b"6" 还是 "6"，int() 都能正确转换：

python

int(b"6")   # → 6（bytes 自动解码）
int("6")    # → 6

第二章：阻塞机制与连接管理

本章导读

本章围绕“阻塞机制与连接管理”展开，先交代适用场景，再说明核心原理，随后给出典型示例，最后补充小结与延伸阅读。

阅读顺序： 结论 → 原理 → 场景 → 代码 → 小结

Q5：BRPOP 阻塞期间，Redis 服务端的主线程会卡住吗？其他客户端的命令能执行吗？

答：

不会卡住。BRPOP 的"阻塞"是客户端层面的，服务端只是把客户端加入等待链表后立即去处理下一条命令。

0ms:   主线程执行 A 的 BRPOP → 队列空 → A 挂起（加入等待链表）→ 立即返回
0.1ms: 主线程执行 B 的 BRPOP → 队列空 → B 挂起 → 立即返回
0.2ms: 主线程执行 C 的 BRPOP → 队列空 → C 挂起 → 立即返回
0.3ms: 主线程空闲，可以处理其他客户端命令
       ↑ 3 个 Worker 都在阻塞，但主线程没被阻塞
3000ms: 进程 D 发 LPUSH → 主线程执行 → 队列非空 → 唤醒 A

对比：

命令	客户端行为	服务端行为
`BRPOP`（队列空）	客户端挂起等待	立即把客户端加入等待链表，去处理下一条命令
`BRPOP`（队列非空）	立即收到返回值	立即弹出元素返回，去处理下一条命令
`LPUSH`	立即收到返回值	立即推入并检查等待链表，去处理下一条命令

没有任何命令会让 Redis 主线程阻塞等待。

Q6：一个 Worker 进程中，BRPOP 阻塞期间还能执行 `hset` 写进度吗？如果能，用的是同一个 TCP 连接吗？

答：

分两种情况：

单线程：❌ 不可能。brpop() 的 read_response() 会阻塞整个 Python 线程，hset 根本执行不到。
多线程：✅ 可以。brpop 在线程 A 阻塞，hset 在线程 B 执行，用的是不同的 TCP 连接。

1. 单线程下不可能并行（常见误区）

python

# ❌ 单线程下不可能在 brpop 阻塞期间执行 hset
def worker_single_thread():
    while True:
        result = redis_client.brpop("task:queue", timeout=5)
        # ↑ 阻塞在 read_response()，独占 Conn-1
        # ↑ python 主线程被卡住，下面这行根本执行不到！
        redis_client.hset("task:6:progress", "stage", "IMAGE")

为什么执行不到：brpop() 内部调用 conn.read_response()，这是一个阻塞 I/O 调用——整个 Python 线程会被挂起，直到 socket 有数据可读或超时。期间无法执行任何其他代码。

2. 本项目实际是顺序执行（不是并行）

python

# ✅ 本项目实际代码：brpop 返回后才执行 hset
def process_queue():
    while True:
        result = redis_client.brpop("task:queue", timeout=5)
        # ↑ brpop 阻塞 → 拿到任务或超时后才返回
        # ↑ 返回时 Conn-1 已经归还到池中
        if result:
            run_task(int(result[1]))  # ← hset 在这里执行
            # ↑ 此时 brpop 已结束，Conn-1 已归还
            # ↑ hset 借出 Conn-1（复用），立即完成

def run_task(task_id):
    redis_client.hset(f"task:{task_id}:progress", "stage", "SCRIPT")
    # ↑ 借 Conn-1（brpop 已归还的）→ 执行 → 归还

时间线（本项目单线程顺序执行）:

brpop 阻塞中（5 秒）          brpop 返回
  ├─ 借出 Conn-1                 ├─ 归还 Conn-1
  ├─ read_response() 阻塞         │
  │  _in_use: {Conn-1}            │  _available: [Conn-1]
  │  _available: []                │
  │  线程被挂起                    │
  │  无法执行 hset                 ├─ hset 借出 Conn-1（复用）
  │                                ├─ 立即执行完成
  │                                └─ 归还 Conn-1
  └─ Redis 回复或超时

关键：本项目 brpop 和 hset 是顺序执行，不是并行。每次只有一个命令在执行，一个连接就够了。

3. 真正的并行需要多线程

python

import threading

# ✅ 多线程：brpop 阻塞期间，另一个线程可以执行 hset
def consumer_thread():
    """线程 A：消费队列"""
    while True:
        result = redis_client.brpop("task:queue", timeout=5)
        # ↑ 借出 Conn-1，阻塞在 read_response()
        # ↑ 但只阻塞线程 A，不影响线程 B

def progress_writer_thread():
    """线程 B：写进度"""
    while True:
        time.sleep(1)
        redis_client.hset("task:6:progress", "stage", "PROCESSING")
        # ↑ 借出 Conn-2（因为 Conn-1 被线程 A 独占）
        # ↑ 立即执行完成
        # ↑ 不受 brpop 阻塞影响

threading.Thread(target=consumer_thread).start()
threading.Thread(target=progress_writer_thread).start()

多线程时间线:

线程 A (consumer)               线程 B (progress_writer)
─────────────────              ─────────────────────────
brpop("task:queue", 5)
  借出 Conn-1
  send_command("BRPOP")
  read_response() 阻塞...
  _in_use: {Conn-1}
  _available: [Conn-2]
                                hset("task:6:progress", ...)
                                  get_connection()
                                  ├── _available.pop() → Conn-2  ← 借出 Conn-2
                                  ├── send HSET
                                  ├── read_response() → 立即返回
                                  └── release(Conn-2)
                                  ✓ 不需要等 brpop！

  Redis 回复...
  brpop() 返回
  release(Conn-1)

这才是 max_connections >= 2 真正发挥价值的场景：线程 B 从池中借出 Conn-2，完全不受 Conn-1 被 BRPOP 独占的影响。

4. 连接池状态对比

场景	brpop 期间连接池状态	hset 能否执行	用哪个连接
单线程（本项目）	`_in_use: {Conn-1}` 但线程被阻塞	❌ 执行不到	-
单线程 brpop 返回后	`_available: [Conn-1]`	✅ 顺序执行	Conn-1（复用）
多线程	`_in_use: {Conn-1}` + `_available: [Conn-2]`	✅ 并行执行	Conn-2（新借出）

5. 如果 `max_connections=1` 会怎样

场景	max_connections=1	结果
单线程	brpop 独占 Conn-1	无影响（hset 本来就执行不到）
单线程 brpop 返回后	Conn-1 已归还	hset 复用 Conn-1，正常
多线程	brpop 独占 Conn-1，线程 B 借不到	`ConnectionError: Too many connections`

所以：单线程下 max_connections=1 也能跑；多线程才需要 max_connections >= 2。本项目用 max_connections=10 是为 Seedance 分镜并行（3 线程）留余量。

Q7：`timeout=0`（永久阻塞）和 `timeout=5` 有什么实际区别？为什么不推荐 `timeout=0`？

答：

timeout	行为	能否响应 Ctrl+C	风险
`0`	永久阻塞，直到有元素	❌ 不能	Redis 重启后连接可能假死
`5`	最多等 5 秒，超时回到循环	✅ 每 5 秒检查一次	无

python

# ❌ timeout=0：Ctrl+C 无法中断
while True:
    result = redis_client.brpop("task:queue", timeout=0)
    # 如果队列一直没元素，永远卡在这里
    # Ctrl+C 信号无法被 python 处理

# ✅ timeout=5：每 5 秒回到循环顶部
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 时，每 5 秒 BRPOP 超时返回 None，while True 循环回到顶部，Python 有机会检查 KeyboardInterrupt 信号。5 秒是经验值——太短浪费 CPU（频繁循环），太长退出延迟。

第三章：单线程模型与原子性

本章导读

本章围绕“单线程模型与原子性”展开，先交代适用场景，再说明核心原理，随后给出典型示例，最后补充小结与延伸阅读。

阅读顺序： 结论 → 原理 → 场景 → 代码 → 小结

Q8：Redis 是单线程的，为什么还能支撑 10 万 QPS？单线程不会成为瓶颈吗？

答：

单线程不会成为瓶颈，原因有四个：

因素	说明
纯内存操作	内存读写纳秒级，不需要磁盘 I/O
无锁竞争	单线程不需要加锁、解锁、等待锁
无上下文切换	线程切换成本约 5-10 微秒，Redis 完全避免
I/O 多路复用	epoll 让单线程管理万级连接，不阻塞

单命令耗时通常在 100 纳秒以内：

BRPOP 弹出: ~100ns
LPUSH 推入: ~100ns
LLEN 长度: ~50ns
唤醒等待客户端: ~200ns

即使 10 万 QPS，CPU 利用率也不到 50%。

在本项目中：瓶颈永远在 Worker 处理速度（Seedance API 调用 4 分钟/任务），而非 Redis 队列操作（100 纳秒/操作），差距 7 个数量级。

Q9：10 个 Worker 同时 `BRPOP task:queue 5`，这时有人 `LPUSH task:queue 6`，为什么不会有两个 Worker 同时拿到 "6"？

答：

Redis 通过四层保证确保只有一个 Worker 能拿到元素：

第一层：单线程命令执行

Redis 命令执行引擎只有一个线程。不管多少个客户端同时发命令，它们在主线程中排队执行，不存在两条命令"同时执行"的窗口。

第二层：命令原子性

每条 Redis 命令是原子操作。LPUSH 的"插入元素 → 检查等待链表 → 唤醒客户端"是一个不可被打断的原子步骤。

第三层：等待链表 FIFO

多个 Worker 等待时，Redis 会按 BRPOP 真正到达服务端的先后顺序，把对应客户端连接挂到 task:queue 的等待链表中。这里的多个 Worker 隐含的是多个独立线程或多个独立进程，它们各自通过一条 TCP 连接向 Redis 发起 BRPOP。

假设 10 个 Worker 依次进入阻塞等待，等待链表不是一个无序集合，而是一条有队头和队尾的 FIFO 队列。下面这张图可以更直观地看出等待链和唤醒顺序：

注：原「Redis List 阻塞等待链表 FIFO 结构」手绘图缺失，见下方「图解说明」表格。

图解说明：

图中元素	含义
list 头部（左）	队头方向，`LPUSH` 从这里推入元素，`LPOP` 从这里弹出；也是最早等待的 Worker 所在位置
list 尾部（右）	队尾方向，`RPUSH` 从这里推入元素，`RPOP` 从这里弹出；最后到达的 Worker 排在这一端
W1（黄色框）	最早进入 `BRPOP` 等待的 Worker，位于队头，`LPUSH` 到达时第一个被唤醒
W2 / W3（米色框）	随后依次进入等待的 Worker，排在 W1 后面，只有 W1 出队后才轮到 W2
W4（绿色框）	最后进入等待的 Worker，位于队尾，需要等前面所有 Worker 都被唤醒后才会轮到它
实线箭头 →	等待链的方向，从队头到队尾依次排列
虚线弧形箭头	一次完整的唤醒流程：`LPUSH` 到达 → 只唤醒队头 W1 → W1 出队 → 队头变为 W2
LPUSH / LPOP	左侧操作，对应队头推入和弹出
RPUSH / RPOP	右侧操作，对应队尾推入和弹出

简单来说，这张图要表达三件事：

等待链是有序的：W1 最先到，排在队头；W4 最后到，排在队尾。
唤醒只取队头：LPUSH 推入元素后，Redis 只会唤醒队头的 W1，不会同时通知 W2、W3、W4。
出队后队列前移：W1 拿到元素离开等待链后，W2 自动成为新的队头，等待下一次 LPUSH 唤醒。

当有人执行 LPUSH task:queue 6 时，Redis 只会从队头取出最早等待的 W1，并把元素 6 交给 W1。

因此，6 只会交给一个等待者；后面的 W2、W3 不会同时拿到同一个元素，它们必须等待后续新的元素进入队列。

LPUSH 唤醒时只处理队头第一个等待客户端（break），不批量唤醒：

// Redis 源码 handleClientsBlockedOnKeys（简化）
while (clientNode) {
    client *receiver = clientNode->value;
    robj *value = listTypePop(o, LIST_TAIL);
    if (value) {
        addReply(receiver, value);      // 发给这个客户端
        listDelNode(clients, clientNode); // 从等待链表移除
        break;                            // ← 只唤醒一个！
    }
    clientNode = listNextNode(clientNode);
}

第四层：事件循环串行

即使 LPUSH 6 7 8 一次推入 3 个元素，也是一个一个唤醒：

第1轮: 弹出 "6" → 发给 W1 → break
第2轮: 弹出 "7" → 发给 W2 → break
第3轮: 弹出 "8" → 发给 W3 → break

Q10：LPUSH 唤醒 BRPOP 时，数据是先存入列表再弹出，还是直接传递？

答：

直接传递，数据不经过列表中转。

LPUSH task:queue 6 到达 Redis:

Redis 主线程执行 LPUSH:
  1. 在列表头部插入 "6" → 列表变为 [6]
  2. 检查等待链表 → 发现 W1 在等
  3. 直接把 "6" 发给 W1（通过 fd 写回）
  4. 从列表删除 "6" → 列表变为 []
  5. 从等待链表移除 W1
  6. LPUSH 命令完成

整个过程耗时: ~200 纳秒

步骤 2-5 全在 LPUSH 的一条命令执行中完成，是原子的。数据从 LPUSH 方直接传递给 BRPOP 方，O(1) 时间完成。

注意：如果等待链表为空（没有 Worker 在等），"6" 会留在列表中，等下一个 BRPOP 来弹出。

Q10.1：BRPOP 的"推送"本质是什么？是延迟响应还是真正的推送？

答：

BRPOP 的唤醒机制本质是延迟响应（Delayed Response），不是 TCP 层面的全双工推送，也不是客户端主动拉取。

1. 三种模型的严格区分

模型	是否符合 BRPOP	说明
纯拉（客户端轮询）	❌ 否	客户端没有反复发请求查询队列状态
纯推（服务端主动连客户端）	❌ 否	客户端必须先发 BRPOP 请求，服务端才会响应
延迟响应	✅ 是	客户端发一次请求，服务端延迟到条件满足才响应

2. 延迟响应的完整流程

普通命令（HSET）—— 立即响应:
  Client ──HSET──► Redis
  Client ◄──OK──── Redis    ← 立即响应（~0.1ms）

BRPOP（延迟响应）:
  Client ──BRPOP─► Redis
                  Redis: 队列空 → 加入等待链表 → 不回复（命令挂起）
                  Redis: 主线程继续处理其他客户端命令
                  （等待中... 可能几秒）
                  Redis: 收到 LPUSH → 弹出元素 → addReply(Client, 元素)
  Client ◄──元素── Redis    ← 延迟响应（可能几秒后才回复）

3. 延迟响应的三个关键特征

特征	说明
命令发送	客户端主动发送 BRPOP（一次请求）
响应时机	由服务端决定（队列有元素或超时时才响应）
等待期间	客户端 socket 阻塞在 `read_response()`，服务端主线程不阻塞

4. "推送"说法的精确含义

LPUSH 唤醒 BRPOP 时，Redis 服务端通过 addReply() 把结果写入等待客户端的输出缓冲区——这一步看起来像"推送"，但本质是延迟回复之前挂起的 BRPOP 请求：

// Redis 源码 t_list.c（简化）

void handleClientsBlockedOnKeys(void) {
    // LPUSH 执行后，主线程事件循环处理 ready_keys
    while (listLength(server.ready_keys) > 0) {
        // 查找等待这个 key 的客户端链表
        list *clients = dictGetVal(de);

        // 遍历等待的客户端
        client *receiver = listNode->value;

        // 弹出列表元素
        robj *value = listTypePop(o, LIST_TAIL);

        if (value) {
            // ★ 关键：把结果写入等待客户端的输出缓冲区
            addReplyArrayLen(receiver, 2);
            addReplyBulk(receiver, key);     // 队列名
            addReplyBulk(receiver, value);    // 弹出的元素
            // ↑ 这就是"延迟响应"——回复之前挂起的 BRPOP 请求
        }
    }
}

5. 与全双工推送的对比

对比项	BRPOP 延迟响应	TCP 全双工推送
谁先发起	客户端先发 BRPOP	服务端可主动发
响应触发	服务端条件满足时回复	服务端任意时刻推送
TCP 模型	单工请求-响应（延迟版本）	双工独立通道
一次请求响应次数	一次请求对应一次响应	一次请求可触发多次推送

6. 为什么说是"服务端主动推送"

虽然本质是延迟响应，但从客户端视角看：

客户端发完 BRPOP 后就阻塞在 read_response()
没有再发任何请求
数据"自动"到达了 socket

所以从客户端视角称为"推送"也没错——但严格来说这是延迟响应，因为客户端必须先发 BRPOP 请求，服务端才能"推"。

7. 总结

说法	是否准确
"BRPOP 是客户端轮询拉取"	❌ 错误（客户端没有反复发请求）
"BRPOP 是服务端主动推送"	⚠️ 不严谨（本质是延迟响应，但客户端视角像推送）
"BRPOP 是延迟响应"	✅ 准确（一次请求，服务端延迟回复）
"LPUSH 触发 BRPOP 的回复"	✅ 准确（LPUSH 执行时调用 addReply 写入客户端缓冲区）
"BRPOP 期间 TCP 是全双工"	❌ 错误（TCP 是全双工，但 BRPOP 是单工请求-响应）

Q11：假设 Worker-1 和 Worker-2 几乎同时发送 BRPOP，API 也几乎同时发送 LPUSH，三种命令到达 Redis 的顺序不同，结果会怎样？

答：

不管命令到达顺序如何，只有一个 Worker 能拿到元素。推演三种情况：

情况 A：BRPOP(W1) → BRPOP(W2) → LPUSH

BRPOP(W1) 执行：列表空 → W1 加入等待链表
BRPOP(W2) 执行：列表空 → W2 加入等待链表（排在 W1 后面）
LPUSH 执行：
  1. 插入 "6"
  2. 检查等待链表 → W1 是第一个
  3. "6" 发给 W1
  4. 删除 "6"
  5. 移除 W1

结果：W1 拿到 "6"，W2 继续等待 ✅

情况 B：LPUSH → BRPOP(W1) → BRPOP(W2)

LPUSH 执行：
  1. 插入 "6"
  2. 等待链表为空，没人等待
  3. "6" 留在列表中

BRPOP(W1) 执行：列表非空 → 直接弹出 "6"，返回给 W1
BRPOP(W2) 执行：列表空 → W2 加入等待链表

结果：W1 拿到 "6"，W2 继续等待 ✅

情况 C：BRPOP(W1) → LPUSH → BRPOP(W2)

BRPOP(W1) 执行：列表空 → W1 加入等待链表
LPUSH 执行：
  1. 插入 "6"
  2. 检查等待链表 → W1 在等
  3. "6" 发给 W1
  4. 删除 "6"
  5. 移除 W1

BRPOP(W2) 执行：列表空 → W2 加入等待链表

结果：W1 拿到 "6"，W2 继续等待 ✅

根本原因：Redis 单线程串行执行，LPUSH 和 BRPOP 不可能"同时"执行，两条命令之间不存在"间隙"让数据不一致。

第四章：多 Worker 与连接池

本章导读

本章围绕“多 Worker 与连接池”展开，先交代适用场景，再说明核心原理，随后给出典型示例，最后补充小结与延伸阅读。

阅读顺序： 结论 → 原理 → 场景 → 代码 → 小结

Q12：3 个 Worker 进程启动后，各自有独立的 `redis_client` 吗？Redis 服务端看到的是 3 个连接还是 1 个？

答：

3 个独立的 redis_client，Redis 端看到 3 个 TCP 连接。

这里的“3 个 Worker 进程”指的是用 3 个独立 OS 进程启动 3 个 Worker。每个 python -m app.worker 是独立进程，各自 import 模块时创建 redis_client 单例。每个进程启动时连接池都是空的（按需建连），brpop 第一次执行时才建立第 1 条 TCP：

进程 A (PID 1001)           进程 B (PID 1002)           进程 C (PID 1003)
┌──────────────────┐       ┌──────────────────┐       ┌──────────────────┐
│ redis_client     │       │ redis_client     │       │ redis_client     │
│  └─ Pool (max=10)│       │  └─ Pool (max=10)│       │  └─ Pool (max=10)│
│  ┌─────┬─────┐   │       │  ┌─────┬─────┐   │       │  ┌─────┬─────┐   │
│  │C-1  │     │   │       │  │C-1  │     │   │       │  │C-1  │     │   │
│  │BRPOP│空闲 │   │       │  │BRPOP│空闲 │   │       │  │BRPOP│空闲 │   │
│  └─────┴─────┘   │       │  └─────┴─────┘   │       │  └─────┴─────┘   │
│  _in_use: {C-1}  │       │  _in_use: {C-1}  │       │  _in_use: {C-1}  │
│  _available: []  │       │  _available: []  │       │  _available: []  │
│  实际连接: 1 个  │       │  实际连接: 1 个  │       │  实际连接: 1 个  │
└───────┬──────────┘       └───────┬──────────┘       └───────┬──────────┘
        │ TCP:54321                │ TCP:54322                │ TCP:54323
        ▼                          ▼                          ▼
┌──────────────────────────────────────────────────────────────────────────┐
│ Redis 服务端（端口 6380）                                                │
│  client(fd=10, addr=:54321, cmd=brpop)  ← 等待链表节点 1  ← 进程 A
│  client(fd=11, addr=:54322, cmd=brpop)  ← 等待链表节点 2  ← 进程 B
│  client(fd=12, addr=:54323, cmd=brpop)  ← 等待链表节点 3  ← 进程 C
│                                                                          │
│  task:queue → [client(fd=10), client(fd=11), client(fd=12)]             │
│                ↑ 第一个被 LPUSH 唤醒                                    │
└──────────────────────────────────────────────────────────────────────────┘

关键点：

每个进程的连接池 max=10，但实际只创建 1 个连接（brpop 独占 Conn-1）
每条 TCP 连接源端口不同（54321/54322/54323），目的端口都是 6380
Redis 端 CLIENT LIST 看到 3 行，无法判断它们属于几个进程

用 CLIENT LIST 验证：

id=1  addr=127.0.0.1:54321  fd=10  cmd=brpop  age=5  idle=5
id=2  addr=127.0.0.1:54322  fd=11  cmd=brpop  age=5  idle=5
id=3  addr=127.0.0.1:54323  fd=12  cmd=brpop  age=5  idle=5

3 个 Worker = 3 行 = 3 个 TCP 连接 = 3 个等待链表节点。

Q13：Redis 等待链表里存的是什么？是"Worker 进程"吗？

答：

不是 Worker 进程或 Worker 线程这种应用侧执行单元的抽象概念，而是具体的 redisClient 结构体指针，每个对应一个 TCP 连接的文件描述符（fd）。

// Redis 源码 server.h
typedef struct client {
    int fd;              // ← TCP 连接的文件描述符
    connection *conn;    // ← 连接对象
    // ... 其他字段
} client;

// 阻塞 key 的等待客户端链表
dict *db->blocking_keys;
//  key → list of client*

唤醒时 Redis 通过 fd 把数据写回给客户端。一个 client = 一个 fd = 一个 TCP 连接。

这里要特别区分两个概念：redis_client 是应用侧创建的 Redis 客户端对象，而等待链表中的 client 是 Redis 服务端维护的、正在阻塞等待某个 key 的 TCP 连接。两者不是同一个层面的概念，但它们通常通过同一条连接建立对应关系。

即使一个进程内有 3 个线程同时 BRPOP，redis-py 连接池也会借出 3 个不同的连接，Redis 端看到 3 个 client，等待链表有 3 个节点。Redis 不关心客户端是进程还是线程，只看 TCP 连接。

⚠️ 关键澄清：等待链表节点数 ≠ 连接总数

一个 Python 进程借出 3 个 TCP 连接 → Redis 端有 3 个 client 结构体。但等待链表里存几个，取决于有几个连接正在 BRPOP 等待。

核心规则：

等待链表节点数 = 当前正在执行 BRPOP 且未返回的 TCP 连接数
              ≠ 客户端进程数
              ≠ redis_client 对象数
              ≠ 连接池里的连接总数
              ≠ Redis 端的 client 结构体总数

场景对比：一个进程 + 3 个连接，等待链表节点数不同

场景 A：3 个连接都在 BRPOP（等待链表 3 个节点）

python

# 同一个进程，3 个线程同时 BRPOP 同一个 key
import threading

def worker():
    # 每个线程独立调用一次 brpop，借出各自的连接
    redis_client.brpop("task:queue", timeout=5)
    # ↑ 线程1 借 Conn-1，线程2 借 Conn-2，线程3 借 Conn-3
    # ↑ 3 个线程并发执行，3 个连接同时阻塞在 read_response()

# 启动 3 个线程
threads = [threading.Thread(target=worker) for _ in range(3)]
for t in threads:
    t.start()
for t in threads:
    t.join()

# 3 个线程 → 连接池借出 3 个 TCP 连接 → Redis 端 3 个 client → 都在等待

3 个线程并发执行 brpop 时的连接池状态:
  _in_use: {Conn-1, Conn-2, Conn-3}      ← 3 个连接都被借出
  _available: []                          ← 池空（还有 7 个余量未创建）

Redis 端:
  client(fd=10, cmd=brpop)  ← Conn-1（线程1）
  client(fd=11, cmd=brpop)  ← Conn-2（线程2）
  client(fd=12, cmd=brpop)  ← Conn-3（线程3）

等待链表:
  task:queue → [client(fd=10), client(fd=11), client(fd=12)]
                 ↑ 3 个节点，虽然来自同一个进程

LPUSH 时只唤醒 fd=10（FIFO 第一个），fd=11、fd=12 继续等。

场景 B：1 个 BRPOP + 2 个 hset（等待链表 1 个节点）

python

# 同一个进程，1 个线程 BRPOP 阻塞，2 个线程并行写进度
import threading

# 线程 1：BRPOP 阻塞等待任务
def consumer():
    result = redis_client.brpop("task:queue", timeout=5)
    # ↑ 借出 Conn-1，阻塞在 read_response()
    # ↑ 加入等待链表，未归还 Conn-1

# 线程 2：写任务进度（立即归还）
def writer_progress_image():
    redis_client.hset("task:6:progress", "stage", "IMAGE")
    # ↑ 借出 Conn-2 → 发送 HSET → 收到回复 → 立即归还 Conn-2

# 线程 3：写任务结果（立即归还）
def writer_result_done():
    redis_client.hset("task:6:progress", "stage", "DONE")
    # ↑ 借出 Conn-3 → 发送 HSET → 收到回复 → 立即归还 Conn-3

# 启动 3 个线程（consumer 先启动确保先进入 BRPOP）
t1 = threading.Thread(target=consumer)
t2 = threading.Thread(target=writer_progress_image)
t3 = threading.Thread(target=writer_result_done)

t1.start()
import time; time.sleep(0.1)   # 确保 consumer 先进入 BRPOP
t2.start()
t3.start()

t1.join(); t2.join(); t3.join()

线程并发期间连接池状态:
  _in_use: {Conn-1}                       ← 只有 Conn-1 还在阻塞（BRPOP 没返回）
  _available: [Conn-2, Conn-3]             ← Conn-2、Conn-3 已归还
  ↑ 注意：Conn-2、Conn-3 已归还但连接对象仍在池中（TCP 没断）

Redis 端:
  client(fd=10, cmd=brpop)    ← Conn-1 在等待链表里
  client(fd=11, cmd=hset)     ← Conn-2 已归还，但 client 结构体还在（TCP 仍存活）
  client(fd=12, cmd=hset)     ← Conn-3 已归还，但 client 结构体还在（TCP 仍存活）

等待链表:
  task:queue → [client(fd=10)]
                 ↑ 只有 1 个节点（只有 Conn-1 在 BRPOP）

3 个 client 都存在（TCP 连接还活着），但只有 1 个在等待链表里。

关键澄清：

归还 ≠ 关闭 TCP：连接归还到池中只是从 _in_use 移到 _available，TCP 连接保持打开状态，Redis 端 client 结构体仍然存在
client 结构体存活到 TCP 关闭：只有显式 disconnect() 或连接池销毁时，TCP 才关闭，Redis 端 client 才会从 CLIENT LIST 消失

场景 C：本项目实际（等待链表 1 个节点）

python

# 主循环（单线程）
def process_queue():
    while True:
        result = redis_client.brpop("task:queue", timeout=5)
        # ↑ 主线程借 Conn-1 BRPOP 阻塞（等待链表 1 个节点）
        if result:
            task_id = int(result[1])
            # ↑ brpop 已返回，Conn-1 已归还到池中
            run_task(task_id)
            # ↑ 此时主线程在执行 run_task，没有线程在 BRPOP

# run_task 内部启动 3 个分镜线程（ThreadPoolExecutor）
def run_task(task_id):
    redis_client.hset(f"task:{task_id}:progress", "stage", "SCRIPT")
    # ↑ 主线程借 Conn-1（复用）→ 立即归还

    # 3 线程并行调 Seedance API + 写进度
    from concurrent.futures import ThreadPoolExecutor
    def generate_frame_video(frame):
        # 调 Seedance API（不在 Redis 上阻塞）
        # ...
        # 写进度（借连接）
        redis_client.hset(f"task:{task_id}:progress",
                          f"frame_{frame.seq}", "done")
        # ↑ 各线程借 Conn-2/3/4 → 立即归还

    with ThreadPoolExecutor(max_workers=3) as executor:
        futures = [executor.submit(generate_frame_video, f) for f in frames]
        for future in futures:
            future.result()

阶段 1：BRPOP 阻塞期间（主线程在 brpop 中）
  连接池: _in_use: {Conn-1}   _available: []
  Redis 端: client(fd=10, cmd=brpop)
  等待链表: task:queue → [client(fd=10)]    ← 1 个节点

阶段 2：BRPOP 返回后，run_task 执行中（分镜并行）
  连接池: _in_use: {}   _available: [Conn-1, Conn-2, Conn-3, Conn-4]
           ↑ 主线程 hset 借 Conn-1 → 归还
           ↑ 3 个分镜线程 hset 借 Conn-2/3/4 → 归还
  Redis 端: client(fd=10,11,12,13, cmd=hset)
  等待链表: task:queue → []    ← 空！没有线程在 BRPOP

阶段 3：run_task 返回，主线程回到 while 循环顶部再次 brpop
  连接池: _in_use: {Conn-1}   _available: [Conn-2, Conn-3, Conn-4]
  Redis 端: client(fd=10, cmd=brpop)
  等待链表: task:queue → [client(fd=10)]    ← 1 个节点

关键：等待链表节点数随主线程状态变化——brpop 阻塞时 1 个节点，run_task 执行时 0 个节点（brpop 已返回，没有人在 BRPOP）。

速查表

概念	场景 A（3 线程 BRPOP）	场景 B（1 BRPOP + 2 hset）	场景 C（本项目）
Python 进程数	1	1	1
同时工作的线程数	3	3	1（主循环）/ 4（分镜并行时）
连接池 `_in_use` 大小	3	1（另外 2 个已归还）	1（brpop 期间）/ 0（run_task 期间）
连接池 TCP 连接总数	3	3（一旦建立不关闭）	1-4（按需创建）
Redis 端 client 结构体数	3	3（TCP 仍存活）	1-4（动态）
等待链表节点数	3	1	1（brpop 期间）/ 0（run_task 期间）

核心规律：

关系	说明
等待链表节点数 = ？	当前正在执行 BRPOP 且未返回的 TCP 连接数
已归还连接 → Redis 端 client 是否消失？	❌ 不消失，TCP 仍存活，client 结构体保留
已归还连接 → 等待链表是否还有？	❌ 没有，已归还说明 BRPOP 已返回
连接池 TCP 连接总数 = ？	历史最大借出数（一旦建立不主动关闭）

用 CLIENT LIST 验证

场景 A（3 个线程都在 BRPOP）：

bash

redis-cli CLIENT LIST

id=1  addr=127.0.0.1:54321  fd=10  cmd=brpop  age=5  idle=5   ← 在等待链表
id=2  addr=127.0.0.1:54322  fd=11  cmd=brpop  age=5  idle=5   ← 在等待链表
id=3  addr=127.0.0.1:54323  fd=12  cmd=brpop  age=5  idle=5   ← 在等待链表

3 行 = 3 个 client = 3 个 TCP 连接 = 等待链表 3 个节点（3 个都在等）。

场景 B（1 个 BRPOP + 2 个 hset）：

id=1  addr=127.0.0.1:54321  fd=10  cmd=brpop  age=5  idle=5   ← 在等待链表
id=2  addr=127.0.0.1:54322  fd=11  cmd=hset   age=5  idle=0   ← 不在等待链表
id=3  addr=127.0.0.1:54323  fd=12  cmd=hset   age=5  idle=0   ← 不在等待链表

3 行 = 3 个 client = 3 个 TCP 连接，但等待链表只有 1 个节点（只有 cmd=brpop 且 idle 持续增长的在等）。

本章小结

等待链表里存的是"正在 BRPOP 等待的 TCP 连接"，不是"客户端进程"。一个进程借出 3 个连接，如果 3 个都在 BRPOP，等待链表就有 3 个节点；如果只有 1 个在 BRPOP，等待链表就只有 1 个节点。Redis 只看 TCP 连接，不看连接属于哪个进程。

补充：每条 TCP 连接都有独立的源端口

关键概念：一条 TCP 连接由四元组唯一标识：

TCP 连接 = (源 IP, 源端口, 目的 IP, 目的端口)

本项目场景：

目的 IP:目的端口 = 127.0.0.1:6380   ← Redis 服务端固定
源 IP            = 127.0.0.1        ← 客户端机器固定
源端口           = ???              ← 每条连接不同！

源端口是操作系统分配的，每建立一条 TCP 连接，内核就分配一个空闲的源端口：

Conn-1: 127.0.0.1:54321 ←► 127.0.0.1:6380
Conn-2: 127.0.0.1:54322 ←► 127.0.0.1:6380
Conn-3: 127.0.0.1:54323 ←► 127.0.0.1:6380
         ↑ 源端口不同       ↑ 目的端口相同（都是 Redis）

同一进程的 3 个连接也是不同端口（代码示例见上文 Q13 场景 A）：

python

# 简化示例：一个进程，主线程 + 2 个子线程，共 3 个 TCP 连接
import threading, redis

redis_client = redis.Redis(host="localhost", port=6380, max_connections=10)

# 主线程借 Conn-1
redis_client.hset("k", "main", "v1")

# 子线程各自借 Conn-2、Conn-3
def worker(tag):
    redis_client.hset("k", tag, f"v-{tag}")

t1 = threading.Thread(target=worker, args=("t1",))
t2 = threading.Thread(target=worker, args=("t2",))
t1.start(); t2.start(); t1.join(); t2.join()

此时 CLIENT LIST 输出（3 条 TCP 连接，3 个源端口）：

id=1  addr=127.0.0.1:54321  fd=10  cmd=hset   age=1  idle=0   ← Conn-1（主线程）
id=2  addr=127.0.0.1:54322  fd=11  cmd=hset   age=1  idle=0   ← Conn-2（线程 t1）
id=3  addr=127.0.0.1:54323  fd=12  cmd=hset   age=1  idle=0   ← Conn-3（线程 t2）
              ↑                ↑
              3 个源端口不同     3 个 fd 不同
              但是同一个进程！

Redis 端无法从 CLIENT LIST 判断这 3 个连接属于同一个进程还是不同进程——它只看到 3 条独立的 TCP 连接。

为什么每条 TCP 连接需要独立端口

TCP 协议规定，一个端口同一时刻只能被一条 TCP 连接的某一端使用：

端口 54321  → 只能被 Conn-1 的客户端侧占用
端口 54322  → 只能被 Conn-2 的客户端侧占用
端口 54323  → 只能被 Conn-3 的客户端侧占用
端口 6380   → Redis 服务端共用一个端口（但能区分连接）

Redis 服务端怎么用同一个端口 6380 服务多条连接：靠 fd（文件描述符）区分。每个 TCP 连接到达服务端后，Redis 通过 accept() 创建一个新的 fd，然后把它和 client 结构体绑定。

// Redis 服务端源码（简化）
while (1) {
    int client_fd = accept(server_fd, ...);  // ← 每个连接 accept 一次
    // ↑ server_fd 绑定 6380 端口，但 client_fd 是新的
    client *c = createClient(client_fd);
    // ↑ 每个 client_fd 对应一个 client 结构体
}

层面	端口	fd
客户端侧	每条 TCP 连接独立源端口	每条连接独立 fd
服务端侧	共用 6380 监听端口	每条连接独立 fd

用 `lsof` / `netstat` 验证

bash

# 查看 Redis 服务端（6380）的连接
lsof -i :6380

COMMAND   PID  USER   FD  TYPE  DEVICE  SIZE/OFF  NODE  NAME
redis-ser 1234  user   6u  IPv4  0x1234    0t0   TCP  *:6380 (LISTEN)        ← 监听 socket
redis-ser 1234  user  10u  IPv4  0x5678    0t0   TCP  localhost:6380→localhost:54321
redis-ser 1234  user  11u  IPv4  0x5679    0t0   TCP  localhost:6380→localhost:54322
redis-ser 1234  user  12u  IPv4  0x567a    0t0   TCP  localhost:6380→localhost:54323
                              ↑                             ↑
                              服务端 fd（每条不同）          客户端源端口（每条不同）

3 条连接 = 3 个服务端 fd（10/11/12）= 3 个客户端源端口（54321/54322/54323）。

端口范围

操作系统给客户端分配源端口时，从一个固定范围内选：

bash

# Linux 查看端口范围
$ cat /proc/sys/net/ipv4/ip_local_port_range
32768   60999

# macOS 查看
$ sysctl net.inet.ip.portrange.first net.inet.ip.portrange.last
net.inet.ip.portrange.first: 1024
net.inet.ip.portrange.last: 65535

理论上一个客户端机器最多能连 28000 多条 TCP 到同一个 Redis 端口（Linux），但实际 Worker 进程不会这么多。

端口总结表

问题	答案
同一客户端的 3 个连接，CLIENT LIST 显示 3 行吗？	✅ 是
这 3 行的端口相同吗？	❌ 不同，每条 TCP 连接独立源端口
每条 TCP 连接都有独立端口吗？	✅ 是（客户端侧源端口不同）
Redis 服务端端口相同吗？	✅ 相同（都是 6380，靠 fd 区分）
Redis 能从 CLIENT LIST 判断连接属于哪个进程吗？	❌ 不能，只看 TCP 连接

核心：一条 TCP 连接 = (源 IP, 源端口, 目的 IP, 目的端口) 四元组。同一个进程的 3 条连接，源端口必然不同，所以 CLIENT LIST 看到 3 行端口不同的记录。

端口与连接速查表

维度	客户端侧	服务端（Redis）侧
IP	127.0.0.1（客户端机器）	127.0.0.1（Redis 机器）
端口	每条 TCP 连接独立源端口（54321/54322/54323）	共用 6380 监听端口
fd	每条连接独立 fd	每条连接独立 fd（10/11/12）
区分连接靠什么	源端口	fd（accept 返回的新 fd）
一个端口能被多少连接使用	同一时刻只能被一条 TCP 连接的客户端侧占用	同一端口可服务多条连接（靠 fd 区分）

场景	Python 进程数	TCP 连接数	源端口数	Redis 端 fd 数	CLIENT LIST 行数
3 进程 × 1 连接	3	3	3（每进程 1 个）	3	3
1 进程 × 3 连接（3 线程 BRPOP）	1	3	3（同一进程）	3	3
1 进程 × 1 连接（本项目主循环）	1	1	1	1	1
1 进程 × 4 连接（分镜并行）	1	4	4	4	4

问题	答案
同一客户端的 3 个连接，CLIENT LIST 显示 3 行吗？	✅ 是
这 3 行的端口相同吗？	❌ 不同，每条 TCP 连接独立源端口
每条 TCP 连接都有独立端口吗？	✅ 是（客户端侧源端口不同）
Redis 服务端端口相同吗？	✅ 相同（都是 6380，靠 fd 区分）
Redis 能从 CLIENT LIST 判断连接属于哪个进程吗？	❌ 不能，只看 TCP 连接
同一进程的 3 条连接源端口能相同吗？	❌ 不能，TCP 协议规定一个端口同一时刻只能被一条连接的某一端使用
Redis 服务端怎么用 6380 服务多条连接？	监听 socket 绑定 6380，accept 时为每条连接创建独立 fd
Linux 客户端源端口范围	32768-60999（约 28000 个）
macOS 客户端源端口范围	1024-65535

验证命令	作用
`redis-cli CLIENT LIST`	查看 Redis 端所有 client（含 addr、fd、cmd、idle）
`lsof -i :6380`	查看 6380 端口的所有 TCP 连接（含客户端源端口）
`netstat -an \| grep 6380`	查看与 6380 相关的连接状态
`cat /proc/sys/net/ipv4/ip_local_port_range`	Linux 查看客户端源端口范围
`sysctl net.inet.ip.portrange.first net.inet.ip.portrange.last`	macOS 查看客户端源端口范围

Q14：如果用 `multiprocessing.Process` 的 `fork` 模式启动 Worker，Redis 连接会出什么问题？

答：

fork 会复制父进程的内存，包括 redis_client 和已建立的 TCP 连接的 fd。两个进程向同一个 socket 写数据会导致 Redis 协议错乱。

python

# ❌ fork 模式的问题
import multiprocessing
from app.redis_client import redis_client  # ← 父进程已创建

def worker():
    # ← fork 后子进程继承了父进程的 redis_client
    # ← 连接池中的 TCP fd 被复制
    # ← 两个进程操作同一个 socket → 数据错乱！
    redis_client.brpop("task:queue", timeout=5)

p = multiprocessing.Process(target=worker)
p.start()  # fork!

正确做法：子进程中重新创建 redis_client：

python

def worker():
    import redis
    # 子进程中重新创建，不复用父进程的连接池
    client = redis.Redis(
        host="localhost",
        port=6380,
        decode_responses=True,
        max_connections=10,
    )
    client.brpop("task:queue", timeout=5)

本项目不受影响：python -m app.worker 每次都是全新进程，不走 fork，不存在继承问题。

Q15：一个 Worker 进程实际同时占用的连接数是多少？连接池 `max_connections=10` 是不是浪费？

答：

实际同时占用的连接数取决于是否启用 Seedance 分镜并行：

模式	占用连接数	说明
单线程顺序执行（无分镜并行）	1 个	brpop 返回后才执行 hset，全程只用 1 个连接
3 线程分镜并行（本项目实际启用）	3-4 个	3 个分镜线程 + 主线程 BRPOP

1. 单线程顺序执行模式（1 个连接）

阶段	占用连接	说明
`process_queue` 中 BRPOP 阻塞	1 个	只有 BRPOP 在执行，线程被挂起
`run_task` 中写进度	1 个	BRPOP 已归还，HSET 复用 Conn-1
`run_task` 中推审核队列	1 个	同上
峰值	1 个	brpop 和 hset 顺序执行，不会同时

2. 3 线程分镜并行模式（3-4 个连接，本项目实际配置）

python

# orchestrator.py 实际代码
with ThreadPoolExecutor(max_workers=3) as executor:
    futures = [executor.submit(generate_frame_video, frame) for frame in frames]
    # ↑ 3 个线程并行调用 Redis 写进度
    # ↑ 同时主线程可能还在 brpop 阻塞（Conn-1 被独占）

# 连接池状态：
#   _in_use: {Conn-1(BRPOP), Conn-2, Conn-3, Conn-4}
#   _available: [Conn-5, ..., Conn-10]
#   ↑ max_connections=10 留 6 个余量，不浪费

阶段	占用连接	说明
主线程 BRPOP 阻塞	Conn-1	独占
分镜线程 1 写进度	Conn-2	并行借出
分镜线程 2 写进度	Conn-3	并行借出
分镜线程 3 写进度	Conn-4	并行借出
峰值	4 个	1 BRPOP + 3 分镜并行

3. 为什么 `max_connections=10` 不浪费

连接是惰性创建的：max_connections=10 只是上限，实际用几个建几个，不用的不建
分镜并行需要 3-4 个：3 线程并行 + 主线程 BRPOP，至少需要 4 个连接
留 6 个余量：未来扩展到 5-6 线程并行也不用改配置

4. 多 Worker 进程的总连接数

Worker 进程数	单进程连接峰值	总 TCP 连接	Redis max_clients
1	4	4	默认 10000
3	4	12	足够
10	4	40	足够
50	4	200	足够
200	4	800	足够

详见 Redis连接池详解.md 1.3 节 — 单线程单连接 vs 多线程多连接的完整对比。

第五章：Redis 6.0 多线程 I/O

本章导读

本章围绕“Redis 6.0 多线程 I/O”展开，先交代适用场景，再说明核心原理，随后给出典型示例，最后补充小结与延伸阅读。

阅读顺序： 结论 → 原理 → 场景 → 代码 → 小结

Q16：Redis 6.0 引入了多线程 I/O，这会影响 BRPOP 的原子性保证吗？

答：

不会影响。多线程 I/O 只用于读写 socket 数据，命令执行仍然在主线程串行进行。

                  ┌──────────────────────────┐
                  │       Redis 6.0          │
                  │                          │
  多线程 I/O      │  ┌────┐ ┌────┐ ┌────┐    │
  （读取/写入     │  │IO-1│ │IO-2│ │IO-3│    │
   socket 数据）  │  └─┬──┘ └─┬──┘ └─┬──┘    │
                  │    │      │      │       │
                  │    └──────┼──────┘       │
                  │           │               │
                  │     ┌─────┴─────┐         │
                  │     │  主线程    │         │
                  │     │ (命令执行) │         │
                  │     └───────────┘         │
                  │                           │
                  └──────────────────────────┘

  1. IO 线程从 socket 读取客户端数据
  2. 主线程执行命令（单线程，串行）  ← 原子性在这里保证
  3. IO 线程把结果写回 socket

3 个 IO 线程可以同时读取 socket 数据，但主线程逐条执行 BRPOP，仍然保证原子性。

层面	线程数	职责
I/O 线程	1-N（可配置）	读取 socket 数据、写回 socket 数据
主线程	1	解析命令、执行命令、操作数据结构

第六章：崩溃恢复与任务丢失

本章导读

本章围绕“崩溃恢复与任务丢失”展开，先交代适用场景，再说明核心原理，随后给出典型示例，最后补充小结与延伸阅读。

阅读顺序： 结论 → 原理 → 场景 → 代码 → 小结

Q17：Worker 从 BRPOP 取出任务后、执行前崩溃了，这个任务会怎样？

答：

任务会永久丢失。因为 BRPOP 是"弹出即删除"，取出后任务已从 Redis 队列消失。

python

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

解决方案：RPOPLPUSH 模式，弹出时同时放入"处理中"队列：

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)

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

参数	类型	说明
`key`	`str`	要操作的列表名
`count`	`int`	正数：从头部开始删 count 个；负数：从尾部开始删；0：删除全部匹配项
`value`	`str`	要删除的元素值

Q18：本项目如何监控队列积压？多少个任务算积压？

答：

用 LLEN 命令检查队列长度：

python

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

积压数量	严重程度	说明
0-5	正常	Worker 能及时消费
5-20	注意	可能需要增加 Worker
20-50	警告	用户等待时间明显增加
50+	严重	需要立即增加 Worker 或检查是否有任务卡住

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

第七章：超时与竞争

本章导读

本章围绕“超时与竞争”展开，先交代适用场景，再说明核心原理，随后给出典型示例，最后补充小结与延伸阅读。

阅读顺序： 结论 → 原理 → 场景 → 代码 → 小结

Q19：BRPOP 超时和 LPUSH 同时发生会怎样？会不会出现"既收到 nil 又拿到元素"的情况？

答：

不会。超时检查和 LPUSH 都在主线程的事件循环中串行执行，不可能"同时"发生。

情况 A：LPUSH 先执行

1. LPUSH 执行，发现 W1 在等待链表
2. 把元素发给 W1
3. 移除 W1
4. W1 超时定时器还在，但 W1 已不在等待链表
5. 超时触发时发现 W1 不在链表 → 跳过

结果：W1 拿到元素 ✅

情况 B：超时先执行

1. 超时触发，W1 从等待链表移除
2. 返回 nil 给 W1
3. LPUSH 执行，发现等待链表空（或只有 W2）
4. 元素留在列表中

结果：W1 收到 nil，元素留给下一个 BRPOP ✅

两种情况都不会出现"W1 既收到 nil 又拿到元素"或"元素丢失"。超时处理不会被 LPUSH 打断，LPUSH 也不会被超时处理打断。

Q20：LPUSH + BRPOP 实现 FIFO 队列，那 LPUSH + LPOP 是什么模型？为什么本项目不用 LPUSH + LPOP？

答：

推入	弹出	组合	效果
LPUSH	BRPOP/RPOP	左推右弹	FIFO（先进先出队列）
LPUSH	LPOP/BLPOP	左推左弹	LIFO（后进先出栈）
RPUSH	BLPOP/LPOP	右推左弹	FIFO（先进先出队列）
RPUSH	BRPOP/RPOP	右推右弹	LIFO（后进先出栈）

本项目用 LPUSH + BRPOP = FIFO 队列：

LPUSH task:queue 6     ──►    [6]
LPUSH task:queue 7     ──►    [7, 6]
LPUSH task:queue 8     ──►    [8, 7, 6]
                              BRPOP ──►  取出 6（最早推入）
                             [8, 7]
                              BRPOP ──►  取出 7
                             [8]
                              BRPOP ──►  取出 8
                             []

如果用 LPUSH + LPOP（LIFO 栈），后提交的任务会先被消费，先提交的任务可能长时间不被处理，不公平。

第八章：综合场景

本章导读

本章围绕“综合场景”展开，先交代适用场景，再说明核心原理，随后给出典型示例，最后补充小结与延伸阅读。

阅读顺序： 结论 → 原理 → 场景 → 代码 → 小结

Q21：100 个用户同时提交任务，从 BRPOP 到任务完成，完整经历了哪些步骤？

答：

时间 0s: 100 个 POST /creative/submit 同时到达
         100 个 task_id 被 LPUSH 到 task:queue
         Redis 列表: [100, 99, 98, ..., 2, 1]

时间 0s: Worker-1 BRPOP → 取出 1（FIFO 最早推入）
         Redis 等待链表: [W2, W3, ..., W10]（9 个 Worker 继续等）
         列表: [100, 99, ..., 2]

         → 主线程执行 LPUSH 后，第1轮唤醒 W1
         → 第2轮唤醒 W2（取出 2）
         → 第3轮唤醒 W3（取出 3）
         → ...
         → 第10轮唤醒 W10（取出 10）
         → 列表: [100, 99, ..., 11]（90 个任务积压）

时间 0s: W1~W10 各自处理任务 1~10
         W11~W100 在列表中等待

时间 4m: W1 完成任务 1，回到 process_queue，BRPOP → 取出 11
         W2 完成任务 2，BRPOP → 取出 12
         ...

时间 40m: 第 100 个任务被处理完成

关键：Redis 单线程在 LPUSH 后逐个唤醒 Worker，10 个 Worker 在几毫秒内全部被唤醒并拿到不同任务。瓶颈不在 Redis，而在 Worker 处理速度（每个任务 4 分钟）。

Q22：如果 Redis 服务重启了，等待链表和队列中的任务会怎样？

答：

全部丢失。Redis 重启后内存数据清空：

重启前:
  task:queue: [50, 49, 48, ..., 1]    ← 50 个待处理任务
  等待链表: [W1, W2, W3]               ← 3 个 Worker 在等

重启后:
  task:queue: []                       ← 空
  等待链表: []                          ← 空
  3 个 Worker 的 BRPOP 超时返回 nil

应对方案：

方案	说明
AOF 持久化	Redis 重启后从 AOF 文件恢复队列数据
RDB 快照	定期快照，但可能丢失最近的任务
数据库兜底	任务同时写入 PostgreSQL，Worker 启动时扫描数据库中 `status=QUEUED` 的任务重新入队

本项目的方案：任务同时存在于 PostgreSQL 和 Redis。如果 Redis 重启，可以从 PostgreSQL 恢复：

python

# Worker 启动时检查数据库中未完成的任务
def recover_tasks():
    db = SessionLocal()
    stuck_tasks = db.scalars(
        select(Task).where(Task.status.in_([
            TaskStatus.QUEUED,
            TaskStatus.SCRIPT,
            TaskStatus.IMAGE,
        ]))
    ).all()
    for task in stuck_tasks:
        redis_client.lpush("task:queue", task.id)
    db.close()

Q23：为什么本项目不用 RabbitMQ 或 Kafka，而是用 Redis BRPOP？

答：

维度	Redis BRPOP	RabbitMQ	Kafka
部署复杂度	低（已有 Redis）	中（需独立部署）	高（需 ZooKeeper + Broker）
原子性保证	单线程串行	Channel 级别锁	分区单消费者
任务确认(ACK)	无（需 RPOPLPUSH 弥补）	内置 ACK 机制	Offset 提交
延迟	< 1ms	~1ms	~10ms
吞吐量	10 万+ QPS	万级 QPS	百万级 QPS
适用规模	中小规模（< 200 Worker）	中大规模	超大规模

选择 Redis 的原因：

项目已经依赖 Redis（进度缓存、审核队列），无需额外部署
MVP 阶段任务量不大（每天几十到几百个任务），Redis 足够
部署简单，运维成本低
BRPOP 延迟极低（< 1ms），满足实时性要求

何时考虑迁移：

Worker 超过 200 个，Redis 单线程开始成为瓶颈
需要复杂的任务路由、重试、死信队列
需要消息持久化保证（Redis 重启丢数据风险）

第九章：源码级深入

本章导读

本章围绕“源码级深入”展开，先交代适用场景，再说明核心原理，随后给出典型示例，最后补充小结与延伸阅读。

阅读顺序： 结论 → 原理 → 场景 → 代码 → 小结

Q24：`signalKeyAsReady` 和 `handleClientsBlockedOnKeys` 的调用关系是什么？为什么唤醒后要 `break`？

答：

调用关系：

LPUSH 命令执行
  └── lpushCommand()
      ├── listTypePush()           ← 插入元素到列表
      └── signalKeyAsReady()      ← 标记 key 为"有待处理的阻塞客户端"
          └── 把 key 加入 server.ready_keys 队列

LPUSH 命令执行完毕后，主线程事件循环:
  └── handleClientsBlockedOnKeys()
      ├── 从 ready_keys 取出 key
      ├── 查找该 key 的等待链表
      ├── 取链表第一个 client
      ├── 弹出列表元素
      ├── 发给 client
      ├── 从等待链表移除 client
      └── break ← 只唤醒一个

为什么 break：

while (clientNode) {
    client *receiver = clientNode->value;
    robj *value = listTypePop(o, LIST_TAIL);
    if (value) {
        addReply(receiver, value);
        listDelNode(clients, clientNode);
        break;  // ← 弹出一个元素，唤醒一个客户端，跳出循环
    }
    clientNode = listNextNode(clientNode);
}

break 保证一次只唤醒一个 Worker。如果不用 break，一个元素可能被发给多个 Worker，导致重复消费。

如果 LPUSH 6 7 8 推入了 3 个元素，主线程会在下一轮事件循环中再次处理 ready_keys，逐个唤醒 3 个 Worker。

Q25：`brpop` 在 redis-py 内部是如何借出和归还连接的？和 `hset` 有什么区别？

答：

redis-py 的 ConnectionPool 采用"借出-归还"模式：

python

# redis-py 内部简化逻辑
class ConnectionPool:
    def get_connection(self):
        if self._available:                    # 有空闲连接
            conn = self._available.pop()
            self._in_use.add(conn)
            return conn
        elif len(self._in_use) < self._max:    # 还没到上限
            conn = self._make_connection()     # 新建 TCP 连接
            self._in_use.add(conn)
            return conn
        else:
            raise ConnectionError("Too many connections")

    def release(self, conn):
        self._in_use.remove(conn)
        self._available.append(conn)           # 归还到空闲栈

1. brpop 和 hset 各自的借还流程

brpop 借出不归还（阻塞期间）：

python

conn = pool.get_connection()      # 借出 Conn-1
conn.send_command("BRPOP", "task:queue", 5)
result = conn.read_response()      # 阻塞！最多等 5 秒
                                   # ← 期间 Conn-1 一直在 _in_use 中
pool.release(conn)                 # 超时或拿到任务后才归还

hset 借出后立即归还：

python

conn = pool.get_connection()       # 借出 Conn-2
conn.send_command("HSET", "task:6:progress", "stage", "IMAGE")
result = conn.read_response()       # 立即返回（纳秒级）
pool.release(conn)                  # 立即归还 Conn-2

命令	借出时长	归还时机	单次占用连接数
`brpop`（阻塞）	最长 5 秒	超时或拿到元素后	1（独占）
`hset`	~100 纳秒	命令返回后立即	1（瞬间）

2. ⚠️ 关键澄清：单线程下两者不会"同时"占用连接

单线程顺序执行（本项目主循环）：

python

def process_queue():
    while True:
        result = redis_client.brpop("task:queue", timeout=5)
        # ↑ brpop 阻塞期间，hset 根本执行不到
        # ↑ python 线程被 read_response() 挂起
        if result:
            run_task(int(result[1]))  # ← hset 在这里执行
            # ↑ 此时 brpop 已归还 Conn-1，hset 复用 Conn-1

单线程下连接池状态：池里实际只创建 1 个连接（Conn-1 反复借出/归还），max_connections=10 设多大都没意义。

3. 真正需要 2 个连接的场景：多线程

python

# 线程 A：brpop 阻塞中（独占 Conn-1）
def consumer():
    while True:
        result = redis_client.brpop("task:queue", timeout=5)

# 线程 B：hset 并行执行（借 Conn-2）
def progress_writer():
    while True:
        redis_client.hset("task:6:progress", ...)
        # ↑ 借 Conn-2（因为 Conn-1 被线程 A 独占）

多线程下连接池状态：_in_use: {Conn-1, Conn-2}，max_connections >= 2 才能跑。

4. 本项目实际的连接占用

场景	brpop 期间	hset 何时执行	连接占用
主循环（单线程）	Conn-1 独占，线程挂起	brpop 返回后顺序执行	1 个（Conn-1 复用）
Seedance 3 线程分镜并行	Conn-1 独占	3 个分镜线程并行借 Conn-2/3/4	4 个

详见 Redis连接池详解.md 1.3 节 — 单线程单连接 vs 多线程多连接的完整对比，以及"连接是按需创建的，不是一次性预创建"的源码验证。

第十章：Redis 常用数据结构与典型使用场景

Q26：Redis 常用的数据结构有哪些？分别适合什么场景？

答：

Redis 不是简单的 key-value 缓存。它提供了多种数据结构，不同结构解决的问题不同。选型时不要只问“能不能存”，而要问：数据是否需要按字段更新、是否需要排序、是否需要去重、是否需要阻塞消费、是否需要可靠确认、是否允许近似统计、是否需要按地理位置查询。

Redis 常用数据结构总览图

教科书式可编辑画板版总图： [redis-ch10-data-structures-overview.html](file:///Users/binny/workspace-sdd/aiNotes/redis/redis-ch10-data-structures-overview.html)

Excalidraw 源文件： [redis-ch10-data-structures-overview.excalidraw](file:///Users/binny/workspace-sdd/aiNotes/redis/redis-ch10-data-structures-overview.excalidraw)

下面保留一份纯文本版总览图，便于在不支持画板预览的 Markdown 阅读器中快速查看结构：

text

                         Redis
                           │
   ┌───────────────┬───────┼───────────────┬───────────────┐
   │               │       │               │               │
 String         Hash     List / Stream    Set / ZSet      其他模块
   │               │       │               │               │
单值缓存、计数   对象字段   队列/消息流   去重/排序/排名   JSON、搜索、图形化、统计
                              │               │               │
                           阻塞消费         会员关系      向量、过滤器、指标

可以先用下面这张表建立整体印象：

数据结构 / 能力	核心特点	最典型场景	常用命令
String	单 key 单 value，可做字符串、数字、JSON、二进制	缓存、计数器、验证码、分布式锁	`GET`、`SET`、`INCR`、`SET NX EX`
Hash	一个 key 下多个 field	用户资料、任务进度、商品详情	`HSET`、`HGET`、`HGETALL`、`HINCRBY`
List	有序列表，两端插入/弹出	任务队列、最新列表、栈	`LPUSH`、`BRPOP`、`LRANGE`、`LTRIM`
Set	无序去重集合	点赞用户、标签、共同关注	`SADD`、`SISMEMBER`、`SINTER`
ZSet	带 score 的有序集合	排行榜、延迟队列、时间线排序	`ZADD`、`ZREVRANGE`、`ZRANGEBYSCORE`
Stream	带消息 ID 的日志型消息流	可靠队列、消费组、事件日志	`XADD`、`XREADGROUP`、`XACK`
Bitmap	bit 位状态存储	签到、活跃用户、灰度标记	`SETBIT`、`GETBIT`、`BITCOUNT`
BitField	在字符串中按位段读写整数	多个小整数状态压缩存储	`BITFIELD`
HyperLogLog	大规模去重数量估算	UV、访问人数估算	`PFADD`、`PFCOUNT`
GEO	经纬度位置索引	附近门店、附近的人	`GEOADD`、`GEOSEARCH`、`GEODIST`
Pub/Sub	实时发布订阅	实时通知、配置变更广播	`PUBLISH`、`SUBSCRIBE`
RedisJSON	JSON 文档读写，需模块支持	复杂对象缓存、局部 JSON 更新	`JSON.SET`、`JSON.GET`
Bloom Filter	概率型存在性判断，需模块支持	防缓存穿透、黑名单预判	`BF.ADD`、`BF.EXISTS`
TimeSeries	时间序列数据，需模块支持	监控指标、埋点趋势	`TS.ADD`、`TS.RANGE`
RedisSearch	全文检索和二级索引，需模块支持	商品搜索、文档搜索、任务检索	`FT.CREATE`、`FT.SEARCH`
Vector Set / Vector Search	向量相似度检索，需 Redis Stack / 新版本能力	语义搜索、相似视频、RAG 检索	`FT.SEARCH`、`VSIM`
Cuckoo Filter	可删除的概率型存在性判断，需模块支持	可删除黑名单、临时去重	`CF.ADD`、`CF.EXISTS`、`CF.DEL`
Count-Min Sketch	频次估算，需模块支持	热点 key、热门搜索词、点击频次估算	`CMS.INCRBY`、`CMS.QUERY`
TopK	高频元素近似统计，需模块支持	热门视频、热门标签、热门搜索词	`TOPK.ADD`、`TOPK.LIST`
TDigest	分位数统计，需模块支持	P95/P99 延迟、耗时分布	`TDIGEST.ADD`、`TDIGEST.QUANTILE`

Redis 常用数据结构的图形化、CLI 与案例一体化展示

下面把每种数据结构的场景概述、图形化展示、redis-cli 命令和 Python 示例放在同一个小节里，便于横向对照阅读。

1. String：字符串 / 数值 / JSON 缓存 / 分布式锁

场景概述： String 适合存放单个值、计数器、缓存和锁标记。

图形化展示：

text

user:1001  ──►  "{name: 'Tom', level: 3}"
video:6:view_count  ──►  "1024"
lock:task:6  ──►  "random-token"

redis-cli 命令：

bash

redis-cli SET task:6:status RUNNING EX 900
redis-cli GET task:6:status
redis-cli INCR video:6:view_count
redis-cli SET lock:task:6 abc123 NX EX 300

Python 示例：

python

redis_client.set("task:6:status", "RUNNING", ex=900)
redis_client.incr("video:6:view_count")

适合整体读写单个值、计数、验证码、登录态、分布式锁。

2. Hash：对象字段缓存 / 任务进度 / 用户信息

场景概述： Hash 适合保存对象型数据并进行字段级更新。

图形化展示：

text

task:6:progress
  ├── stage   ──► IMAGE
  ├── percent ──► 45
  └── message ──► 正在生成分镜图

redis-cli 命令：

bash

redis-cli HSET task:6:progress stage IMAGE percent 45 message "正在生成分镜图"
redis-cli HGET task:6:progress stage
redis-cli HGETALL task:6:progress

Python 示例：

python

redis_client.hset("task:6:progress", mapping={"stage": "IMAGE", "percent": 45})
progress = redis_client.hgetall("task:6:progress")

适合一个对象有多个字段、并且经常局部更新。

3. List：队列 / 栈 / 最新列表 / 简单任务流

场景概述： List 适合实现队列、栈和简单的消息流。

图形化展示：

text

task:queue
LPUSH 入队 -> [ task:9 ] -> [ task:8 ] -> [ task:7 ] -> [ task:6 ] <- BRPOP 出队

redis-cli 命令：

bash

redis-cli LPUSH task:queue 6
redis-cli BRPOP task:queue 5
redis-cli LRANGE user:1001:feeds 0 19

Python 示例：

python

redis_client.lpush("task:queue", 6)
result = redis_client.brpop("task:queue", timeout=5)

适合简单队列、最新列表、撤销栈。

4. Set：去重集合 / 标签 / 点赞 / 关系运算

场景概述： Set 适合去重、标签关系和集合运算。

与 ZSet 的区别：

维度	Set	ZSet
底层数据结构	哈希表（dict）	跳表（skiplist）+ 字典（dict）
是否去重	是	是
是否有序	否	是，按 score 排序
元素结构	只有 member	member + score
典型用途	去重、标签、共同关注	排行榜、延迟队列、时间线
适合“只关心是否存在”	是	也可以，但通常没必要
适合“要按顺序取前 N 个”	不适合	很适合

Set 更像“成员名单”，ZSet 更像“带分数的排名名单”。

底层结构说明：

Set 内部通常使用哈希表存储成员，查找、插入、删除平均都是 O(1)。 ZSet 同时维护字典和跳表：字典负责快速按 member 查找，跳表负责按 score 排序和范围查询。

图形化展示：

text

video:6:liked_users
{ 1001, 1002, 1003, 1008 }

redis-cli 命令：

bash

redis-cli SADD video:6:liked_users 1001 1002 1003
redis-cli SISMEMBER video:6:liked_users 1001
redis-cli SINTER user:1:follows user:2:follows

Python 示例：

python

redis_client.sadd("video:6:liked_users", 1001, 1002, 1003)
liked = redis_client.sismember("video:6:liked_users", 1001)

适合去重、共同关注、抽奖用户池。

5. ZSet：排行榜 / 权重排序 / 延迟队列 / 时间线

场景概述： ZSet 适合排行榜、延迟任务和按分值排序。

与 Set 的区别：

维度	Set	ZSet
是否去重	是	是
是否有序	否	是，按 score 排序
元素结构	只有 member	member + score
典型用途	去重、标签、共同关注	排行榜、延迟队列、时间线
取前 N 个元素	只能随机或集合运算	可以按分值直接取前 N 名

Set 更像“无序去重名单”，ZSet 更像“带分数的排序名单”。

字母 z 的来源：

ZSet 全名是 Sorted Set。Redis 早期命名时为了和 Set 区分，就在前面加了一个 z，表示 sorted，所以 zset 可以理解为“有序集合”。

图形化展示：

text

rank:video:hot
98.5  ──► video:8
91.0  ──► video:3
76.5  ──► video:6

redis-cli 命令：

bash

redis-cli ZADD rank:video:hot 98.5 video:8 91.0 video:3 76.5 video:6
redis-cli ZREVRANGE rank:video:hot 0 9 WITHSCORES
redis-cli ZADD delay:order:cancel 1787797000 order:1001
redis-cli ZRANGEBYSCORE delay:order:cancel -inf 1787797060

Python 示例：

python

redis_client.zadd("delay:order:cancel", {"order:1001": 1787797000})
expired = redis_client.zrangebyscore("delay:order:cancel", "-inf", 1787797060)

适合排行榜、延迟任务、订单超时取消。

6. Stream：可靠消息流 / 消费组 / 事件日志

场景概述： Stream 适合可靠消息流、消费组和事件追踪。

图形化展示：

text

stream:task
1690000000000-0  ──► task_id=6, stage=SCRIPT
1690000001000-0  ──► task_id=6, stage=IMAGE

redis-cli 命令：

bash

redis-cli XADD stream:task * task_id 6 stage SCRIPT
redis-cli XGROUP CREATE stream:task video-workers 0 MKSTREAM
redis-cli XREADGROUP GROUP video-workers worker-1 COUNT 10 BLOCK 5000 STREAMS stream:task >
redis-cli XACK stream:task video-workers 1690000000000-0

Python 示例：

python

redis_client.xadd("stream:task", {"task_id": "6", "stage": "SCRIPT"})

适合可靠消息流、事件日志、异步任务消费。

7. Bitmap：签到 / 活跃用户 / 布尔状态压缩存储

场景概述： Bitmap 适合高密度布尔状态和签到统计。

图形化展示：

text

active:2026-08-27
用户偏移: 1001 1002 1003 1004
bit 值:     1    0    1    1

redis-cli 命令：

bash

redis-cli SETBIT active:2026-08-27 1001 1
redis-cli GETBIT active:2026-08-27 1001
redis-cli BITCOUNT active:2026-08-27

Python 示例：

python

redis_client.setbit("active:2026-08-27", 1001, 1)
count = redis_client.bitcount("active:2026-08-27")

适合签到、活跃用户、开关状态。

8. BitField：多个小整数状态压缩存储

场景概述： BitField 适合压缩存储多个小整数状态。

图形化展示：

text

score:user:1001:2026-08
| day1 4bit | day2 4bit | day3 4bit |
|     8     |     6     |     9     |

redis-cli 命令：

bash

redis-cli BITFIELD score:user:1001:2026-08 SET u4 0 8 SET u4 4 6 GET u4 0 GET u4 4

Python 示例：

python

redis_client.execute_command("BITFIELD", "score:user:1001:2026-08", "SET", "u4", 0, 8)

适合评分、等级、小整数压缩。

9. HyperLogLog：大规模 UV 去重估算

场景概述： HyperLogLog 适合大规模去重计数和 UV 估算。

图形化展示：

text

uv:video:6
1001, 1002, 1003, 1001, 1008, 1002
        └── HLL 估算唯一访问数

redis-cli 命令：

bash

redis-cli PFADD uv:video:6 1001 1002 1003
redis-cli PFCOUNT uv:video:6
redis-cli PFMERGE uv:site:range uv:day1 uv:day2

Python 示例：

python

redis_client.pfadd("uv:video:6", 1001, 1002, 1003)
uv = redis_client.pfcount("uv:video:6")

适合 UV 估算、访问人数统计。

10. GEO：附近的人 / 门店距离 / 地理位置查询

场景概述： GEO 适合附近位置查询和距离计算。

图形化展示：

text

geo:stores
store:1 ──► (116.397128, 39.916527)
store:2 ──► (116.410000, 39.920000)

redis-cli 命令：

bash

redis-cli GEOADD geo:stores 116.397128 39.916527 store:1
redis-cli GEOSEARCH geo:stores FROMLONLAT 116.4 39.9 BYRADIUS 5 km WITHDIST
redis-cli GEODIST geo:stores store:1 store:2 km

Python 示例：

python

redis_client.geoadd("geo:stores", {"store:1": (116.397128, 39.916527)})

适合附近门店、同城服务、地理围栏。

11. Pub/Sub：实时发布订阅 / 通知广播

场景概述： Pub/Sub 适合实时广播和轻量通知。

图形化展示：

text

PUBLISH config:changed
        ├── 订阅者 A
        ├── 订阅者 B
        └── 订阅者 C

redis-cli 命令：

bash

redis-cli SUBSCRIBE config:changed
redis-cli PUBLISH config:changed "video:max_workers"

Python 示例：

python

redis_client.publish("config:changed", "video:max_workers")

适合实时广播、配置变更通知。

12. RedisJSON：复杂 JSON 文档缓存

场景概述： RedisJSON 适合嵌套 JSON 文档和局部更新。

图形化展示：

text

json:task:6:config
{ model, ratio, frames[] }

redis-cli 命令：

bash

redis-cli JSON.SET json:task:6:config $ '{"model":"seedance-2.0","ratio":"16:9"}'
redis-cli JSON.GET json:task:6:config $

Python 示例：

python

redis_client.execute_command("JSON.SET", "json:task:6:config", "$", '{"model":"seedance-2.0"}')

适合嵌套 JSON 配置、局部更新。

13. Bloom Filter：存在性预判 / 防缓存穿透

场景概述： Bloom Filter 适合做存在性预判和缓存穿透防护。

图形化展示：

text

bf:goods:ids
已加入: goods:1001, goods:1002, goods:1003

redis-cli 命令：

bash

redis-cli BF.ADD bf:goods:ids goods:1001
redis-cli BF.EXISTS bf:goods:ids goods:1002

Python 示例：

python

redis_client.execute_command("BF.ADD", "bf:goods:ids", "goods:1001")
exists = redis_client.execute_command("BF.EXISTS", "bf:goods:ids", "goods:1002")

适合防缓存穿透、黑名单预判。

14. TimeSeries：监控指标 / 趋势数据

场景概述： TimeSeries 适合记录时间序列指标和趋势。

图形化展示：

text

ts:queue:task:length
1787796000000 -> 12
1787796060000 -> 18

redis-cli 命令：

bash

redis-cli TS.ADD ts:queue:task:length * 12
redis-cli TS.RANGE ts:queue:task:length - +

Python 示例：

python

redis_client.execute_command("TS.ADD", "ts:queue:task:length", int(time.time() * 1000), 12)

适合队列长度、QPS、耗时趋势。

15. RedisSearch：全文检索 / 二级索引 / 多条件筛选

场景概述： RedisSearch 适合全文检索、索引和组合筛选。

图形化展示：

text

idx:tasks
  ├── title 可搜
  ├── stage 可筛选
  └── user_id 可过滤

redis-cli 命令：

bash

redis-cli FT.CREATE idx:tasks ON HASH PREFIX 1 task:search: SCHEMA title TEXT stage TAG user_id NUMERIC
redis-cli FT.SEARCH idx:tasks "@stage:{IMAGE} 海边"

Python 示例：

python

redis_client.execute_command("FT.SEARCH", "idx:tasks", "@stage:{IMAGE} 海边")

适合全文检索和组合筛选。

16. Vector Search：向量相似度检索

场景概述： Vector Search 适合相似检索和语义搜索。

图形化展示：

text

video:6 -> [0.12, 0.88, 0.31]
video:7 -> [0.10, 0.86, 0.35]
查询向量 q -> 找最相似的视频

redis-cli 命令：

bash

redis-cli FT.CREATE idx:vec ON HASH PREFIX 1 video: SCHEMA embedding VECTOR HNSW 6 TYPE FLOAT32 DIM 3 DISTANCE_METRIC COSINE
redis-cli FT.SEARCH idx:vec '*=>[KNN 3 @embedding $vec]' PARAMS 2 vec "..." SORTBY __vector_score DIALECT 2

Python 示例：

python

redis_client.execute_command("FT.SEARCH", "idx:vec", "*=>[KNN 3 @embedding $vec]")

适合相似视频推荐、语义搜索、RAG 检索。

17. Cuckoo Filter：可删除的存在性判断

场景概述： Cuckoo Filter 适合可删除的去重和存在性判断。

图形化展示：

text

cf:task:ids
CF.ADD task:6
CF.EXISTS task:6
CF.DEL task:6

redis-cli 命令：

bash

redis-cli CF.ADD cf:task:ids task:6
redis-cli CF.EXISTS cf:task:ids task:6
redis-cli CF.DEL cf:task:ids task:6

Python 示例：

python

redis_client.execute_command("CF.ADD", "cf:task:ids", "task:6")

适合可删除去重、临时黑名单。

18. Count-Min Sketch：大规模频次估算

场景概述： Count-Min Sketch 适合大规模频次估算。

图形化展示：

text

cms:keywords
redis -> 12000
video -> 8500
worker -> 4300

redis-cli 命令：

bash

redis-cli CMS.INCRBY cms:keywords redis 1 video 1 worker 1
redis-cli CMS.QUERY cms:keywords redis video worker

Python 示例：

python

redis_client.execute_command("CMS.INCRBY", "cms:keywords", "redis", 1)

适合热点词、频次趋势。

19. TopK：高频元素近似统计

场景概述： TopK 适合近似统计高频元素。

图形化展示：

text

topk:videos
video:8, video:6, video:3, video:12, video:1

redis-cli 命令：

bash

redis-cli TOPK.RESERVE topk:videos 5
redis-cli TOPK.ADD topk:videos video:8 video:6 video:3
redis-cli TOPK.LIST topk:videos

Python 示例：

python

redis_client.execute_command("TOPK.ADD", "topk:videos", "video:8", "video:6")

适合热门视频、热门词榜单。

20. TDigest：分位数统计 / P95 / P99

场景概述： TDigest 适合分位数和延迟分布统计。

图形化展示：

text

td:video:duration
样本 -> P50 / P95 / P99

redis-cli 命令：

bash

redis-cli TDIGEST.CREATE td:video:duration
redis-cli TDIGEST.ADD td:video:duration 2.1 2.4 3.0 8.2 15.6 30.1
redis-cli TDIGEST.QUANTILE td:video:duration 0.95

Python 示例：

python

redis_client.execute_command("TDIGEST.ADD", "td:video:duration", 2.1, 2.4, 3.0)

适合延迟分布、耗时分位数。

15. 使用场景中函数参数含义速查

第十章中的示例代码涉及较多 Redis 命令和 Python 辅助函数。阅读这些案例时，重点不是只记住函数名，而是理解每个参数对应 Redis 中的 key、value、field、score、member、offset、过期时间或消费组语义。

String / 通用 key 操作

函数	示例	参数含义
`redis_client.set(name, value, ex=None, nx=False)`	`set("task:6:status", "RUNNING", ex=900)`	`name` 是 Redis key；`value` 是要保存的值；`ex` 是秒级过期时间；`nx=True` 表示 key 不存在时才写入，常用于分布式锁
`redis_client.get(name)`	`get("task:6:status")`	`name` 是要读取的 Redis key，返回该 key 当前保存的 value
`redis_client.incr(name, amount=1)`	`incr("video:6:view_count")`	`name` 是计数器 key；`amount` 是递增步长，默认加 1
`redis_client.expire(name, time)`	`expire(key, 60)`	`name` 是要设置过期时间的 key；`time` 是过期秒数
`redis_client.delete(*names)`	`delete(lock_key)`	`names` 是一个或多个要删除的 key。分布式锁中不建议直接删除，容易误删别人的锁
`redis_client.eval(script, numkeys, *keys_and_args)`	`eval(UNLOCK_SCRIPT, 1, lock_key, lock_token)`	`script` 是 Lua 脚本；`numkeys` 表示后面有几个参数属于 Redis key；前 `numkeys` 个参数进入 Lua 的 `KEYS`；剩余参数进入 Lua 的 `ARGV`

Hash 操作

函数	示例	参数含义
`redis_client.hset(name, key=None, value=None, mapping=None)`	`hset("task:6:progress", "stage", "IMAGE")`	`name` 是 Hash key；`key` 是字段名 field；`value` 是字段值；`mapping` 可一次写入多个 field-value
`redis_client.hget(name, key)`	`hget(f"order:{oid}", "status")`	`name` 是 Hash key；`key` 是要读取的字段名
`redis_client.hgetall(name)`	`hgetall("task:6:progress")`	`name` 是 Hash key，返回这个 Hash 下所有 field-value
`redis_client.hincrby(name, key, amount=1)`	`hincrby("cart:user:1001", "goods:2001", 1)`	`name` 是 Hash key；`key` 是字段名；`amount` 是该字段的整数增量

List 操作

函数	示例	参数含义
`redis_client.lpush(name, *values)`	`lpush("task:queue", 6)`	`name` 是 List key；`values` 是要从左侧压入的一个或多个元素
`redis_client.brpop(keys, timeout=0)`	`brpop("task:queue", timeout=5)`	`keys` 是一个 key 或多个 key；`timeout` 是最大阻塞等待秒数，`0` 表示一直等待
`redis_client.lrange(name, start, end)`	`lrange("user:1001:feeds", 0, 19)`	`name` 是 List key；`start` 是起始下标；`end` 是结束下标，包含该位置，`-1` 表示最后一个元素
`redis_client.ltrim(name, start, end)`	`ltrim("user:1001:feeds", 0, 19)`	`name` 是 List key；只保留 `start` 到 `end` 范围内的元素，常用于限制最新列表长度
`redis_client.lpop(name)`	`lpop("undo:user:1001")`	`name` 是 List key，从左侧弹出一个元素
`redis_client.llen(name)`	`llen("task:queue")`	`name` 是 List key，返回列表长度

Set 操作

函数	示例	参数含义
`redis_client.sadd(name, *values)`	`sadd("video:6:liked_users", 1001)`	`name` 是 Set key；`values` 是要加入集合的成员，重复成员会自动去重
`redis_client.sismember(name, value)`	`sismember("video:6:liked_users", 1001)`	`name` 是 Set key；`value` 是要判断是否存在的成员
`redis_client.scard(name)`	`scard("video:6:liked_users")`	`name` 是 Set key，返回集合成员数量
`redis_client.sinter(keys, *args)`	`sinter("user:1:follows", "user:2:follows")`	参数是多个 Set key，返回这些集合的交集
`redis_client.srandmember(name, number=None)`	`srandmember("lottery:202608:users", 2)`	`name` 是 Set key；`number` 是随机返回的成员数量

ZSet 操作

函数	示例	参数含义
`redis_client.zadd(name, mapping)`	`zadd("delay:order:cancel", {order_id: expire_at})`	`name` 是 ZSet key；`mapping` 是 `{member: score}` 字典，member 是成员，score 是排序分数
`redis_client.zincrby(name, amount, value)`	`zincrby("rank:video:hot", 1, "video:6")`	`name` 是 ZSet key；`amount` 是分数增量；`value` 是要增加分数的成员
`redis_client.zrevrange(name, start, end, withscores=False)`	`zrevrange("rank:video:hot", 0, 9, withscores=True)`	`name` 是 ZSet key；`start/end` 是排名范围；`withscores=True` 表示同时返回 score
`redis_client.zrevrank(name, value)`	`zrevrank("rank:video:hot", "video:6")`	`name` 是 ZSet key；`value` 是成员，返回从高到低排序时的名次
`redis_client.zrangebyscore(name, min, max, start=None, num=None)`	`zrangebyscore("delay:tasks", 0, now, start=0, num=10)`	`name` 是 ZSet key；`min/max` 是 score 范围；`start` 是偏移量；`num` 是最多返回数量
`redis_client.zrem(name, *values)`	`zrem("delay:order:cancel", oid)`	`name` 是 ZSet key；`values` 是要删除的成员。多 Worker 场景中，谁 `ZREM` 成功谁获得处理权

Stream 操作

函数	示例	参数含义
`redis_client.xadd(name, fields, id="*")`	`xadd("stream:task", {"task_id": "6"})`	`name` 是 Stream key；`fields` 是消息字段字典；`id="*"` 表示 Redis 自动生成消息 ID
`redis_client.xgroup_create(name, groupname, id="0", mkstream=False)`	`xgroup_create("stream:task", "video-workers", id="0", mkstream=True)`	`name` 是 Stream key；`groupname` 是消费组名；`id` 表示从哪个消息 ID 开始消费；`mkstream=True` 表示 Stream 不存在时自动创建
`redis_client.xreadgroup(groupname, consumername, streams, count=None, block=None)`	`xreadgroup("video-workers", "worker-1", {"stream:task": ">"}, count=10, block=5000)`	`groupname` 是消费组；`consumername` 是消费者名；`streams` 指定要读的 Stream 和起始 ID；`count` 是最多读取条数；`block` 是阻塞等待毫秒数
`redis_client.xack(name, groupname, *ids)`	`xack("stream:task", "video-workers", message_id)`	`name` 是 Stream key；`groupname` 是消费组；`ids` 是要确认完成的消息 ID
`redis_client.xrange(name, min="-", max="+")`	`xrange("stream:video:event", min="-", max="+")`	`name` 是 Stream key；`min/max` 是消息 ID 范围，`-` 表示最早，`+` 表示最新

Bitmap / BitField 操作

函数	示例	参数含义
`redis_client.setbit(name, offset, value)`	`setbit("active:2026-08-27", 1001, 1)`	`name` 是 Bitmap key；`offset` 是 bit 下标；`value` 只能是 `0` 或 `1`
`redis_client.getbit(name, offset)`	`getbit("signin:2026-08:user:1001", day_index)`	`name` 是 Bitmap key；`offset` 是要读取的 bit 下标
`redis_client.bitcount(key)`	`bitcount("active:2026-08-27")`	`key` 是 Bitmap key，返回值为 1 的 bit 数量
`redis_client.bitop(operation, dest, *keys)`	`bitop("AND", "active:and", "active:day1", "active:day2")`	`operation` 是位运算类型，如 `AND/OR/XOR/NOT`；`dest` 是结果 key；`keys` 是参与运算的源 key
`redis_client.bitfield(key, *args)`	`bitfield("score:user:1001:2026-08", "SET", "u4", 26 * 4, 9)`	`key` 是底层字符串 key；`args` 是 BitField 子命令序列，例如 `SET/GET`、整数类型 `u4`、bit 偏移量和写入值

HyperLogLog / GEO / Pub/Sub 操作

函数	示例	参数含义
`redis_client.pfadd(name, *values)`	`pfadd("uv:video:6", 1001, 1002, 1003)`	`name` 是 HyperLogLog key；`values` 是要参与去重估算的元素
`redis_client.pfcount(*sources)`	`pfcount("uv:video:6")`	`sources` 是一个或多个 HyperLogLog key，返回不重复数量估算值
`redis_client.pfmerge(dest, *sources)`	`pfmerge("uv:site:range", "uv:day1", "uv:day2")`	`dest` 是合并后的目标 key；`sources` 是要合并的多个 HyperLogLog key
`redis_client.geoadd(name, values)`	`geoadd("geo:stores", [116.397128, 39.916527, "store:1"])`	`name` 是 GEO key；`values` 按经度、纬度、成员名传入
`redis_client.geosearch(name, longitude=None, latitude=None, member=None, radius=None, unit="m")`	`geosearch("geo:stores", longitude=116.4, latitude=39.9, radius=5, unit="km")`	`name` 是 GEO key；可用经纬度或已有成员作为中心点；`radius` 是半径；`unit` 是距离单位
`redis_client.geodist(name, place1, place2, unit=None)`	`geodist("geo:stores", "store:1", "store:2", unit="km")`	`name` 是 GEO key；`place1/place2` 是两个成员；`unit` 是返回距离单位
`redis_client.publish(channel, message)`	`publish("config:changed", "video:max_workers")`	`channel` 是频道名；`message` 是要发布的消息内容
`redis_client.pubsub()`	`pubsub()`	创建一个 Pub/Sub 订阅对象，后续可通过该对象订阅频道并读取消息

Redis Stack / 模块命令

函数	示例	参数含义
`redis_client.execute_command("JSON.SET", key, path, value)`	`execute_command("JSON.SET", "json:task:6:config", ".", json_text)`	第 1 个参数是模块命令名；`key` 是 JSON key；`path` 是 JSON 路径；`value` 是要写入的 JSON 字符串
`redis_client.execute_command("JSON.GET", key, path)`	`execute_command("JSON.GET", "json:task:6:config", ".")`	`key` 是 JSON key；`path` 是要读取的 JSON 路径
`redis_client.execute_command("BF.ADD", key, item)`	`execute_command("BF.ADD", "bf:goods:ids", "goods:2001")`	`key` 是 Bloom Filter key；`item` 是要加入过滤器的元素
`redis_client.execute_command("BF.EXISTS", key, item)`	`execute_command("BF.EXISTS", "bf:goods:ids", "goods:9999")`	`key` 是 Bloom Filter key；`item` 是要判断是否可能存在的元素
`redis_client.execute_command("TS.ADD", key, timestamp, value)`	`execute_command("TS.ADD", "ts:queue:task:length", ts, queue_length)`	`key` 是时间序列 key；`timestamp` 是毫秒时间戳；`value` 是指标值
`redis_client.execute_command("TS.RANGE", key, from_ts, to_ts, ...)`	`execute_command("TS.RANGE", "ts:queue:task:length", "-", "+")`	`key` 是时间序列 key；`from_ts/to_ts` 是时间范围，`-` 到 `+` 表示全量范围
`redis_client.execute_command("FT.CREATE", index, ...)`	`execute_command("FT.CREATE", "idx:task", "ON", "HASH", ...)`	`index` 是索引名；后续参数定义索引数据来源、前缀、字段类型和排序规则
`redis_client.execute_command("FT.SEARCH", index, query, ...)`	`execute_command("FT.SEARCH", "idx:task", "@stage:{IMAGE}")`	`index` 是索引名；`query` 是查询表达式；后续参数可指定排序、分页、返回字段等
`redis_client.execute_command("CF.ADD", key, item)`	`execute_command("CF.ADD", "cf:task:ids", "task:6")`	`key` 是 Cuckoo Filter key；`item` 是要加入的元素
`redis_client.execute_command("CF.EXISTS", key, item)`	`execute_command("CF.EXISTS", "cf:task:ids", "task:6")`	`key` 是 Cuckoo Filter key；`item` 是要判断是否可能存在的元素
`redis_client.execute_command("CF.DEL", key, item)`	`execute_command("CF.DEL", "cf:task:ids", "task:6")`	`key` 是 Cuckoo Filter key；`item` 是要从过滤器中删除的元素
`redis_client.execute_command("CMS.INCRBY", key, item, increment)`	`execute_command("CMS.INCRBY", "cms:keywords", "redis", 3)`	`key` 是 Count-Min Sketch key；`item` 是统计对象；`increment` 是增加次数
`redis_client.execute_command("CMS.QUERY", key, *items)`	`execute_command("CMS.QUERY", "cms:keywords", "video", "redis")`	`key` 是 Count-Min Sketch key；`items` 是要查询频次的多个元素
`redis_client.execute_command("TOPK.RESERVE", key, k, width, depth, decay)`	`execute_command("TOPK.RESERVE", "topk:videos", 10, 2000, 7, 0.9)`	`key` 是 TopK key；`k` 是保留前多少名；`width/depth/decay` 控制近似统计的空间、精度和衰减策略
`redis_client.execute_command("TOPK.ADD", key, *items)`	`execute_command("TOPK.ADD", "topk:videos", "video:1")`	`key` 是 TopK key；`items` 是要计入高频统计的元素
`redis_client.execute_command("TOPK.LIST", key)`	`execute_command("TOPK.LIST", "topk:videos")`	`key` 是 TopK key，返回当前近似 TopK 列表
`redis_client.execute_command("TDIGEST.CREATE", key)`	`execute_command("TDIGEST.CREATE", "td:video:duration")`	`key` 是 TDigest key，用于保存分位数统计结构
`redis_client.execute_command("TDIGEST.ADD", key, value)`	`execute_command("TDIGEST.ADD", "td:video:duration", 12.5)`	`key` 是 TDigest key；`value` 是要加入统计的数值样本
`redis_client.execute_command("TDIGEST.QUANTILE", key, quantile)`	`execute_command("TDIGEST.QUANTILE", "td:video:duration", 0.95)`	`key` 是 TDigest key；`quantile` 是分位点，例如 `0.95` 表示 P95

Python 辅助函数

函数	示例	参数含义
`time.time()`	`int(time.time()) + 300`	不需要参数，返回当前 Unix 时间戳，单位是秒，常用于计算延迟任务执行时间
`int(value)`	`int(time.time())`	`value` 是要转换成整数的值，常用于把时间戳或 Redis 返回的字符串任务 ID 转成整数
`str(value)`	`str(uuid.uuid4())`	`value` 是要转换成字符串的对象，分布式锁中用于把 UUID 转成可保存的 token 字符串
`uuid.uuid4()`	`uuid.uuid4()`	不需要参数，生成随机 UUID，适合作为分布式锁的唯一 `lock_token`

16. 本项目中的推荐使用方式

结合当前视频任务系统，可以这样选择：

业务数据	推荐结构	原因
待处理任务队列	List	`LPUSH + BRPOP` 简单、高效，适合 Worker 阻塞消费
任务执行进度	Hash	可以单独更新 `stage`、`percent`、`message` 等字段
任务状态缓存	String / Hash	简单状态用 String，多字段状态用 Hash
用户提交频率限制	String	使用 `INCR + EXPIRE` 实现限流计数
视频播放量	String	使用 `INCR` 做计数器
视频点赞用户	Set	自动去重，方便判断用户是否点过赞
视频热度排行	ZSet	用热度分或播放量作为 score 排序
延迟重试任务	ZSet / Stream	简单延迟用 ZSet，需要 ACK 和恢复用 Stream
可靠事件日志	Stream	如果后续需要 ACK、重试、消费组，可从 List 升级到 Stream
在线配置广播	Pub/Sub	通知在线服务刷新配置，消息不需要持久化
队列积压趋势	TimeSeries / 普通指标系统	如果需要趋势图和聚合，可以记录时间序列
防止无效 ID 穿透数据库	Bloom Filter	大规模 ID 存在性预判，减少无效 DB 查询

17. 选型口诀表

需求关键词	优先选择	典型命令	示例场景	注意点
一个 key 存一个完整值	String	`SET`、`GET`、`EXPIRE`	验证码、登录 token、任务状态缓存	适合整体读写，不适合频繁局部更新对象字段
原子计数、自增自减	String	`INCR`、`DECR`、`INCRBY`	播放量、点赞数、接口调用次数	计数器可以配合 `EXPIRE` 做限流窗口
一个对象有多个字段	Hash	`HSET`、`HGET`、`HGETALL`	用户资料、任务进度、视频元信息	适合字段级更新，不适合按字段值排序
简单队列、阻塞消费	List	`LPUSH`、`BRPOP`	视频任务队列、异步任务分发	适合简单队列，不支持 ACK 和消费组
最新列表、操作历史	List	`LPUSH`、`LRANGE`、`LTRIM`	最新动态、最近操作、撤销栈	要配合 `LTRIM` 控制列表长度
去重、判断是否存在	Set	`SADD`、`SISMEMBER`、`SCARD`	点赞用户、抽奖参与用户、黑名单	无序集合，不适合按时间或分数排序
集合关系运算	Set	`SINTER`、`SUNION`、`SDIFF`	共同关注、共同标签、权限对比	数据量很大时要注意集合运算成本
排名、权重、分数排序	ZSet	`ZADD`、`ZREVRANGE`、`ZREVRANK`	视频热榜、用户积分榜、优先级队列	score 是排序核心，更新频繁时要评估写入成本
时间戳排序、延迟执行	ZSet	`ZADD`、`ZRANGEBYSCORE`、`ZREM`	延迟任务、最近访问、定时重试	多 Worker 取任务时要用 `ZREM` 或 Lua 保证不重复执行
可靠消息、消费组、ACK	Stream	`XADD`、`XREADGROUP`、`XACK`	可靠任务队列、订单事件、视频生成事件流	比 List 复杂，但支持 pending、ACK 和故障恢复
大量是/否状态	Bitmap	`SETBIT`、`GETBIT`、`BITCOUNT`	用户签到、日活统计、功能灰度	offset 要可控，稀疏大 ID 可能浪费空间
多个小整数状态压缩	BitField	`BITFIELD`	每日评分、设备状态档位、小型状态机	可读性较差，适合内存极度敏感场景
只统计不重复数量，不关心成员	HyperLogLog	`PFADD`、`PFCOUNT`、`PFMERGE`	网站 UV、视频 UV、活动访问人数	结果是估算值，不能列出具体成员
经纬度附近查询	GEO	`GEOADD`、`GEOSEARCH`、`GEODIST`	附近门店、附近设备、同城资源	适合基础附近查询，不替代专业 GIS
在线实时广播	Pub/Sub	`PUBLISH`、`SUBSCRIBE`	配置变更通知、WebSocket 网关广播	消息不持久化，订阅者离线会丢消息
复杂 JSON 局部更新	RedisJSON	`JSON.SET`、`JSON.GET`	复杂配置、用户画像、商品详情缓存	需要 RedisJSON 模块支持，标准 Redis 默认没有
大规模存在性预判	Bloom Filter	`BF.ADD`、`BF.EXISTS`	防缓存穿透、URL 去重、风险名单预判	有误判率，不能替代精确集合
指标趋势和时间范围查询	TimeSeries	`TS.ADD`、`TS.RANGE`	队列长度趋势、接口 QPS、任务耗时监控	需要 RedisTimeSeries 模块支持，也可交给专业监控系统

第十一章：Python 可运行案例

Q28：如何让文档里的 Redis 示例连接到本机 6379？

答：

当前示例统一把 Redis 连接信息写到 redis/python示例/.env，并通过 python-dotenv 自动加载。默认连接本机 127.0.0.1:6379，密码从 .env 中的 REDIS_PASSWORD 读取。

运行前先安装依赖：

bash

python3 -m pip install -r /Users/binny/workspace-sdd/aiNotes/g; srequirements.txt

.env 文件内容如下：

bash

REDIS_HOST=127.0.0.1
REDIS_PORT=6379
REDIS_DB=0
REDIS_PASSWORD='你的 Redis 密码'

所有示例共享同一个连接入口：

python

import os
from pathlib import Path

from dotenv import load_dotenv
from redis import Redis

ENV_PATH = Path(__file__).resolve().parent / '.env'
load_dotenv(ENV_PATH)


def get_client(decode_responses: bool = True) -> Redis:
    password = os.environ.get('REDIS_PASSWORD')
    if not password:
        raise RuntimeError(f'请先在 {ENV_PATH} 中配置 REDIS_PASSWORD')
    return Redis(
        host=os.environ.get('REDIS_HOST', '127.0.0.1'),
        port=int(os.environ.get('REDIS_PORT', '6379')),
        password=password,
        db=int(os.environ.get('REDIS_DB', '0')),
        decode_responses=decode_responses,
        socket_connect_timeout=3,
        socket_timeout=5,
        health_check_interval=30,
    )

对应文件：[redis_connection.py](file:///Users/binny/workspace-sdd/aiNotes/redis/python示例/redis_connection.py)

这个写法符合 Redis 生产实践中的几条基本原则：

.env 由 python-dotenv 加载，运行脚本时无需再手动 export；
.gitignore 已忽略 .env，避免把真实密码提交到仓库；
给连接设置 socket_connect_timeout 和 socket_timeout，避免网络异常时无限等待；
使用 decode_responses=True，让命令返回字符串，便于教学演示；
所有示例使用 demo:* 前缀，方便清理，避免误删真实业务 key。

如果运行时出现 invalid password，说明 Python 已经连到 Redis 服务，但 .env 中的 REDIS_PASSWORD 与当前 REDIS_HOST:REDIS_PORT 指向的 Redis 实例不匹配，需要确认本机 6379 是否就是目标 Redis，以及密码是否为该实例的实际认证密码。

Q29：String 如何演示缓存、计数器和分布式锁？

答：

String 适合存简单状态、缓存值、计数器和轻量锁。示例会真实执行 SET EX、GET、INCR、SET NX EX。

运行方式：

bash

python3 /Users/binny/workspace-sdd/aiNotes/redis/python示例/string_cache_and_lock.py

核心代码：

python

from redis_connection import flush_demo_keys, run


def demo(client):
    flush_demo_keys(client, 'demo:string:*', 'demo:lock:*')
    client.set('demo:string:task:6:status', 'RUNNING', ex=900)
    print(client.get('demo:string:task:6:status'))
    print(client.incr('demo:string:video:6:view_count'))
    locked = client.set('demo:lock:task:6', 'abc123', nx=True, ex=300)
    print(locked)
    print(client.get('demo:lock:task:6'))

对应文件：[string_cache_and_lock.py](file:///Users/binny/workspace-sdd/aiNotes/redis/python示例/string_cache_and_lock.py)

Q30：Hash 如何演示任务进度？

答：

Hash 适合同一个实体有多个字段的场景，比如任务进度、用户资料、订单状态。它比把整个对象序列化成 String 更适合局部字段更新。

运行方式：

bash

python3 /Users/binny/workspace-sdd/aiNotes/redis/python示例/hash_progress.py

核心代码：

python

from redis_connection import flush_demo_keys, run


def demo(client):
    key = 'demo:hash:task:6:progress'
    flush_demo_keys(client, key)
    client.hset(key, mapping={'stage': 'IMAGE', 'percent': '45', 'message': '正在生成分镜图'})
    print(client.hget(key, 'stage'))
    print(client.hgetall(key))
    client.hincrby(key, 'percent', 10)
    print(client.hgetall(key))

对应文件：[hash_progress.py](file:///Users/binny/workspace-sdd/aiNotes/redis/python示例/hash_progress.py)

Q31：List 与 BRPOP 如何演示任务队列？

答：

List 的典型用法是队列。LPUSH 负责投递任务，BRPOP 负责阻塞等待任务，适合文档前面讲到的 Worker 模型。

运行方式：

bash

python3 /Users/binny/workspace-sdd/aiNotes/redis/python示例/list_queue_and_latest.py
python3 /Users/binny/workspace-sdd/aiNotes/redis/python示例/brpop_worker_demo.py

核心代码：

python

import threading
import time

from redis_connection import flush_demo_keys, get_client, run


def demo(client):
    queue_key = 'demo:brpop:task:queue'
    flush_demo_keys(client, queue_key)
    result_holder = []

    def worker_loop():
        worker_client = get_client()
        result_holder.append(worker_client.brpop(queue_key, timeout=5))

    thread = threading.Thread(target=worker_loop)
    thread.start()
    time.sleep(0.5)
    client.lpush(queue_key, 'task-6')
    thread.join(timeout=6)
    print(result_holder)

对应文件：

[list_queue_and_latest.py](file:///Users/binny/workspace-sdd/aiNotes/redis/python示例/list_queue_and_latest.py)
[brpop_worker_demo.py](file:///Users/binny/workspace-sdd/aiNotes/redis/python示例/brpop_worker_demo.py)

Q32：Set、ZSet、Bitmap、HyperLogLog、Geo 分别适合演示什么？

答：

这些结构适合业务侧高频场景：

数据结构	适合场景	示例文件
Set	去重、点赞、共同关注	[set_dedup_and_relation.py](file:///Users/binny/workspace-sdd/aiNotes/redis/python示例/set_dedup_and_relation.py)
ZSet	排行榜、延迟任务	[zset_ranking_and_delay.py](file:///Users/binny/workspace-sdd/aiNotes/redis/python示例/zset_ranking_and_delay.py)
Bitmap	签到、布尔状态统计	[bitmap_signin.py](file:///Users/binny/workspace-sdd/aiNotes/redis/python示例/bitmap_signin.py)
HyperLogLog	UV 去重估算	[hll_uv.py](file:///Users/binny/workspace-sdd/aiNotes/redis/python示例/hll_uv.py)
Geo	附近门店、地理位置检索	[geo_nearby_search.py](file:///Users/binny/workspace-sdd/aiNotes/redis/python示例/geo_nearby_search.py)

运行方式示例：

bash

python3 /Users/binny/workspace-sdd/aiNotes/redis/python示例/set_dedup_and_relation.py
python3 /Users/binny/workspace-sdd/aiNotes/redis/python示例/zset_ranking_and_delay.py
python3 /Users/binny/workspace-sdd/aiNotes/redis/python示例/bitmap_signin.py
python3 /Users/binny/workspace-sdd/aiNotes/redis/python示例/hll_uv.py
python3 /Users/binny/workspace-sdd/aiNotes/redis/python示例/geo_nearby_search.py

Q33：Stream 和 Pub/Sub 的可运行案例有什么区别？

答：

Stream 适合可靠消息队列，消息会保存在流里，可以配合消费者组、ACK、Pending List 使用；Pub/Sub 适合在线广播，不保存历史消息，订阅者离线时会错过消息。

运行方式：

bash

python3 /Users/binny/workspace-sdd/aiNotes/redis/python示例/stream_consumer_group.py
python3 /Users/binny/workspace-sdd/aiNotes/redis/python示例/pubsub_broadcast.py

对应文件：

[stream_consumer_group.py](file:///Users/binny/workspace-sdd/aiNotes/redis/python示例/stream_consumer_group.py)
[pubsub_broadcast.py](file:///Users/binny/workspace-sdd/aiNotes/redis/python示例/pubsub_broadcast.py)

Q34：连接池如何演示？

答：

Redis 客户端通常不会每次命令都新建 TCP 连接，而是复用连接池。示例中使用 BlockingConnectionPool，限制最大连接数，并设置获取连接的等待超时时间。

运行方式：

bash

python3 /Users/binny/workspace-sdd/aiNotes/redis/python示例/connection_pool_demo.py

核心代码：

python

from redis import BlockingConnectionPool, Redis

from redis_connection import run


def demo(client):
    source_pool = client.connection_pool
    kwargs = source_pool.connection_kwargs.copy()
    pool = BlockingConnectionPool(max_connections=10, timeout=3, **kwargs)
    pooled_client = Redis(connection_pool=pool, decode_responses=True)
    print(pooled_client.ping())

对应文件：[connection_pool_demo.py](file:///Users/binny/workspace-sdd/aiNotes/redis/python示例/connection_pool_demo.py)

Q35：哪些案例依赖 Redis Stack 或额外模块？

答：

以下案例已经写成“可运行但可降级”的形式：如果你的 Redis 只安装了基础服务，它们会提示模块不可用；如果安装了 Redis Stack 或对应模块，就会真实执行命令。

示例	依赖能力	示例文件
RedisJSON	RedisJSON 模块	[redisjson_document.py](file:///Users/binny/workspace-sdd/aiNotes/redis/python示例/redisjson_document.py)
Bloom / Cuckoo / CMS / TopK / TDigest	RedisBloom 模块	[probabilistic_filters.py](file:///Users/binny/workspace-sdd/aiNotes/redis/python示例/probabilistic_filters.py)
FT.SEARCH / TimeSeries	RediSearch、RedisTimeSeries	[search_vector_timeseries.py](file:///Users/binny/workspace-sdd/aiNotes/redis/python示例/search_vector_timeseries.py)

运行方式：

bash

python3 /Users/binny/workspace-sdd/aiNotes/redis/python示例/redisjson_document.py
python3 /Users/binny/workspace-sdd/aiNotes/redis/python示例/probabilistic_filters.py
python3 /Users/binny/workspace-sdd/aiNotes/redis/python示例/search_vector_timeseries.py

这里要注意：Redis Query Engine 查询建议显式使用 DIALECT 2，索引只建实际会查询的字段；向量检索和 RAG 场景要根据数据规模选择 HNSW 或 FLAT，不要盲目套用同一种索引算法。

Q36：如何一次性运行所有基础示例？

答：

可以用下面的命令逐个运行。它会自动读取 redis/python示例/.env 中的连接配置：

bash

for file in \
  string_cache_and_lock.py \
  hash_progress.py \
  list_queue_and_latest.py \
  set_dedup_and_relation.py \
  zset_ranking_and_delay.py \
  stream_consumer_group.py \
  bitmap_signin.py \
  hll_uv.py \
  geo_nearby_search.py \
  connection_pool_demo.py \
  brpop_worker_demo.py \
  atomicity_demo.py
  do python3 "/Users/binny/workspace-sdd/aiNotes/redis/python示例/$file"
done

所有示例都使用 demo:* 前缀；如果需要清理示例数据，可以在 redis-cli 中执行：

bash

redis-cli -h 127.0.0.1 -p 6379 -a '你的 Redis 密码' --scan --pattern 'demo:*' | xargs redis-cli -h 127.0.0.1 -p 6379 -a '你的 Redis 密码' DEL

速查表：六篇文档核心知识点

序号	文档	知识点	一句话总结
1	BRPOP 深度解析	BRPOP vs RPOP	空队列时 BRPOP 阻塞等待，RPOP 立即返回 nil
2	BRPOP 深度解析	多 key 优先级	按声明顺序从左到右检查，左边有就不查右边
3	BRPOP 深度解析	timeout=5	每 5 秒回到循环，能响应 Ctrl+C
4	BRPOP 深度解析	返回值	`tuple(队列名, 元素值)` 或 `None`
5	BRPOP 深度解析	LPUSH 返回值	列表长度，不是推入的值
6	BRPOP 深度解析	FIFO 队列	LPUSH + BRPOP = 左推右弹 = 先进先出
7	单线程与多客户端交互	多进程 → 单线程	N 个进程的命令在 Redis 主线程串行执行
8	单线程与多客户端交互	等待链表节点	每个节点 = 一个 TCP 连接 = 一个 `client` 结构体
9	单线程与多客户端交互	BRPOP 不阻塞主线程	服务端加入等待链表后立即处理下一条命令
10	单线程与多客户端交互	连接池	每个进程独立，进程内复用，进程间不共享
11	单线程与多客户端交互	fork 陷阱	fork 复制 fd，两个进程操作同一个 socket 会错乱
12	单线程与多客户端交互	Redis 6.0 多线程	只加速网络 I/O，命令执行仍单线程
13	原子性保证机制	四层保证	单线程 → 命令原子 → 等待链表 FIFO → 事件循环串行
14	原子性保证机制	LPUSH 唤醒	插入元素 → 检查等待链表 → 发给第一个等待者 → break
15	原子性保证机制	批量 LPUSH	3 个元素会逐轮唤醒 3 个 Worker
16	原子性保证机制	超时原子性	超时和 LPUSH 不可能同时执行，Redis 会按事件循环顺序串行处理
17	原子性保证机制	数据直达	LPUSH 的元素不一定先进列表再弹出，可直接发给正在 BRPOP 的客户端
18	原子性保证机制	break 语句	保证一次只唤醒一个 Worker，不会重复消费同一个元素
19	连接池详解	没有连接池时	每次操作新建 TCP，包含建连、命令执行、关闭连接成本
20	连接池详解	连接池核心	借出-归还模式，不 close，复用已有 TCP 连接
21	连接池详解	单线程 vs 多线程	单线程通常只用 1 个连接，多线程会按并发数借出多个连接
22	连接池详解	max_connections	连接数量上限保护，不是预创建连接数
23	连接池详解	health_check_interval	连接健康检查配置，和连接数量没有必然关系
24	连接池详解	socket_timeout	单命令读响应超时兜底，使用 BRPOP 时必须大于 BRPOP timeout
25	连接池详解	三种复用 TCP 的方式	纯 socket、单 Connection、连接池都能复用 TCP，只是封装层级不同
26	连接池详解	连接池真正附加价值	自动重连、线程安全、统一配置应用、并发连接上限保护
27	推拉模型与源码解析	BRPOP 本质	延迟响应（Delayed Response），不是纯推也不是纯拉
28	推拉模型与源码解析	LPUSH 唤醒	服务端主动 `addReply` 写入等待客户端输出缓冲区
29	推拉模型与源码解析	零额外往返	LPUSH 执行中完成唤醒，不需要客户端再次发送请求
30	推拉模型与源码解析	TCP 模型	BRPOP 是请求-响应模型的延迟返回版本，不是服务端随意推送
31	推拉模型与源码解析	唤醒触发	`signalKeyAsReady` → `handleClientsBlockedOnKeys`
32	推拉模型与源码解析	关键源码	`addReplyArrayLen` + `addReplyBulk` 写入 client 缓冲区
33	Worker 编排器深度解析	任务状态机	QUEUED → SCRIPT → IMAGE → TTS → VIDEO → PENDING_REVIEW → APPROVED / REJECTED
34	Worker 编排器深度解析	BRPOP 消费循环	5 秒超时 + Ctrl+C 响应 + 异常不中断循环
35	Worker 编排器深度解析	Seedance 分镜并行	`ThreadPoolExecutor(max_workers=3)` 并行调用 Seedance API
36	Worker 编排器深度解析	进度写入	`hset` 更新 Redis Hash 字段：`task:{id}:progress`
37	Worker 编排器深度解析	超时控制	`ARK_VIDEO_MAX_CONCURRENCY` 全局信号量限制外部 API 并发
38	Worker 编排器深度解析	任务超时	900 秒任务级超时 + 10 秒轮询间隔

参考文献与扩展阅读

文档	路径	主要内容
BRPOP 深度解析	`Redis-BRPOP深度解析.md`	BRPOP 语法、阻塞机制、多 Worker 行为、注意事项
单线程与多客户端交互	`Redis单线程与多客户端交互.md`	多进程客户端如何与单线程服务端交互
原子性保证机制	`Redis原子性保证机制.md`	单线程模型、等待链表源码级分析、命令原子性
连接池详解	`Redis连接池详解.md`	连接池内部机制、单/多线程对比、超时参数配置建议
推拉模型与源码解析	`Redis推拉模型与源码解析.md`	BRPOP 唤醒机制、延迟响应、源码级流程
Worker 编排器深度解析	`Worker编排器深度解析.md`	Worker 启动、消费循环、Seedance 分镜并行、异常处理

Redis 技术问答：队列、连接与并发专题总集 ​

合并来源 ​

术语说明：文档里的 Worker 默认代表独立执行单元 ​

第一章：BRPOP 基础语法 ​

本章导读 ​

Q1：BRPOP 和 RPOP 的核心区别是什么？为什么本项目用 BRPOP？ ​

RPOP 轮询方式 ​

BRPOP 阻塞等待方式 ​

BRPOP 不会阻塞 Redis 服务端主线程 ​

本章小结 ​

Q2：brpop(["task:queue:vip", "task:queue:normal"], timeout=5) 的优先级规则是什么？如果 VIP 队列持续有任务，普通队列会怎样？ ​

Q3：redis_client.lpush("task:queue", 101) 的返回值是什么？为什么不是 101？ ​

Q4：为什么 Worker 取出 task_id 后需要 int(task_id) 转换？ ​

第二章：阻塞机制与连接管理 ​

本章导读 ​

Q5：BRPOP 阻塞期间，Redis 服务端的主线程会卡住吗？其他客户端的命令能执行吗？ ​

Q6：一个 Worker 进程中，BRPOP 阻塞期间还能执行 hset 写进度吗？如果能，用的是同一个 TCP 连接吗？ ​

1. 单线程下不可能并行（常见误区） ​

2. 本项目实际是顺序执行（不是并行） ​

3. 真正的并行需要多线程 ​

4. 连接池状态对比 ​

5. 如果 max_connections=1 会怎样 ​

Q7：timeout=0（永久阻塞）和 timeout=5 有什么实际区别？为什么不推荐 timeout=0？ ​

第三章：单线程模型与原子性 ​

本章导读 ​

Q8：Redis 是单线程的，为什么还能支撑 10 万 QPS？单线程不会成为瓶颈吗？ ​

Q9：10 个 Worker 同时 BRPOP task:queue 5，这时有人 LPUSH task:queue 6，为什么不会有两个 Worker 同时拿到 "6"？ ​

Q10：LPUSH 唤醒 BRPOP 时，数据是先存入列表再弹出，还是直接传递？ ​

Q10.1：BRPOP 的"推送"本质是什么？是延迟响应还是真正的推送？ ​

1. 三种模型的严格区分 ​

2. 延迟响应的完整流程 ​

3. 延迟响应的三个关键特征 ​

4. "推送"说法的精确含义 ​

5. 与全双工推送的对比 ​

6. 为什么说是"服务端主动推送" ​

7. 总结 ​

Q11：假设 Worker-1 和 Worker-2 几乎同时发送 BRPOP，API 也几乎同时发送 LPUSH，三种命令到达 Redis 的顺序不同，结果会怎样？ ​

第四章：多 Worker 与连接池 ​

本章导读 ​

Q12：3 个 Worker 进程启动后，各自有独立的 redis_client 吗？Redis 服务端看到的是 3 个连接还是 1 个？ ​

Q13：Redis 等待链表里存的是什么？是"Worker 进程"吗？ ​

⚠️ 关键澄清：等待链表节点数 ≠ 连接总数 ​

场景对比：一个进程 + 3 个连接，等待链表节点数不同 ​

速查表 ​

用 CLIENT LIST 验证 ​

本章小结 ​

补充：每条 TCP 连接都有独立的源端口 ​

为什么每条 TCP 连接需要独立端口 ​

用 lsof / netstat 验证 ​

端口范围 ​

端口总结表 ​

端口与连接速查表 ​

Q14：如果用 multiprocessing.Process 的 fork 模式启动 Worker，Redis 连接会出什么问题？ ​

Q15：一个 Worker 进程实际同时占用的连接数是多少？连接池 max_connections=10 是不是浪费？ ​

1. 单线程顺序执行模式（1 个连接） ​

2. 3 线程分镜并行模式（3-4 个连接，本项目实际配置） ​

3. 为什么 max_connections=10 不浪费 ​

4. 多 Worker 进程的总连接数 ​

第五章：Redis 6.0 多线程 I/O ​

本章导读 ​

Q16：Redis 6.0 引入了多线程 I/O，这会影响 BRPOP 的原子性保证吗？ ​

第六章：崩溃恢复与任务丢失 ​

本章导读 ​

Q17：Worker 从 BRPOP 取出任务后、执行前崩溃了，这个任务会怎样？ ​

Q18：本项目如何监控队列积压？多少个任务算积压？ ​

第七章：超时与竞争 ​

本章导读 ​

Q19：BRPOP 超时和 LPUSH 同时发生会怎样？会不会出现"既收到 nil 又拿到元素"的情况？ ​

Q20：LPUSH + BRPOP 实现 FIFO 队列，那 LPUSH + LPOP 是什么模型？为什么本项目不用 LPUSH + LPOP？ ​

第八章：综合场景 ​

本章导读 ​

Q21：100 个用户同时提交任务，从 BRPOP 到任务完成，完整经历了哪些步骤？ ​

Q22：如果 Redis 服务重启了，等待链表和队列中的任务会怎样？ ​

Q23：为什么本项目不用 RabbitMQ 或 Kafka，而是用 Redis BRPOP？ ​

第九章：源码级深入 ​

本章导读 ​

Q24：signalKeyAsReady 和 handleClientsBlockedOnKeys 的调用关系是什么？为什么唤醒后要 break？ ​

Q25：brpop 在 redis-py 内部是如何借出和归还连接的？和 hset 有什么区别？ ​

1. brpop 和 hset 各自的借还流程 ​

2. ⚠️ 关键澄清：单线程下两者不会"同时"占用连接 ​

Redis 技术问答：队列、连接与并发专题总集

合并来源

术语说明：文档里的 Worker 默认代表独立执行单元

第一章：BRPOP 基础语法

本章导读

Q1：BRPOP 和 RPOP 的核心区别是什么？为什么本项目用 BRPOP？

RPOP 轮询方式

BRPOP 阻塞等待方式

BRPOP 不会阻塞 Redis 服务端主线程

本章小结

Q2：`brpop(["task:queue:vip", "task:queue:normal"], timeout=5)` 的优先级规则是什么？如果 VIP 队列持续有任务，普通队列会怎样？

Q3：`redis_client.lpush("task:queue", 101)` 的返回值是什么？为什么不是 101？

Q4：为什么 Worker 取出 task_id 后需要 `int(task_id)` 转换？

第二章：阻塞机制与连接管理

本章导读

Q5：BRPOP 阻塞期间，Redis 服务端的主线程会卡住吗？其他客户端的命令能执行吗？

Q6：一个 Worker 进程中，BRPOP 阻塞期间还能执行 `hset` 写进度吗？如果能，用的是同一个 TCP 连接吗？

1. 单线程下不可能并行（常见误区）

2. 本项目实际是顺序执行（不是并行）

3. 真正的并行需要多线程

4. 连接池状态对比

5. 如果 `max_connections=1` 会怎样

Q7：`timeout=0`（永久阻塞）和 `timeout=5` 有什么实际区别？为什么不推荐 `timeout=0`？

第三章：单线程模型与原子性

本章导读

Q8：Redis 是单线程的，为什么还能支撑 10 万 QPS？单线程不会成为瓶颈吗？

Q9：10 个 Worker 同时 `BRPOP task:queue 5`，这时有人 `LPUSH task:queue 6`，为什么不会有两个 Worker 同时拿到 "6"？

Q10：LPUSH 唤醒 BRPOP 时，数据是先存入列表再弹出，还是直接传递？

Q10.1：BRPOP 的"推送"本质是什么？是延迟响应还是真正的推送？

1. 三种模型的严格区分

2. 延迟响应的完整流程

3. 延迟响应的三个关键特征

4. "推送"说法的精确含义

5. 与全双工推送的对比

6. 为什么说是"服务端主动推送"

7. 总结

Q11：假设 Worker-1 和 Worker-2 几乎同时发送 BRPOP，API 也几乎同时发送 LPUSH，三种命令到达 Redis 的顺序不同，结果会怎样？

第四章：多 Worker 与连接池

本章导读

Q12：3 个 Worker 进程启动后，各自有独立的 `redis_client` 吗？Redis 服务端看到的是 3 个连接还是 1 个？

Q13：Redis 等待链表里存的是什么？是"Worker 进程"吗？

⚠️ 关键澄清：等待链表节点数 ≠ 连接总数

场景对比：一个进程 + 3 个连接，等待链表节点数不同

速查表

用 CLIENT LIST 验证

本章小结

补充：每条 TCP 连接都有独立的源端口

为什么每条 TCP 连接需要独立端口

用 `lsof` / `netstat` 验证

端口范围

端口总结表

端口与连接速查表

Q14：如果用 `multiprocessing.Process` 的 `fork` 模式启动 Worker，Redis 连接会出什么问题？

Q15：一个 Worker 进程实际同时占用的连接数是多少？连接池 `max_connections=10` 是不是浪费？

1. 单线程顺序执行模式（1 个连接）

2. 3 线程分镜并行模式（3-4 个连接，本项目实际配置）

3. 为什么 `max_connections=10` 不浪费

4. 多 Worker 进程的总连接数

第五章：Redis 6.0 多线程 I/O

本章导读

Q16：Redis 6.0 引入了多线程 I/O，这会影响 BRPOP 的原子性保证吗？

第六章：崩溃恢复与任务丢失

本章导读

Q17：Worker 从 BRPOP 取出任务后、执行前崩溃了，这个任务会怎样？

Q18：本项目如何监控队列积压？多少个任务算积压？

第七章：超时与竞争

本章导读

Q19：BRPOP 超时和 LPUSH 同时发生会怎样？会不会出现"既收到 nil 又拿到元素"的情况？

Q20：LPUSH + BRPOP 实现 FIFO 队列，那 LPUSH + LPOP 是什么模型？为什么本项目不用 LPUSH + LPOP？

第八章：综合场景

本章导读

Q21：100 个用户同时提交任务，从 BRPOP 到任务完成，完整经历了哪些步骤？

Q22：如果 Redis 服务重启了，等待链表和队列中的任务会怎样？

Q23：为什么本项目不用 RabbitMQ 或 Kafka，而是用 Redis BRPOP？

第九章：源码级深入

本章导读

Q24：`signalKeyAsReady` 和 `handleClientsBlockedOnKeys` 的调用关系是什么？为什么唤醒后要 `break`？

Q25：`brpop` 在 redis-py 内部是如何借出和归还连接的？和 `hset` 有什么区别？

1. brpop 和 hset 各自的借还流程

2. ⚠️ 关键澄清：单线程下两者不会"同时"占用连接