Redis 技术笔记

主题

Redis 推拉模型与源码解析：BRPOP 唤醒机制 ​

Redis 如何让等待中的客户端收到列表元素 日期：2026-08-26

本文以 Redis 7.x 的实现思路为例。源码函数名和文件位置可能随 Redis 版本变化；文中的 C 代码均为删减后的示意代码，不是可直接编译的源码。

一、核心结论 ​

Redis 在 BRPOP 等待期间不需要客户端再次发送 RPOP 或轮询请求。

更准确地说，客户端先发送 BRPOP；列表为空时，Redis 暂不返回，而是记录客户端的阻塞状态和等待 key。生产者执行 LPUSH 后，Redis 处理已就绪的阻塞 key，生成原 BRPOP 请求的响应，将 RESP 数据加入客户端输出缓冲区，再通过可写事件发送到 socket。

LPUSH 到达 → Redis 执行 LPUSH → 标记 key 已就绪
           → 事件循环处理阻塞客户端 → 弹出元素并生成 BRPOP 响应
           → 加入输出缓冲区 → 可写事件写入 socket → 客户端收到

术语澄清：业务语义上可以把它称为“服务端推动任务到等待客户端”；协议层面，BRPOP 严格来说是对既有请求的延迟响应（Delayed Response），不是无请求的通用服务端推送。详见第十章。

二、完整交互流程 ​

2.1 时序图 ​

Worker 进程                               Redis 服务端
                                         主线程
─────────────                           ──────────────
brpop("task:queue", 5)
  │
  ├─ 1. 发送 BRPOP 命令 ──────────────►  收到 BRPOP
  │                                      ├─ 列表为空
  │                                      ├─ 把 client(fd=10) 加入等待链表
  │                                      ├─ 记录 5 秒阻塞截止时间
  │                                      ├─ 当前请求暂不生成响应
  │                                      └─ 主线程继续处理其他命令
  │
  ├─ 2. read_response() 阻塞              （等待中...）
  │    客户端 socket fd 挂起             （等待中...）
  │    不占 CPU                           （等待中...）
  │                                      （等待中...）
  │
  │                                      API 发来 LPUSH task:queue 6
  │                                      ├─ 主线程执行 LPUSH
  │                                      ├─ 插入 "6" 到列表
  │                                      ├─ 标记 task:queue 已就绪
  │                                      ├─ 事件循环处理 ready key，发现 client(fd=10)
  │                                      ├─ 弹出 "6"
  │                                      ├─ addReply(...) 加入输出缓冲区
  │                                      ├─ 解除 client(fd=10) 的阻塞状态
  │                                      └─ 可写事件随后把缓冲区写入 socket
  │
  ◄──────────────────────────────────────  3. 数据到达 socket
  │
  read_response() 解除阻塞
  result = ("task:queue", "6")
  │
  └─ Worker 拿到任务，开始处理

2.2 三个阶段详解 ​

阶段 1：Worker 发送 BRPOP，Redis 加入等待链表

python
# Worker 进程（redis-py 内部）
conn = pool.get_connection()           # 借出 TCP 连接
conn.send_command("BRPOP", "task:queue", 5)
result = conn.read_response()           # 阻塞，等待 Redis 回复
c
// Redis 服务端（列表命令和阻塞客户端处理相关源码，文件位置随版本变化）
void blockingPopGenericCommand(client *c, int where) {
    // 检查列表是否非空
    robj *o = lookupKey(c->db, c->argv[1]);
    if (o && listTypeLength(o) > 0) {
        // 非空：直接弹出返回（不阻塞）
        // ...
        return;
    }
    // 空列表：记录阻塞 key、客户端和阻塞截止时间
    blockForKeys(c, c->argv + 1, 1, timeout, NULL);
    // ↑ 把 client 加入 db->blocking_keys["task:queue"] 等待链表
    // ↑ 记录超时截止时间，由 Redis 的事件/定时任务检查
    // ↑ client 状态标记为 BLOCKED
    // ↑ 命令不返回（不写入输出缓冲区）
    // ↑ 主线程继续处理下一个客户端的命令
}

阶段 2：Redis 收到 LPUSH，处理等待中的 BRPOP

c
// Redis 服务端（列表命令相关源码；具体文件和函数随版本变化）
void lpushCommand(client *c) {
    // 1. 插入元素到列表头部
    listTypePush(o, c->argv[j], LIST_HEAD);

    // 2. 标记 key 为"有待处理的阻塞客户端"
    signalKeyAsReady(c->db, c->argv[1]);
    // ↑ 把 key 加入 server.ready_keys 队列
    //   不同 Redis 版本中调用者可能是 signalKeyAsReady 或列表专用包装函数
    // ↑ 注意：这里还不唤醒客户端，只是标记
}

// LPUSH 命令返回后，事件循环处理 ready_keys
void handleClientsBlockedOnKeys(void) {
    while (listLength(server.ready_keys) > 0) {
        // 取出 ready key
        // 查找等待这个 key 的客户端链表
        list *clients = dictGetVal(de);

        // 按阻塞时间顺序处理等待客户端；只要列表仍有元素，
        // 一个 ready key 可能服务多个等待客户端。
        while (listLength(clients) > 0 && listTypeLength(o) > 0) {
            client *receiver = listFirst(clients)->value;

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

            if (value) {
                // ★ 关键：把延迟的 BRPOP 响应加入客户端输出缓冲区
                addReplyArrayLen(receiver, 2);
                addReplyBulk(receiver, key);       // 队列名
                addReplyBulk(receiver, value);     // 元素值

                // 解除阻塞通常会负责从相关等待链表中移除客户端
                unblockClient(receiver);
                continue;
            }
            break;
        }
    }
}

阶段 3：Worker 的 socket 收到数据，read_response 解除阻塞

python
# Worker 进程（redis-py 内部）
# Redis 把挂起的 BRPOP 响应通过可写事件发送到 socket，
# 操作系统的 recv 队列随后有数据了
# conn.read_response() 从 socket 读取 RESP 协议数据
# 解析为 python 对象
result = ("task:queue", "6")  # decode_responses=True 时的返回值

三、推/拉模型对比 ​

3.1 两种理论模型 ​

模型机制消费者额外请求延迟
阻塞请求的延迟响应（Redis 实际行为）LPUSH 使已挂起的 BRPOP 请求获得响应无额外 RPOP/轮询请求取决于部署和负载
通知后拉取（假设模型）客户端收到通知后再发 RPOP至少多一次拉取请求取决于部署和负载

3.2 如果是"客户端拉取"模型（Redis 实际不这样做） ​

假设的拉取模型流程:

LPUSH 到达 → Redis 执行 LPUSH → 列表变为 [6]
           → Redis 通知 Worker "task:queue 有数据了"
           → 通知写入 Worker 的 socket
           → Worker 收到通知
           → Worker 发送 RPOP task:queue 命令 ────► Redis
           → Redis 执行 RPOP → 返回 "6" ──────► Worker
           → Worker 拿到 "6"

这里的通知和拉取次数取决于通知机制，不能直接换算成固定 RTT 或固定毫秒数。

3.3 Redis 实际的延迟响应模型 ​

实际的处理流程:

LPUSH 到达 → Redis 执行 LPUSH → 列表变为 [6]
           → 标记 key 已就绪，事件循环发现 Worker 在等
           → 弹出 "6"
           → addReply() 把 BRPOP 响应加入 Worker 的输出缓冲区
           → Redis 通过可写事件写入 Worker 的 socket
           → Worker 的 read_response() 收到数据

消费者侧：0 次额外 RPOP/轮询请求
实际延迟：取决于 Redis 事件循环、输出缓冲区、网络和客户端调度

3.4 对比表 ​

维度BRPOP 延迟响应（实际）通知后拉取（假设）
触发方LPUSH 使挂起的 BRPOP 获得响应客户端收到通知后发起 RPOP
消费者额外请求0 条至少 1 条拉取命令
总延迟取决于实现和部署取决于实现和部署，通常多一个请求路径
Redis 命令生产者 LPUSH + 消费者既有 BRPOP生产者 LPUSH + 通知机制 + 消费者 RPOP
实现复杂度服务端处理客户端需要额外逻辑
取数时机Redis 处理 ready key 时完成通知和拉取之间可能被其他客户端抢先

3.5 "客户端拉取"模型的竞争问题 ​

如果 Redis 用"拉取"模型，会出现竞争问题：

LPUSH task:queue 6
  → Redis 通知 W1 "有数据"

在 W1 发 RPOP 之前：
  W2 也发了 RPOP task:queue
  → W2 抢先弹出 "6"

W1 发 RPOP
  → 列表已空，返回 nil
  → W1 白通知了

BRPOP 的优势是服务端可以在阻塞客户端处理阶段直接完成原 BRPOP 请求的取数和响应，消费者不必先收到通知再发送 RPOP。至于不同命令之间的严格先后关系，应以具体 Redis 版本和命令排列为准，不能简单概括为整个“插入→唤醒→写 socket”都属于 LPUSH 命令内部。

四、源码级解析 ​

4.1 addReply：把响应加入客户端输出缓冲区 ​

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

void addReply(client *c, robj *obj) {
    // 检查客户端是否可写
    if (prepareClientToWrite(c) != C_OK) return;

    // 写入客户端的输出缓冲区
    // 不需要客户端发任何请求
    if (sdslen(obj->ptr) > 0) {
        _addReplyToBuffer(c, obj->ptr, sdslen(obj->ptr));
    }

    // 标记客户端有数据待发送，并注册可写事件。
    // 后续事件循环通过写事件把缓冲区内容写入 socket。
}

关键：addReply 负责构造并排队响应，不等于已经完成 socket 写入。Redis 通常为该客户端注册可写事件，后续由写事件处理函数刷新输出缓冲区；写入可能因 socket 暂不可写而分多次完成。

4.2 addReplyBulk：写入一个完整的 RESP 批量字符串 ​

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

void addReplyBulk(client *c, robj *obj) {
    // RESP 协议格式: $长度\r\n数据\r\n
    // 例如: $1\r\n6\r\n  （字符串 "6"）
    addReplyBulkLen(c, obj);     // 写入 $1\r\n
    addReply(c, obj);            // 写入 6
    addReply(c, shared.crlf);    // 写入 \r\n
}

void addReplyArrayLen(client *c, long length) {
    // RESP 协议格式: *元素数\r\n
    // 例如: *2\r\n  （数组，2 个元素）
    // 完整 BRPOP 返回值: *2\r\n$10\r\ntask:queue\r\n$1\r\n6\r\n
}

4.3 BRPOP 唤醒时的完整 addReply 调用 ​

c
// Redis 源码示意（列表命令和阻塞客户端处理相关文件随版本变化）

void handleClientsBlockedOnKeys(void) {
    // ...
    while (listLength(clients) > 0 && listTypeLength(o) > 0) {
        client *receiver = listFirst(clients)->value;

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

        if (value) {
            // ★ 把延迟的 BRPOP 响应加入客户端输出缓冲区
            // 等价于写入: *2\r\n$10\r\ntask:queue\r\n$1\r\n6\r\n
            addReplyArrayLen(receiver, 2);          // *2\r\n
            addReplyBulk(receiver, key);            // $10\r\ntask:queue\r\n
            addReplyBulk(receiver, value);          // $1\r\n6\r\n

            // 解除阻塞状态；同时从相关等待关系中移除客户端
            unblockClient(receiver);
            // ↑ 从阻塞状态和相关等待关系中移除
            // ↑ 客户端状态从 BLOCKED 改为 UNBLOCKED

            // 如果列表仍有元素，处理循环可以继续服务其他等待客户端
        } else {
            break;
        }
    }
}

4.4 数据从 Redis 内存到 Worker 进程的完整路径 ​

Redis 主线程                           Worker 进程
┌─────────────────────┐               ┌─────────────────────┐
│ 1. listTypePop()    │               │                     │
│    弹出 "6"          │               │ 5. socket recv()    │
│                     │               │    从 socket 读取   │
│ 2. addReply()       │               │    RESP 数据         │
│    写入 client 的    │               │                     │
│    输出缓冲区        │               │ 6. redis-py 解析    │
│    buf: *2\r\n...   │               │    RESP 协议         │
│                     │               │                     │
│ 3. 事件循环检测到   │               │ 7. read_response()  │
│    client 有数据    │               │    解除阻塞          │
│    待发送            │               │    返回结果          │
│                     │               │                     │
│ 4. 可写事件处理      │════════════►│                     │
│    刷新输出缓冲区    │   TCP 网络    │                     │
└─────────────────────┘               └─────────────────────┘

| 步骤 | 位置 | 操作 | |---|---|---|---| | 1 | Redis 主线程 | listTypePop 弹出元素 | | 2 | Redis 主线程 | addReply 构造并排队 RESP 响应 | | 3 | Redis 主线程 | 为客户端安排可写事件 | | 4 | Redis 主线程 → 操作系统 | 可写时刷新输出缓冲区 | | 5 | 操作系统 → Worker 进程 | TCP 传输，Worker 端 socket 变为可读 | | 6 | Worker 进程 | redis-py 读取并解析 RESP | | 7 | Worker 进程 | read_response 返回 Python 对象 |

耗时说明：这里不提供固定纳秒或微秒数字。实际延迟必须在目标硬件、网络拓扑、Redis 配置和客户端实现上进行基准测试。

五、redis-py 客户端的视角 ​

5.1 redis-py 的 brpop 内部实现 ​

python
# redis-py 源码 client.py（简化）

class Redis:
    def brpop(self, keys, timeout=0):
        # 1. 借出连接
        conn = self.connection_pool.get_connection("BRPOP")

        try:
            # 2. 发送 BRPOP 命令
            conn.send_command("BRPOP", keys, timeout)

            # 3. 阻塞等待 Redis 回复
            #    这里 socket.recv() 会挂起
            #    Redis 发送挂起的 BRPOP 响应后，recv 返回
            result = conn.read_response()

            return result  # ("task:queue", "6") 或 None
        finally:
            # 4. 归还连接
            self.connection_pool.release(conn)

5.2 read_response 的阻塞与解除 ​

python
# redis-py 源码 connection.py（简化）

class Connection:
    def read_response(self):
        # 从 socket 读取数据
        # 如果 Redis 还没发送挂起的 BRPOP 响应，这里会阻塞
        response = self._parser.read()

        # Redis 发送响应后，socket 有数据可读
        # _parser.read() 解析 RESP 协议，返回 python 对象
        return response
python
# _parser.read() 内部（HiredisParser 或 PythonParser）
def read(self):
    # 底层调用 socket.recv()
    # 操作系统层面：socket 的接收队列没有数据时，recv 阻塞
    # Redis 发送响应后，socket 接收队列有数据，recv 解除阻塞
    chunk = self._sock.recv(65536)
    # 解析 RESP 协议: *2\r\n$10\r\ntask:queue\r\n$1\r\n6\r\n
    # → ("task:queue", "6")

5.3 Worker 端代码的视角 ​

从 Worker 的 Python 代码看，整个过程是一次同步调用：

python
# Worker 进程
def process_queue():
    while True:
        result = redis_client.brpop("task:queue", timeout=5)
        # ↑ 这一行可能阻塞数秒
        # 但 Worker 什么都没做，只是 socket recv 阻塞
        # Redis 把挂起的 BRPOP 响应排队并通过 socket 发送
        # recv 收到数据，brpop 返回

        if result:
            queue_name, task_id = result
            run_task(int(task_id))

Worker 不需要再发送 RPOP 或轮询命令，只是在 socket 上等待原 BRPOP 请求的响应。

六、为什么 Redis 使用阻塞请求而不是通知后拉取 ​

6.1 原因一：减少网络往返 ​

BRPOP 延迟响应模型：Worker 先发送一次 BRPOP，之后不再发送 RPOP
  消费者额外请求：0 条
  延迟：取决于 Redis、网络和客户端调度

通知后拉取模型：Worker 收到通知后再发送 RPOP
  消费者额外请求：至少 1 条拉取命令
  延迟：取决于通知机制、Redis、网络和客户端调度

对于高并发队列场景，每减少一次网络往返都意味着吞吐量提升。

6.2 原因二：避免竞争 ​

Redis 的命令执行是串行的，阻塞客户端也按等待顺序处理：

LPUSH 和后续阻塞客户端处理的关系：
  1. 插入元素
  2. 标记 key 已就绪
  3. ready key 处理阶段按等待顺序弹出元素并生成 BRPOP 响应
  4. 解除已服务客户端的阻塞状态

这里的“原子”只能指单条 Redis 命令执行不会被另一条命令插入；
不能把插入、ready key 处理和 socket 写入都当成 LPUSH 函数内部的原子步骤。

拉取模型的通知和拉取之间存在竞争窗口：

LPUSH 执行:
  1. 插入元素
  2. 通知 W1 "有数据"

W1 还没拉取，W2 发了 RPOP:
  3. W2 抢先弹出元素
  4. W1 拉取时列表已空

结果：W1 白通知了，需要重试

6.3 原因三：简化客户端实现 ​

推送模型下，客户端只需要一次调用：

python
result = redis_client.brpop("task:queue", timeout=5)
# 一次调用，阻塞等待，直接拿到结果

拉取模型下，客户端需要通知 + 拉取 + 重试逻辑：

python
# 假设的拉取模型客户端伪代码
while True:
    notification = wait_for_notification(timeout=5)
    if not notification:
        continue  # 超时

    result = redis_client.rpop("task:queue")
    if result is None:
        continue  # 被其他客户端抢先了，重试

    return result

6.4 总结对比 ​

维度BRPOP 延迟响应（实际）通知后拉取（假设）
消费者额外请求0 条至少 1 条拉取命令
延迟取决于实现和部署取决于实现和部署
取数时机Redis 处理 ready key 时完成通知和拉取之间可能被其他客户端抢先
客户端复杂度一次调用通知 + 拉取 + 重试
超时处理Redis 记录并检查阻塞截止时间客户端管理通知超时
多 Worker 顺序同一 key 的阻塞客户端按等待顺序处理取决于通知和拉取实现

七、与其他推送模型的对比 ​

系统推送机制客户端是否需要额外请求
Redis BRPOPLPUSH 触发挂起的 BRPOP 请求获得响应否，无额外 RPOP 请求
RabbitMQ消息推送到消费者 Channel否，但需要先声明消费者
Kafka消费者 poll 拉取是，客户端主动 poll
WebSocket服务端主动推送帧否，建立连接后自动推送
HTTP Long Polling服务端 hold 住请求，有数据时返回否，但需要重新建立连接
HTTP SSE服务端单向推送事件否，建立连接后持续推送

在本表列举的系统中，Kafka 使用消费者主动 poll() 的拉取模型，适合通过批量拉取提高吞吐量。Redis BRPOP 则让消费者通过一个长期挂起的请求等待结果，通常适合低延迟任务分发。Kafka 并不是所有消息系统中唯一的拉取模型。

八、验证方法 ​

8.1 用 tcpdump 抓包验证 ​

bash
# 抓取 Redis 端口 6380 的 TCP 数据；如果 Redis 使用默认端口，请改为 6379
sudo tcpdump -i lo port 6380 -X

# 1. Worker 发送 BRPOP
#    客户端 → Redis: *3\r\n$5\r\nBRPOP\r\n$10\r\ntask:queue\r\n$1\r\n5\r\n

# 2. Redis 不回复（列表为空，阻塞）

# 3. API 发送 LPUSH
#    客户端 → Redis: *3\r\n$5\r\nLPUSH\r\n$10\r\ntask:queue\r\n$1\r\n6\r\n

# 4. Redis 回复 LPUSH 客户端: :1\r\n  （列表长度）

# 5. Redis 发送挂起的 BRPOP 响应给 Worker（Worker 没有发任何新命令）
#    Redis → Worker: *2\r\n$10\r\ntask:queue\r\n$1\r\n6\r\n

步骤 5 是关键证据：Worker 没有发送新的 RPOP 或轮询命令，但已挂起的 BRPOP 收到了响应。这证明没有额外的消费者拉取请求，不证明响应一定在 LPUSH 函数内部直接完成 socket 写入。

8.2 用 redis-cli MONITOR 验证 ​

bash
redis-cli -p 6380 MONITOR

输出：

# Worker 发送 BRPOP
1710000000.1 [1 127.0.0.1:54321] "brpop" "task:queue" "5"

# API 发送 LPUSH
1710000003.5 [1 127.0.0.1:54322] "lpush" "task:queue" "6"

# 注意：Worker 没有发送第二条命令
# 但 Worker 的 brpop() 返回了 ("task:queue", "6")
# 说明 Redis 没有要求 Worker 再发送 RPOP；这是挂起 BRPOP 的延迟响应

MONITOR 只显示客户端发送的命令，不显示普通命令响应。Worker 只发了一条 BRPOP 就拿到了数据，说明没有额外的 RPOP/轮询命令；从协议层面更准确地说，这是挂起 BRPOP 的延迟响应。

九、Redis 是全双工通信吗？ ​

9.1 需要区分传输层和应用层 ​

层面是否全双工说明
TCP 传输层✅ 是收发通道独立，可同时进行
Redis RESP 协议取决于功能普通命令按序请求-响应；RESP3 还支持 push 类型消息
BRPOP❌ 否延迟响应，不是推送
Pub/Sub支持服务端发送消息订阅后服务端可持续发送消息
单连接命令执行非并发执行命令按序处理；pipeline 可让多个非阻塞命令在途

9.2 TCP 传输层：全双工 ​

TCP 协议本身就是全双工的，收发可以同时进行：

Worker 进程                Redis 服务端
    │                          │
    ├── 发送数据 ────────────►│
    │                  ◄───────┤ 推送数据
    │     （两个方向可同时）     │
    │                          │

一个 TCP 连接有两个独立的通道：发送通道和接收通道。一方发送时，另一方可以同时发送回来。

9.3 Redis 应用层：不是全双工 ​

虽然 TCP 底层是全双工的，但 Redis 普通命令通常按请求顺序执行并按顺序匹配响应。这个“按序请求-响应”不应称为 TCP 半双工：

普通命令（非阻塞）:
  Client ──► 发送 GET key        ──► Redis
  Client ◄── Redis 返回 value    ◄── Redis
  ← 一问一答，客户端发完等回复

BRPOP（阻塞命令）:
  Client ──► 发送 BRPOP key 5    ──► Redis
  Client ← (等待... Redis 不回复)  ← Redis 在处理其他客户端
  Client ← (等待...)               ← Redis 在处理其他客户端
  Client ◄── Redis 返回 ("key","6") ◄── LPUSH 触发后回复
  ← 仍然是一问一答，只是"答"延迟了

9.4 BRPOP 不是"服务端主动推送" ​

第一章说"Redis 主动推送"，严格来说是延迟响应，不是真正的推送：

真正的全双工推送（如 WebSocket）:
  Client ──► 发送请求
  Client ◄── Server 回复
  Client ◄── Server 主动推送（客户端没发请求）  ← 真正的推送
  Client ◄── Server 再次主动推送              ← 没有对应的请求

Redis BRPOP:
  Client ──► 发送 BRPOP（请求）
  Client ◄── Redis 回复（延迟响应）  ← 是对 BRPOP 的回复，不是主动推送
  ← 没有新的 BRPOP 请求，Redis 不会凭空产生一个 BRPOP 响应

Redis 的服务器可以在客户端已经建立订阅或启用相关功能后发送没有对应新命令的消息，例如 Pub/Sub、RESP3 push 消息和客户端跟踪失效通知。BRPOP 返回的元素则仍然是对既有 BRPOP 请求的响应。

9.5 Redis 单连接上的命令顺序 ​

Redis 在单连接上按顺序处理命令；阻塞命令之后的命令即使已经发送到输入缓冲区，也通常要等前一个阻塞命令结束后才会处理。pipeline 可以提前发送多个非阻塞命令，但这不代表命令会并发执行：

时间轴: ─────────────────────────────────────────►

Client:  发GET─→        发BRPOP─→        收到回复
              │                │              │
Redis:        ├─返回value─→   ├─...(等待)─→  │
              │                │              │
              │                │ LPUSH到达    │
              │                ├─弹出"6"     │
              │                └─回复Client─→│

注意：Redis 回复时，Client 没有在发送命令
但这是对之前 BRPOP 请求的回复，不是主动推送
python
# 同一个连接可以在协议层面发送多条命令，但不能让后续命令绕过阻塞命令并发执行
conn.send_command("BRPOP", "task:queue", 5)
conn.send_command("GET", "some_key")  # 通常要等 BRPOP 结束后才会被处理

9.6 例外：Pub/Sub 和其他 push 消息 ​

Redis 的 Pub/Sub（发布订阅） 模式允许服务端在订阅建立后持续发送消息：

python
# 订阅频道
pubsub = redis_client.pubsub()
pubsub.subscribe("channel:notifications")

# 客户端没有发送新的发布命令，服务端可以发送订阅消息
for message in pubsub.listen():
    print(message)
    # {'type': 'message', 'channel': 'channel:notifications', 'data': 'hello'}
    # ↑ 客户端没有发请求，服务端主动推过来的
Pub/Sub 通信模型:

Client ──► SUBSCRIBE channel  ──► Redis
Client ◄── Redis: "订阅成功"
（Client 没有再发命令）
Client ◄── Redis: {"channel":"...", "data":"msg1"}  ← 主动推送
Client ◄── Redis: {"channel":"...", "data":"msg2"}  ← 再次推送
Client ◄── Redis: {"channel":"...", "data":"msg3"}  ← 持续推送

RESP2 下 Pub/Sub 消息在订阅连接上表现为持续的协议消息；RESP3 可以将这类消息表示为 push 类型。它们不改变 TCP 是全双工传输，也不意味着 Redis 命令在单连接上并发执行。

9.7 三种模式对比 ​

模式客户端发送服务端回复是否全双工
普通命令（GET/SET）1 条命令1 条回复否（一问一答）
BRPOP1 条命令1 条延迟回复否（延迟的一问一答）
Pub/Sub1 条 SUBSCRIBEN 条服务端消息支持服务端主动发送，但不是命令并发
WebSocket可随时发可随时推是（真正的全双工）

9.8 本项目的实际影响 ​

同一连接不能同时 BRPOP 和写进度：

python
# ❌ 同一个 TCP 连接
conn = redis_client.connection_pool.get_connection("BRPOP")
conn.send_command("BRPOP", "task:queue", 5)
# 这个连接现在在等 BRPOP 回复
# 在 BRPOP 返回之前，不能用这个连接发其他命令

conn.send_command("HSET", "task:6:progress", "stage", "IMAGE")
# ↑ 必须等 BRPOP 返回后才能发

当 BRPOP 和 HSET 需要并发执行时，连接池至少要能提供 max_connections >= 2：一个连接给 BRPOP（阻塞），另一个给 HSET（写进度）。如果程序是单线程顺序执行，BRPOP 返回后才写 HSET，则一个连接也可以。

9.9 Pipeline 不是全双工 ​

python
# Pipeline 是批量发送，不是全双工
pipe = redis_client.pipeline()
pipe.set("key1", "value1")
pipe.set("key2", "value2")
pipe.execute()
# 客户端一次性发两条命令，Redis 逐条执行，逐条回复
# 仍然是请求-响应，只是"批量"的
方式发送回复是否全双工
普通命令发 1 条 → 等回复1 条否
BRPOP发 1 条 → 等回复（延迟）1 条否
Pipeline发 N 条 → 一次收 N 条回复N 条否（批量请求-响应）
Pub/Sub发 1 条 SUBSCRIBEN 条服务端消息支持服务端主动发送
WebSocket随时发随时推是

十、术语精确化 ​

10.1 "推送"一词的三种含义 ​

在前文和本文中，"推送"一词可能指不同层面，需要精确区分：

层面"推送"含义是否真正推送
TCP 层Redis 通过 socket fd 发送字节TCP 是全双工，但不等于应用层消息是无请求的
RESP 协议层BRPOP 收到对既有请求的延迟响应；RESP3 还支持 push 消息BRPOP 不是无请求 push
应用层语义Worker 没有主动发"拉取"命令就拿到了数据可以称为“服务端推动”，但应注明是延迟响应

10.2 BRPOP 的精确描述 ​

不准确的说法:
  "Redis 主动推送数据给 Worker"
  → 暗示 Redis 可以随时给客户端发数据

准确的描述:
  "Redis 对 Worker 的 BRPOP 请求进行延迟响应"
  → Worker 先发了 BRPOP 请求
  → Redis 暂时不回复（加入等待链表）
  → key 就绪后，Redis 处理并回复之前挂起的 BRPOP 请求
  → 这是请求-响应模型，只是响应被延迟了

10.3 为什么前文仍然用"推送" ​

从前文和业务视角看，"推送"是一个合理的语义描述：

  • Worker 没有发"拉取"命令（没有发 RPOP）
  • Worker 只发了一条 BRPOP，然后数据"自己来了"
  • 和"客户端拉取"模型对比，确实是"推送"语义

但从严格的网络协议角度，应该理解为延迟响应。

十一、相关文档索引 ​

文档路径内容
BRPOP 深度解析Redis-BRPOP深度解析.mdBRPOP 语法、阻塞机制、多 Worker 行为
单线程与多客户端交互Redis单线程与多客户端交互.md多进程客户端 → 单线程服务端、连接池
原子性保证机制Redis原子性保证机制.md四层保证、等待链表源码、LPUSH 唤醒流程
技术问答Redis技术问答.md25 道技术问答，覆盖三篇文档全部知识点