Redis 技术笔记

主题

第十一章：Python 可运行案例 ​

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

答：

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

运行前先安装依赖：

bash
python3 -m pip install -r /Users/binny/workspace-sdd/aiNotes/redis/python示例/requirements.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)
HyperLogLogUV 去重估算[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 或对应模块，就会真实执行命令。

示例依赖能力示例文件
RedisJSONRedisJSON 模块[redisjson_document.py](file:///Users/binny/workspace-sdd/aiNotes/redis/python示例/redisjson_document.py)
Bloom / Cuckoo / CMS / TopK / TDigestRedisBloom 模块[probabilistic_filters.py](file:///Users/binny/workspace-sdd/aiNotes/redis/python示例/probabilistic_filters.py)
FT.SEARCH / TimeSeriesRediSearch、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