Redis 技术笔记

主题

Redis BRPOP 完整时序与代码解析 ​

10 个 Worker 抢 1 个任务，为什么不会有两个同时拿到 日期：2026-10-14

一、命令语义补全 ​

1.1 BRPOP task:queue 5 的完整语义 ​

文章里常见的简写：

Worker-1: BRPOP task:queue 5

这不是可运行语法，而是伪代码。逐部分拆解：

部分含义
BRPOPBlocking Right POP，阻塞式从列表右端弹出元素
task:queue列表 key 名
5阻塞超时时间，单位秒

完整语义：从 task:queue 列表右端弹出一个元素；若列表为空，则阻塞等待最多 5 秒；超时仍为空则返回 nil。

1.2 那行伪代码的完整含义 ​

Worker-1 这个进程，通过它自己那条 TCP 连接，向 Redis 发送 BRPOP task:queue 5 命令，请求从任务队列右端取一个任务，最多等 5 秒。

1.3 对应的真实代码 ​

python
# app/worker.py — Worker 进程入口
import redis

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

# 循环调用：持续消费队列，直到进程被终止
while True:
    # 这一行就是 "Worker-1: BRPOP task:queue 5" 的真实代码
    result = redis_client.brpop("task:queue", timeout=5)

    if result:
        # result 是元组: ("task:queue", "6")
        queue_name, task_id = result
        run_task(int(task_id))
    else:
        # 5 秒超时，返回 None → 跳过，进入下一轮 BRPOP
        pass

为什么必须用 while True：单次调用拿到一个任务（或超时）后函数就返回、进程退出，队列里剩下的任务没人处理。Worker 是常驻进程，必须循环调用 BRPOP 才能持续消费。

超时返回 None 时循环自然继续：timeout=5 的设计正是为了配合循环——超时返回 None → if result 为假 → 跳过 → 下一轮继续 BRPOP。这就是 5 秒超时的意义所在（详见 1.5）。

生产环境还需补优雅退出与异常兜底，见 1.6。

1.4 10 个 Worker 的完整展开 ​

Worker-1:  BRPOP task:queue 5   →  进程A 的 TCP连接(fd=10) 发出
Worker-2:  BRPOP task:queue 5   →  进程B 的 TCP连接(fd=11) 发出
Worker-3:  BRPOP task:queue 5   →  进程C 的 TCP连接(fd=12) 发出
...
Worker-10: BRPOP task:queue 5   →  进程J 的 TCP连接(fd=19) 发出

每个 Worker 都是独立进程 + 独立 TCP 连接，10 个 Worker = 10 条连接 = Redis 等待链表里 10 个 client 节点。

1.5 为什么超时是 5 秒 ​

Worker 的消费循环是 while True 反复调用：

python
while True:
    result = redis_client.brpop("task:queue", timeout=5)
    if result:
        run_task(int(result[1]))

超时返回 nil 后循环继续下一轮 BRPOP。设 5 秒而不是 0 的原因：

取值行为问题
timeout=0永久阻塞连接断开或需优雅退出时，Worker 卡死在 BRPOP 上无法响应
timeout=5最多等 5 秒折中：不频繁空转，又能周期性检查退出信号、重连、上报心跳

关键：这个 5 秒只阻塞客户端，不阻塞 Redis 服务端。

1.6 生产级 Worker 循环：优雅退出 + 异常兜底 ​

1.5 节的 while True 是最简形态，生产环境还需补两点：优雅退出（否则 kill 时可能丢任务）和异常兜底（否则一次 Redis 抖动就整个进程挂掉）。

1.6.1 优雅退出 ​

while True 无法响应终止信号，进程被 kill 时可能正在 run_task 中途，任务状态不一致。改用信号控制循环条件：

python
# app/worker.py — 优雅退出
import signal

running = True

def handle_signal(signum, frame):
    global running
    running = False

signal.signal(signal.SIGTERM, handle_signal)  # kill / docker stop
signal.signal(signal.SIGINT, handle_signal)   # Ctrl+C

while running:
    result = redis_client.brpop("task:queue", timeout=5)
    if result:
        run_task(int(result[1]))

为什么必须配合 timeout=5：收到信号时 Worker 可能正阻塞在 BRPOP 上。若 timeout=0（永久阻塞），信号处理函数执行完 running = False 后，主线程仍卡在 BRPOP 里无法返回，循环条件永远检查不到。timeout=5 保证最多 5 秒后 BRPOP 返回 None，循环重新判断 running，进程正常退出。

1.6.2 异常兜底 ​

Redis 连接抖动、任务处理抛异常都会中断循环。需要分层捕获：

python
# app/worker.py — 异常兜底
import time
import logging
import redis

logger = logging.getLogger(__name__)

while running:
    try:
        result = redis_client.brpop("task:queue", timeout=5)
        if result:
            run_task(int(result[1]))
    except redis.ConnectionError as e:
        # 连接类异常：退避重试，不退出循环
        logger.error("Redis 连接异常，5 秒后重试", exc_info=True)
        time.sleep(5)
    except redis.TimeoutError as e:
        # 读写超时：同样退避重试
        logger.error("Redis 超时，5 秒后重试", exc_info=True)
        time.sleep(5)
    except Exception as e:
        # 任务处理异常：记录后继续消费下一个任务
        logger.error("任务处理异常", exc_info=True)

1.6.3 异常分类处理原则 ​

异常类型处理方式原因
redis.ConnectionError退避 5 秒后重试网络抖动/Redis 重启，连接可恢复
redis.TimeoutError退避 5 秒后重试服务端繁忙，稍后重试即可
任务处理异常（run_task 内）记录日志，继续下一个任务单个任务失败不应拖垮整个 Worker
配置错误（如 key 类型错误）记录日志，继续属于数据问题，非进程问题

核心原则：Worker 循环永不因异常退出。任何异常都只影响当前这一轮，记录日志后继续下一轮 BRPOP。

1.6.4 完整生产级模板 ​

python
# app/worker.py — 生产级 Worker 循环
import signal
import time
import logging
import redis
from app.redis_client import redis_client

logger = logging.getLogger(__name__)

running = True

def handle_signal(signum, frame):
    global running
    logger.info("收到退出信号 %s，等待当前任务完成后退出", signum)
    running = False

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

def process_queue():
    while running:
        try:
            result = redis_client.brpop("task:queue", timeout=5)
            if result:
                queue_name, task_id = result
                run_task(int(task_id))
        except (redis.ConnectionError, redis.TimeoutError) as e:
            logger.error("Redis 异常，5 秒后重试", exc_info=True)
            time.sleep(5)
        except Exception as e:
            logger.error("任务处理异常", exc_info=True)

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

退出时序：

收到 SIGTERM
    ↓
handle_signal 执行 → running = False
    ↓
若正在 run_task → 等当前任务跑完（不中断）
    ↓
若正在 BRPOP 阻塞 → 最多等 5 秒超时返回 None
    ↓
循环条件 running 为假 → 退出循环
    ↓
打印 "Worker 已优雅退出" → 进程结束

二、完整时序：10 个 Worker 抢 1 个任务 ​

2.1 初始状态 ​

task:queue: []
waiting_clients: []

2.2 阶段一：10 个 Worker 依次阻塞 ​

t=0.0ms   Worker-1 发 BRPOP task:queue 5
          → Redis 主线程执行 → 队列空 → client(fd=10) 挂入等待链表 → 立即返回处理下一条

t=0.1ms   Worker-2 发 BRPOP task:queue 5
          → Redis 主线程执行 → 队列空 → client(fd=11) 挂入等待链表 → 立即返回

t=0.2ms   Worker-3 发 BRPOP task:queue 5
          → Redis 主线程执行 → 队列空 → client(fd=12) 挂入等待链表 → 立即返回

...

t=0.9ms   Worker-10 发 BRPOP task:queue 5
          → Redis 主线程执行 → 队列空 → client(fd=19) 挂入等待链表 → 立即返回

         此时 10 个 Worker 全部阻塞在客户端侧，Redis 主线程空闲

等待链表状态：

task:queue → [client(fd=10), client(fd=11), client(fd=12), ..., client(fd=19)]
               ↑ Worker-1      ↑ Worker-2      ↑ Worker-3            ↑ Worker-10
               （按 BRPOP 到达顺序 FIFO 排队）

2.3 阶段二：LPUSH 唤醒 ​

t=3000ms  API 发 LPUSH task:queue 6

Redis 主线程执行 LPUSH（单线程，串行）:
  1. 执行 LPUSH → 列表变为 [6]
  2. 调用 signalKeyAsReady("task:queue")
  3. LPUSH 命令返回

主线程事件循环处理 ready_keys:
  4. 检查 task:queue → 有元素
  5. 检查等待链表 → client(fd=10) 是第一个
  6. 弹出 "6"（右端）→ 通过 fd=10 发给 Worker-1
  7. 从列表删除 "6"，列表变为 []
  8. 从等待链表移除 client(fd=10)
  9. break ← 只唤醒一个，不唤醒 fd=11~19

2.4 阶段三：结果 ​

Worker-1  收到 "6" ✅  → brpop() 返回 → 开始处理任务
Worker-2  仍在等待（fd=11 还在链表里）
Worker-3  仍在等待
...
Worker-10 仍在等待

task:queue: []
waiting_clients: [client(fd=11), client(fd=12), ..., client(fd=19)]

2.5 完整时序图 ​

Worker-1   Worker-2   Worker-3   ...  Worker-10      API        Redis 主线程
    │          │          │              │            │              │
    │ BRPOP ───┤          │              │            │              │
    │          │          │              │            │  执行 BRPOP   │
    │          │          │              │            │  列表空       │
    │          │          │              │            │  fd=10→链表   │
    │ 等待...  │          │              │            │              │
    │          │ BRPOP ───┤              │            │              │
    │          │          │              │            │  执行 BRPOP   │
    │          │          │              │            │  列表空       │
    │          │          │              │            │  fd=11→链表   │
    │          │ 等待...  │              │            │              │
    │          │          │ BRPOP ───────┤            │              │
    │          │          │              │            │  ...         │
    │          │          │              │ BRPOP ─────┤              │
    │          │          │              │            │  fd=19→链表   │
    │          │          │              │  等待...   │              │
    │          │          │              │            │ LPUSH "6" ───┤
    │          │          │              │            │              │
    │          │          │              │            │  ┌─执行 LPUSH─┐
    │          │          │              │            │  │ 1.插入"6"  │
    │          │          │              │            │  │ 2.检查链表 │
    │          │          │              │            │  │ 3.fd=10在等│
    │          │          │              │            │  │ 4.弹出"6"  │
    │          │          │              │            │  │ 5.发给W1   │
    │          │          │              │            │  │ 6.移除W1   │
    │          │          │              │            │  │ 7.break    │
    │          │          │              │            │  └────────────┘
    │ ← "6" ───┤          │              │            │              │
    │          │          │              │            │  "6" 已被W1取走│
    │          │ 仍在等待 │              │            │              │
    │          │          │ 仍在等待     │            │              │

三、完整代码实现 ​

3.1 Worker 进程（客户端侧） ​

python
# app/redis_client.py — 模块级单例
import redis
from app.config import settings

redis_client = redis.Redis(
    host=settings.redis_host,
    port=settings.redis_port,
    db=0,
    decode_responses=True,
    max_connections=10,
    socket_timeout=30,
    socket_connect_timeout=5,
    health_check_interval=30,
)
python
# app/worker.py — Worker 入口（最简形态，生产级见 1.6.4）
from app.redis_client import redis_client  # ← 每个进程各自 import

def process_queue():
    while True:
        # 这一行就是 "Worker-N: BRPOP task:queue 5"
        result = redis_client.brpop("task:queue", timeout=5)
        if result:
            queue_name, task_id = result
            run_task(int(task_id))
        # result 为 None 表示 5 秒超时，循环继续

生产环境请使用 1.6.4 的完整模板（含优雅退出与异常兜底）。

启动 3 个独立进程：

bash
.venv/bin/python -m app.worker &   # 进程 A
.venv/bin/python -m app.worker &   # 进程 B
.venv/bin/python -m app.worker &   # 进程 C

3.2 生产者（API 侧） ​

python
# app/api.py — 任务推入
from app.redis_client import redis_client

def submit_task(task_id: int):
    # 推入队列，触发 Redis 唤醒等待中的 Worker
    redis_client.lpush("task:queue", task_id)

3.3 Redis 服务端源码（唤醒逻辑） ​

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

void lpushCommand(client *c) {
    // 1. 查找或创建列表
    robj *o = lookupCreateKey(c->db, c->argv[1]);

    // 2. 插入元素到列表头部
    listTypePush(o, c->argv[2], LIST_HEAD);

    // 3. 关键：检查是否有 BRPOP 等待者
    signalKeyAsReady(c->db, c->argv[1]);
}
c
// 唤醒等待客户端（命令执行完成后，主线程检查 ready_keys）

void handleClientsBlockedOnKeys(void) {
    while (listLength(server.ready_keys) > 0) {
        listNode *ln = listFirst(server.ready_keys);
        readyList *rl = ln->value;

        dictEntry *de = dictFind(rl->db->blocking_keys, rl->key);
        list *clients = dictGetVal(de);

        listNode *clientNode = listFirst(clients);
        while (clientNode) {
            client *receiver = clientNode->value;

            robj *value = listTypePop(o, LIST_TAIL);
            if (value) {
                // 把值发给这个客户端
                addReplyArrayLen(receiver, 2);
                addReplyBulk(receiver, rl->key);
                addReplyBulk(receiver, value);

                // 从等待链表移除
                listDelNode(clients, clientNode);

                break; // ← 只唤醒一个客户端（关键！）
            }
            clientNode = listNextNode(clientNode);
        }
    }
}

break 是核心：弹出一个元素、发给一个客户端后立即退出循环。即使等待链表中有 10 个 Worker，也只唤醒第一个。

四、为什么不可能两个 Worker 同时拿到 ​

4.1 三种到达顺序推演 ​

假设 Worker-1 和 Worker-2 几乎同时发送 BRPOP：

时间 0.000ms: Worker-1 发出 BRPOP task:queue 5
时间 0.001ms: Worker-2 发出 BRPOP task:queue 5
时间 0.002ms: API 发出 LPUSH task:queue 6

由于网络延迟，三条命令到达 Redis 的顺序可能不同。但不管什么顺序，Redis 串行执行：

情况 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 继续等待 ✅

所有情况都只有一个 Worker 拿到元素。

4.2 根本原因 ​

任意时刻，只有一条命令在执行
    ↓
LPUSH 和 BRPOP 不可能"同时"执行
    ↓
LPUSH 执行时检查等待链表是 LPUSH 自己的原子步骤
    ↓
BRPOP 弹出元素是 BRPOP 自己的原子步骤
    ↓
两条命令之间不存在"间隙"让数据不一致

4.3 与多线程数据库的对比 ​

多线程数据库：
  Thread-1 执行 LPUSH:
    1. 锁住列表
    2. 插入 "6"
    3. 检查等待队列
    4. 发给 Worker-1
    5. 解锁列表

  Thread-2 同时执行 BRPOP:
    1. 等待锁...
    2. 获取锁
    3. 列表非空（如果 LPUSH 还没删 "6"）
    4. 弹出 "6"    ← 两个线程拿到同一个 "6"！
    5. 解锁

多线程必须用锁保证安全，但锁有性能开销和死锁风险。Redis 用单线程直接避免了这些问题。

五、批量 LPUSH 的完整时序 ​

5.1 命令 ​

bash
LPUSH task:queue 6 7 8

5.2 执行流程 ​

LPUSH 执行（原子）:
  列表变为 [8, 7, 6]
  调用 signalKeyAsReady

主线程处理 ready_keys（仍在 LPUSH 命令内）:

  第1轮循环:
    检查 task:queue → 有元素
    弹出 "6"（右端）→ 发给 W1
    从等待链表移除 W1
    列表变为 [8, 7]
    break ← 只唤醒一个

  主线程继续事件循环，发现 task:queue 还在 ready_keys 里

  第2轮循环:
    检查 task:queue → 还有元素
    弹出 "7" → 发给 W2
    从等待链表移除 W2
    列表变为 [8]
    break

  第3轮循环:
    检查 task:queue → 还有元素
    弹出 "8" → 发给 W3
    从等待链表移除 W3
    列表变为 []
    break

  第4轮循环:
    检查 task:queue → 空，不弹出
    结束处理

结果：3 个元素分别发给 W1、W2、W3，W4~W10 继续等待。

六、超时与 LPUSH 的竞争 ​

6.1 超时触发 ​

当 BRPOP 的 timeout 到期时，Redis 会把客户端从等待链表移除，并返回 nil。

c
int blockingClientTimeout(client *c) {
    unblockClient(c);      // 从等待链表移除
    addReplyNull(c);       // 返回 nil
}

6.2 竞争场景 ​

如果在超时发生的同一瞬间，有人 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 又拿到元素"或"元素丢失"。

6.3 超时移除也是原子的 ​

超时检查在主线程的事件循环中执行，与命令执行一样是串行的：

主线程事件循环:
  处理命令1（LPUSH）
  处理命令2（GET）
  检查超时定时器 ← 如果有 BRPOP 超时了，在这里处理
  处理命令3（SET）

超时处理不会被 LPUSH 打断，LPUSH 也不会被超时处理打断。

七、四层保证总结 ​

第一层：单线程命令执行
  ── 所有命令串行执行，不存在并发
  ── LPUSH 和 BRPOP 不可能同时执行

第二层：命令原子性
  ── 一条命令内部的多个步骤不可被打断
  ── LPUSH 的"插入 + 检查等待链表 + 唤醒"是一个原子操作

第三层：等待链表 FIFO
  ── 多个 Worker 等待时，按 BRPOP 调用顺序排队
  ── 唤醒时只取链表第一个（break），不批量唤醒

第四层：事件循环串行
  ── 命令执行、超时检查、唤醒操作都在同一个事件循环中
  ── 不存在"命令执行了一半被超时打断"的情况

这四层叠加，构成了 Redis 对"只有一个 Worker 能拿到元素"的完整保证。

八、关键概念速查 ​

概念说明
谁在阻塞客户端（Worker 进程）阻塞，不是 Redis 服务端
阻塞多久最多 5 秒，或被 LPUSH 提前唤醒
为什么不会重复拿单线程串行执行 + 等待链表 FIFO + 唤醒后立即 break
5 的作用给 Worker 一个周期性醒来的机会，用于检查退出信号/重连/心跳
数据是否经过列表不经过。LPUSH 直接把值发给等待者，O(1) 完成
多线程 I/O 是否影响不影响。命令执行仍在主线程串行
Worker 崩溃任务是否丢会丢。需 RPOPLPUSH 模式弥补（见下）

8.1 任务丢失的弥补方案 ​

Redis BRPOP 没有 ACK 机制，Worker 取走任务后崩溃则任务丢失。用 RPOPLPUSH 弥补：

python
# 弹出并转入"处理中"队列
task_id = redis_client.rpoplpush("task:queue", "task:processing")
# 处理完成后删除
redis_client.lrem("task:processing", 1, task_id)
# Worker 重启时扫描 task:processing 恢复

九、相关文档索引 ​

文档路径内容
Redis 单线程与多客户端交互Redis单线程与多客户端交互.md多进程客户端与单线程服务端的连接拓扑、连接池
Redis 原子性保证机制Redis原子性保证机制.md单线程模型、等待链表源码级分析、命令原子性
BRPOP 深度解析Redis-BRPOP深度解析.mdBRPOP 语法、阻塞机制、多 Worker 行为、注意事项
Worker 编排器解析Worker编排器深度解析.mdWorker 启动、消费循环、异常处理、进度缓存