Redis 技术笔记

主题

七、阻塞期间的连接管理 ​

7.1 redis-py 的连接占用 ​

python
result = redis_client.brpop("task:queue", timeout=5)
# ← 这 5 秒内，这个连接被独占

BRPOP 期间，该 Redis 连接被独占，不能用来执行其他命令。如果同时需要查 Redis 进度，需要另一个连接：

python
# ❌ 错误：同一个连接
redis_client.brpop("task:queue", timeout=60)  # 阻塞 60 秒
redis_client.get("some_key")                   # 无法执行，连接被占

# ✅ 正确：连接池自动分配不同连接
# redis-py 的 ConnectionPool 会在 get() 时借用另一个连接
# 但如果 pool_size=1 且连接被 BRPOP 占用，get() 会等待

get(key) 参数说明 ​

参数类型必填说明
keystr是要读取的键名

返回值：str | bytes | None

返回值含义
"50%"键存在，返回对应的值
None键不存在

用途：在本项目中用于读取 Redis Hash 中的任务进度（task:{id}:progress）。

与 brpop 的区别：get 是非阻塞命令，立即返回，不等待。

set(key, value, ex=None) 参数说明 ​

参数类型必填说明
keystr是要设置的键名
valuestr 或 int 或 bytes是要存储的值
exint否过期时间（秒）。不传则永不过期。设置后到时间自动删除

返回值：True（总是成功，除非 Redis 故障）

示例：

python
# 设置进度，永不过期
redis_client.set("task:6:progress", "50%")

# 设置进度，60 秒后自动过期
redis_client.set("task:6:progress", "50%", ex=60)

ex 参数的作用：避免 Redis 中堆积大量过期的进度数据。任务完成后进度键不再需要，设置过期时间让 Redis 自动清理。

redis.Redis(...) 连接参数说明 ​

建议：Worker 的 Redis 连接池 max_connections 至少设为 2（一个给 BRPOP，一个给进度写入）。

python
# redis_client.py
import redis

redis_client = redis.Redis(
    host="localhost",
    port=6379,
    db=0,
    decode_responses=True,
    max_connections=10,
    socket_timeout=30,
    socket_connect_timeout=5,
    health_check_interval=30,
    retry_on_timeout=True,
)
参数类型默认值说明
hoststr"localhost"Redis 服务器地址。本项目用 "localhost"（Docker 映射）
portint6379Redis 端口。本项目 Docker 映射为 6380
dbint0Redis 数据库编号（0-15）。本项目用 0
decode_responsesboolFalseTrue 时所有返回值自动 decode("utf-8") 为 str；False 时返回 bytes，需手动 .decode()
max_connectionsint2**31连接池上限。超过时抛 ConnectionError: Too many connections
socket_timeoutintNone单次命令的 socket 超时（秒）。超时抛 TimeoutError。推荐 30 秒
socket_connect_timeoutintNone建立连接的超时（秒）。推荐 5 秒，避免 Redis 不可用时长时间卡住
health_check_intervalint0健康检查间隔（秒）。30 表示每 30 秒自动发 PING 检测连接是否存活。0 = 不检查
retry_on_timeoutboolFalseTrue 时遇到 TimeoutError 自动重试一次。适合网络不稳定环境

decode_responses 对返回值的影响：

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

result = redis_client.get("task:6:progress")
# result = b"50%"                   ← bytes

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

result = redis_client.get("task:6:progress")
# result = "50%"                    ← str，直接可用

socket_timeout 与 brpop 的关系：

brpop(timeout=5)       → BRPOP 最多阻塞 5 秒
socket_timeout=30      → socket 层面最多等 30 秒

如果 Redis 宕机:
  brpop 的 5 秒超时不会触发（Redis 没有返回）
  socket_timeout=30 在 30 秒后抛 TimeoutError

所以 socket_timeout 必须 > brpop 的 timeout，否则 BRPOP 还没超时 socket 就报错了

推荐配置：socket_timeout=30，brpop(timeout=5)，确保 BRPOP 能正常超时 6 次后才触发 socket 超时。

7.2 超长阻塞的风险 ​

python
# ❌ timeout=0 永久阻塞
result = redis_client.brpop("task:queue", timeout=0)

风险：

  • Ctrl+C 无法中断（需要强制 kill）
  • 如果 Redis 服务重启，客户端可能不感知（连接假死）
  • 无法定期检查其他信号（如优雅退出标志）

推荐：始终用 timeout=5（或不超过 30 秒），在 while True 循环里重新调用。

7.3 Worker 进程与 Redis 连接池的对应关系 ​

7.3.1 核心结论 ​

层面关系说明
进程级1 个 Worker 进程 = 1 个 redis_client = 1 个 ConnectionPool每个 python -m app.worker 是独立进程，各自 import 模块时创建单例
连接级1 个 ConnectionPool 内部复用多个 TCP 连接brpop 阻塞期间独占 1 个连接，其他命令用其他连接
Redis 端1 个 TCP 连接 = 1 个 client = 等待链表中 1 个节点Redis 不关心客户端是进程还是线程，只看 TCP 连接

7.3.2 多进程部署模型 ​

本项目通过启动多个 python -m app.worker 进程实现多 Worker：

bash
# 启动 3 个独立 Worker 进程
.venv/bin/python -m app.worker &   # 进程 A
.venv/bin/python -m app.worker &   # 进程 B
.venv/bin/python -m app.worker &   # 进程 C

每个进程独立 import redis_client 模块，各自创建 redis_client 单例：

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 入口
from app.redis_client import redis_client  # ← 每个进程各自 import

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

3 个进程的内存结构：

进程 A (PID 1001)                    进程 B (PID 1002)                    进程 C (PID 1003)
┌─────────────────────┐             ┌─────────────────────┐             ┌─────────────────────┐
│ redis_client        │             │ redis_client        │             │ redis_client        │
│  └─ ConnectionPool  │             │  └─ ConnectionPool  │             │  └─ ConnectionPool  │
│      max=10         │             │      max=10         │             │      max=10         │
│  ┌─────┬─────┬───┐ │             │  ┌─────┬─────┬───┐ │             │  ┌─────┬─────┬───┐ │
│  │C-1  │C-2  │...│ │             │  │C-1  │C-2  │...│ │             │  │C-1  │C-2  │...│ │
│  │BRPOP│空闲 │   │ │             │  │BRPOP│空闲 │   │ │             │  │BRPOP│空闲 │   │ │
│  └─────┴─────┴───┘ │             │  └─────┴─────┴───┘ │             │  └─────┴─────┴───┘ │
└────────┬────────────┘            └────────┬────────────┘            └────────┬────────────┘
         │ TCP                              │ TCP                              │ TCP
         ▼                                  ▼                                  ▼
    127.0.0.1:54321                   127.0.0.1:54322                   127.0.0.1:54323
         │                                  │                                  │
         ▼                                  ▼                                  ▼
┌─────────────────────────────────────────────────────────────────────────────────────────────┐
│ Redis 服务端                                                                                │
│  client(fd=10, cmd=brpop)  ← 等待链表节点 1                                  │
│  client(fd=11, cmd=brpop)  ← 等待链表节点 2                                  │
│  client(fd=12, cmd=brpop)  ← 等待链表节点 3                                  │
│                                                                                             │
│  task:queue → waiting_clients: [client(fd=10), client(fd=11), client(fd=12)]              │
└─────────────────────────────────────────────────────────────────────────────────────────────┘

关键：3 个进程各自独立的连接池，互不共享。Redis 端看到 3 个 TCP 连接，等待链表有 3 个节点。

7.3.3 单进程内的连接复用 ​

同一个 Worker 进程内，redis_client 是单例，所有 Redis 调用共用一个连接池：

python
# 同一个 Worker 进程内的不同阶段
def run_task(task_id: int):
    # 阶段1：写进度（借 Conn-2）
    redis_client.hset(f"task:{task_id}:progress", "stage", "SCRIPT")

    # 阶段2：写进度（复用 Conn-2，Conn-1 仍被 BRPOP 独占）
    redis_client.hset(f"task:{task_id}:progress", "stage", "IMAGE")

    # 阶段3：完成后推入审核队列（复用 Conn-2）
    redis_client.lpush("review:queue", task_id)
Worker 进程内部:
  redis_client (单例)
    └── ConnectionPool (max=10)
          ├── Conn-1 ←── brpop() 阻塞中（从 process_queue 开始就独占）
          ├── Conn-2 ←── hset / lpush 轮流借用（用完归还）
          └── Conn-3~10 ←── 空闲

brpop 独占 Conn-1 直到超时或拿到任务。期间 hset、lpush 等命令从池中借 Conn-2，用完归还。不互相阻塞。

7.3.4 连接借还生命周期 ​

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

python
# redis-py 内部简化逻辑
class ConnectionPool:
    def __init__(self, max_connections):
        self._available = []      # 空闲连接栈
        self._in_use = set()     # 正在使用的连接
        self._max = max_connections

    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)   # 归还到空闲栈

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

python
# redis_client.brpop("task:queue", timeout=5)
# 内部流程:
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
# redis_client.hset("task:6:progress", "stage", "IMAGE")
# 内部流程:
conn = pool.get_connection()       # 借出 Conn-2
conn.send_command("HSET", "task:6:progress", "stage", "IMAGE")
result = conn.read_response()       # 立即返回
pool.release(conn)                  # 立即归还 Conn-2

7.3.5 连接数计算 ​

单个 Worker 进程的连接数：

python
# Worker 运行时的连接使用:
redis_client.brpop("task:queue", timeout=5)  # Conn-1: 阻塞 5 秒
                                               # Conn-1 超时归还，循环重调

# 任务处理中:
redis_client.hset(...)                        # Conn-2: 借→用→还
redis_client.hset(...)                        # Conn-2: 借→用→还（复用）
redis_client.lpush(...)                        # Conn-2: 借→用→还（复用）

# 同一时刻最多占用 2 个连接:
#   - 1 个被 brpop 独占（但 brpop 在任务处理期间不会执行，因为 process_queue 是串行的）
#   - 1 个被当前命令临时使用

实际同时占用的连接数：

阶段占用连接说明
process_queue 中 BRPOP 阻塞1 个只有 BRPOP 在执行
run_task 中写进度1 个BRPOP 已归还，HSET 临时借
run_task 中推审核队列1 个同上
理论峰值1-2 个BRPOP 和其他命令不会同时执行

多 Worker 进程的总连接数：

Worker 进程数每进程连接峰值总 TCP 连接数Redis max_clients
122默认 10000
326足够
10220足够
502100足够
2002400足够

7.3.6 fork 场景的陷阱 ​

如果用 multiprocessing.Process 的 fork 模式（Unix 默认）启动 Worker：

python
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!

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

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

python
import multiprocessing
import redis

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

p = multiprocessing.Process(target=worker)
p.start()

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

7.3.7 对比三种部署模式 ​

python
# 模式 1：多进程（本项目使用）
# 每个 python -m app.worker 是独立进程
import redis

redis_client = redis.Redis(
    host="localhost", port=6380,
    decode_responses=True, max_connections=10,
)
# 进程 A: redis_client_A, Pool_A, TCP_A
# 进程 B: redis_client_B, Pool_B, TCP_B
# 进程 C: redis_client_C, Pool_C, TCP_C
# → 3 个独立连接池，3 个 TCP 连接

# 模式 2：单进程多线程
import threading

def worker():
    # 所有线程共享同一个 redis_client 和连接池
    redis_client.brpop("task:queue", timeout=5)
    # 线程 1: 借 Conn-1
    # 线程 2: 借 Conn-2
    # 线程 3: 借 Conn-3
    # → 1 个连接池，3 个 TCP 连接

t1 = threading.Thread(target=worker)
t2 = threading.Thread(target=worker)
t3 = threading.Thread(target=worker)

# 模式 3：多进程 spawn（推荐）
import multiprocessing

def worker():
    import redis
    # 子进程内部创建新连接池
    client = redis.Redis(host="localhost", port=6380)
    client.brpop("task:queue", timeout=5)

if __name__ == "__main__":
    for _ in range(3):
        p = multiprocessing.Process(target=worker)
        p.start()
模式redis_client 数量连接池数量TCP 连接数隔离性fork 风险
多进程（python -m）333-6完全隔离无
单进程多线程113-10共享池无
多进程 fork继承父进程继承fd 冲突不隔离有
多进程 spawn333-6完全隔离无

7.3.8 本项目的完整连接拓扑 ​

3 个 Worker 进程（python -m app.worker）

进程 A ──import──► redis_client ──► ConnectionPool(max=10)
                     │                   ├── Conn-1 (BRPOP 独占)
                     │                   └── Conn-2~10 (空闲)
                     │
                     └──TCP──► Redis fd=10 ──► waiting_clients[0]

进程 B ──import──► redis_client ──► ConnectionPool(max=10)
                     │                   ├── Conn-1 (BRPOP 独占)
                     │                   └── Conn-2~10 (空闲)
                     │
                     └──TCP──► Redis fd=11 ──► waiting_clients[1]

进程 C ──import──► redis_client ──► ConnectionPool(max=10)
                     │                   ├── Conn-1 (BRPOP 独占)
                     │                   └── Conn-2~10 (空闲)
                     │
                     └──TCP──► Redis fd=12 ──► waiting_clients[2]

Redis 等待链表:
  task:queue → [client(fd=10), client(fd=11), client(fd=12)]
                 ↑ 进程A的连接    ↑ 进程B的连接    ↑ 进程C的连接

LPUSH task:queue 6:
  → Redis 弹出 client(fd=10) → 通过 fd=10 发送 "6" → 进程A收到
  → fd=10 从等待链表移除
  → 进程A 的 brpop() 返回 → 开始处理任务
  → 进程B、C 继续等待

一句话总结：每个 Worker 进程拥有独立的 redis_client 和连接池，进程内复用连接池中的多个连接，进程间互不共享。Redis 端每个 TCP 连接对应等待链表中的一个节点。