Redis Python 示例执行说明

本目录下的所有脚本用于演示 Redis 各数据结构的典型用法，均通过 redis_connection.py 统一建立连接、从 .env 加载配置，运行时无需任何命令行参数。

运行前准备

1. 安装依赖

python3 -m pip install -r requirements.txt

2. 配置 .env

确认本目录下的 .env 已正确填写 Redis 连接信息：

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

各字段说明：

字段	含义	默认值
REDIS_HOST	Redis 地址	127.0.0.1
REDIS_PORT	Redis 端口	6379
REDIS_DB	数据库编号	0
REDIS_PASSWORD	连接密码	无（必填）

说明：脚本使用 python-dotenv 在启动时自动加载同目录下的 .env 文件，无需手动 export 环境变量。

脚本执行命令

所有脚本均在本目录下直接运行，常见 Redis 数据结构的通用命令如下：

脚本	功能说明	执行命令
redis_connection.py	公共连接模块，封装连接与运行逻辑	无需单独运行
string_cache_and_lock.py	String：缓存、计数器、分布式锁（NX/EX）	python3 string_cache_and_lock.py
hash_progress.py	Hash：任务进度字段的读写与自增	python3 hash_progress.py
list_queue_and_latest.py	List：队列 FIFO 与最新列表（LPUSH/BRPOP/LRANGE）	python3 list_queue_and_latest.py
brpop_worker_demo.py	List：BRPOP 阻塞消费，体现 LPUSH 唤醒队头	python3 brpop_worker_demo.py
set_dedup_and_relation.py	Set：去重、关系交集（SADD/SISMEMBER/SINTER）	python3 set_dedup_and_relation.py
zset_ranking_and_delay.py	ZSet：排行榜与延迟任务（ZADD/ZREVRANGE/ZRANGEBYSCORE）	python3 zset_ranking_and_delay.py
stream_consumer_group.py	Stream：消费者组（XGROUP/XADD/XREADGROUP/XACK）	python3 stream_consumer_group.py
bitmap_signin.py	Bitmap：签到统计（SETBIT/GETBIT/BITCOUNT）	python3 bitmap_signin.py
hll_uv.py	HyperLogLog：UV 去重估算与合并（PFADD/PFCOUNT/PFMERGE）	python3 hll_uv.py
geo_nearby_search.py	Geo：附近门店检索与距离（GEOADD/GEOSEARCH/GEODIST）	python3 geo_nearby_search.py
pubsub_broadcast.py	Pub/Sub：配置变更广播（SUBSCRIBE/PUBLISH）	python3 pubsub_broadcast.py
connection_pool_demo.py	连接池：BlockingConnectionPool 复用与参数	python3 connection_pool_demo.py
atomicity_demo.py	原子性：单命令原子、多命令非原子的对比	python3 atomicity_demo.py
redisjson_document.py	RedisJSON：文档存读改（JSON.SET/GET/NUMINCRBY，需模块）	python3 redisjson_document.py
probabilistic_filters.py	概率数据结构：Bloom/Cuckoo/CMS/TopK/TDigest（需模块）	python3 probabilistic_filters.py
search_vector_timeseries.py	检索与时间序列：RediSearch/TimeSeries（需模块）	python3 search_vector_timeseries.py

说明：脚本统一使用 demo:* 前缀存放演示数据，每次运行前会自动清空本脚本用到的 key，不会影响你的真实业务数据。

示例

以下是一个典型脚本的执行演示：

cd /Users/binny/workspace-sdd/aiNotes/redis/python示例

# 执行 String 缓存与分布式锁示例
python3 string_cache_and_lock.py

预期输出（结构示意）：

=== String：缓存、计数器、分布式锁 ===
PING -> True
RUNNING
1
True
abc123

redis_connection.py 的 run() 会打印标题、PING 结果，以及脚本本身的输出；若连接失败则会打印 Redis 执行失败：...。

补充说明

关于需要额外模块的脚本

以下脚本依赖 Redis 官方扩展模块，若你的 Redis 实例未安装对应模块，脚本会自动捕获异常并打印「模块不可用或命令不支持」，不会中断整个演示：

• redisjson_document.py → RedisJSON
• probabilistic_filters.py → RedisBloom
• search_vector_timeseries.py → RediSearch + RedisTimeSeries

关于连接池复用

redis_connection.py 中的 Redis 客户端默认使用连接池；connection_pool_demo.py 演示了显式使用 BlockingConnectionPool 的方式。

关于 .env 安全

.env 中存放了密码，已被 .gitignore 忽略，请勿提交到版本库。

---

所有脚本详细说明

以下按脚本逐个讲解：做了什么、用了哪些命令、关键设计点、以及输出的含义。每个脚本都会先用 demo:*: 前缀清理自己用到的 key，避免污染真实数据。

1. redis_connection.py（公共连接模块）

这是所有示例的「地基」，其它脚本都从它导入 get_client / run / flush_demo_keys。

• get_client(decode_responses=True)：
  • 通过 python-dotenv 读取同目录 .env，取 REDIS_HOST / REDIS_PORT / REDIS_DB / REDIS_PASSWORD；
  • 密码为空时抛 RuntimeError，提示先配置 .env；
  • 返回 redis.Redis 实例，decode_responses=True 表示返回值自动从 bytes 解码为 str，脚本里才能直接打印中文和普通字符串；
  • 设置了 socket_connect_timeout=3、socket_timeout=5、health_check_interval=30，避免断连后长期不可用。
• flush_demo_keys(client, *patterns)：用 SCAN 遍历匹配 demo:* 的 key 并删除（scan_iter 而非 keys，避免阻塞 Redis）。
• run(title, func)：统一的执行外壳——打印标题、PING 探活、调用 func(client)，并用 try/except 捕获 RedisError 和 RuntimeError，打印友好错误。这就是为什么连接失败时脚本不崩，而是打印「Redis 执行失败」的原因。

关键点：decode_responses 默认开启，所以所有示例的返回值都是可读字符串；若需要原始字节，可传 False。

2. string_cache_and_lock.py（String：缓存 / 计数器 / 分布式锁）

client.set('demo:string:task:6:status', 'RUNNING', ex=900)   # 缓存：带 900 秒过期
client.get('demo:string:task:6:status')                       # 读取缓存
client.incr('demo:string:video:6:view_count')                 # 计数器：原子自增
client.set('demo:lock:task:6', 'abc123', nx=True, ex=300)     # 分布式锁：NX 只在不存在的时写
client.get('demo:lock:task:6')

• 缓存：set 的 ex=900 表示 15 分钟后自动过期，典型用于缓存视频生成状态。
• 计数器：INCR 单命令原子，适合点赞/观看量，无需「读改写」三步。
• 分布式锁：nx=True（SET 带 NX，只有 key 不存在才成功）+ ex=300（过期自动释放，防止死锁）。返回值 True 表示抢锁成功，False 表示已被别人持有。

输出顺序：RUNNING → 自增后的数字（首次为 1）→ 锁获取结果 True → 锁的值 abc123。

3. hash_progress.py（Hash：任务进度）

client.hset(key, mapping={'stage':'IMAGE', 'percent':'45', 'message':'...'})  # 一次写多个字段
client.hget(key, 'stage')        # 取单个字段
client.hgetall(key)              # 取全部字段
client.hincrby(key, 'percent', 10)  # 数值字段原子自增

Hash 用「一个 key + 多个 field」存储任务进度，比多个 String key 更紧凑。HINCRBY 是 INCRBY 的 Hash 版，直接给 percent 加 10，无需先读再写，避免并发覆盖。

4. list_queue_and_latest.py（List：队列 + 最新列表）

client.lpush(queue_key, 'task-6')          # 从左推入
client.brpop(queue_key, timeout=5)          # 从右阻塞弹出（队头先出）
client.lpush(feed_key, 'video-1', 'video-2', 'video-3')  # 一次推入多个
client.lrange(feed_key, 0, 19)              # 取前 20 条

List 是双向队列，LPUSH + BRPOP 组合形成「先进先出」；LPUSH + LRANGE 0 19 则实现「最新 N 条」的动态流（最新加入的在最左侧）。

5. brpop_worker_demo.py（List：BRPOP 阻塞消费 / 唤醒队头）

def worker_loop():
    worker_client = get_client()                 # 每个 worker 独立连接
    result_holder.append(worker_client.brpop(queue_key, timeout=5))

thread.start()          # 启动 worker 线程，阻塞等队列
time.sleep(0.5)         # 确保 worker 已进入 BRPOP 等待
client.lpush(queue_key, 'task-6')  # 主线程推入一个元素
thread.join(timeout=6)  # 等待 worker 被唤醒并消费

这是整个 Redis 技术问答里「BRPOP 阻塞 + LPUSH 唤醒队头」的最小可运行复现：

• worker 线程调用 BRPOP 后阻塞，同时向 Redis 服务端登记到一个「等待链表」；
• 主线程 LPUSH 后，Redis 只唤醒等待链表队头的那个 waiter，把元素交给它；
• 所以这里的输出是 [('task-6', ...)]，即 worker 成功拿到了 task-6。

与「轮询 RPOP + sleep」的对比：BRPOP 由服务端在数据到达时主动唤醒，无空轮询、无额外延迟。

6. set_dedup_and_relation.py（Set：去重 + 关系交集）

client.sadd(like_key, '1001', '1002', '1003', '1003')  # 去重：重复的 1003 只写一次
client.sismember(like_key, '1001')                       # 判断成员是否存在
client.sadd(follow_a, '2001', '2002', '2003')
client.sadd(follow_b, '2002', '2004')
client.sinter(follow_a, follow_b)                        # 交集 = 共同关注

• SADD 返回值是「实际新增的元素个数」，重复加 1003 只算一次，体现 Set 天然去重；
• SINTER 求两个关注集合的交集，得到共同关注（2002），是「你可能认识的人 / 共同兴趣」的典型算法基础。

7. zset_ranking_and_delay.py（ZSet：排行榜 + 延迟任务）

client.zadd(rank_key, {'video:6':100, 'video:8':98, 'video:3':95})  # score 为热度分
client.zrevrange(rank_key, 0, 9, withscores=True)   # 按分数从高到低取前十
now = int(time.time())
client.zadd(delay_key, {'task:6': now-1, 'task:7': now+60})  # score = 执行时间戳
client.zrangebyscore(delay_key, 0, now)   # 只取到期任务（score <= 当前时间）

• 排行榜：score 存分数，ZREVRANGE 高到低取 Top N；
• 延迟任务：score 存「到期时间戳」，ZRANGEBYSCORE 0 now 只捞出已到期任务，是 Redis 实现延迟队列的经典做法。

8. stream_consumer_group.py（Stream：消费者组）

client.xgroup_create(stream_key, group_name, id='0', mkstream=True)   # 建组
client.xadd(stream_key, {'type':'video_done','task_id':'6'})           # 写消息
client.xreadgroup(group_name, 'worker-1', {stream_key:'>'}, count=10, block=1000)  # 消费者读新消息
client.xack(stream_key, group_name, *ids)   # 确认消费

• Stream 是 Redis 5.0 后的「消息队列」，> 表示只读组内「尚未分配给任何消费者的新消息」；
• 消费者组保证同一条消息不会被组内多个消费者重复读取（读到后进入 pending 待确认）；
• XACK 确认后消息才算真正处理完，配合 XPENDING 可做「未确认消息重试」。

脚本对 XGROUP CREATE 的 BUSYGROUP 做了容错：组已存在时跳过创建，便于重复运行。

9. bitmap_signin.py（Bitmap：签到统计）

client.setbit(key, 26, 1)      # 把第 26 位（8 月 27 日）置 1
client.getbit(key, 26)         # 查该位
client.bitcount(key)           # 统计整月签到天数

Bitmap 用一个二进制位表示「某天是否签到」，一个月的签到只需 31 bit（约 4 字节），极其省内存。BITCOUNT 直接统计置 1 的位数即签到总天数。适合「活跃用户 / 日签到 / 去重标记」等 0/1 布尔统计。

10. hll_uv.py（HyperLogLog：UV 去重估算）

client.pfadd(key, '1001','1002','1003','1003')  # 统计 UV，重复 1003 不重复计数
client.pfcount(key)                              # 估出去重后的数量（约 3）
client.pfadd('demo:hll:uv:day:1', '1001','1002') # 第 1 天 UV
client.pfadd('demo:hll:uv:day:2', '1002','1003') # 第 2 天 UV
client.pfmerge(merged_key, 'demo:hll:uv:day:1','demo:hll:uv:day:2')  # 合并多天
client.pfcount(merged_key)                       # 合并后去重 UV（约 3，跨天去重 1002）

• HyperLogLog 用极小的固定内存（约 12 KB）就能估算海量集合的基数（去重个数），误差约 0.81%；
• 关键是 PFMERGE 可以合并多天 HLL，算出「跨天去重后的总 UV」，这是普通 Set 很难低成本做到的大规模 UV 统计。

11. geo_nearby_search.py（Geo：附近检索）

client.geoadd(key, (116.397128, 39.916527, 'store:1', 116.407128, 39.926527, 'store:2'))
client.geosearch(key, longitude=116.4, latitude=39.9, radius=5, unit='km', withdist=True)
client.geodist(key, 'store:1', 'store:2', unit='km')

• GEOADD 存经纬度，Redis 内部用 ZSet + Geohash 实现；
• GEOSEARCH 按「圆心 + 半径」找附近门店，withdist=True 返回距离；适合「附近门店 / 骑手 / LBS 距离排序」。

12. pubsub_broadcast.py（Pub/Sub：广播）

pubsub = sub_client.pubsub()          # 建立订阅对象
pubsub.subscribe(channel)             # 订阅频道
for message in pubsub.listen():       # 阻塞监听
    if message['type'] == 'message':
        received.append(message['data']); pubsub.close(); break

client.publish(channel, 'video:max_workers')  # 发布消息

• 子线程先订阅，主线程再发布，展示「发布→订阅者即时收到」；
• Pub/Sub 是广播语义：所有订阅该频道的客户端都会收到，且不落盘、不重投——订阅者不在线就会错过；需要可靠投递应改用 Stream。

13. connection_pool_demo.py（连接池）

pool = BlockingConnectionPool(max_connections=10, timeout=3, **kwargs)
pooled_client = Redis(connection_pool=pool, decode_responses=True)
pooled_client.ping()
pool.max_connections          # 10

• 显式创建 BlockingConnectionPool，最大连接数 10，获取超时 3 秒；
• 连接池让多次命令复用同一组 TCP 连接，避免每次新建连接的开销；
• **kwargs 从 redis_connection 的默认连接池继承 host/port/password 等，保证与环境一致。

14. atomicity_demo.py（原子性：单命令原子 vs 多命令非原子）

def worker(name):
    worker_client = get_client()
    item = worker_client.brpop(queue_key, timeout=5)   # 单命令原子弹出
    results.append((name, item))

# 3 个 worker 并发 BRPOP，再 LPUSH 3 个任务
for task_id in ['task-1','task-2','task-3']:
    client.lpush(queue_key, task_id)

for _ in range(10):
    client.incr(counter_key)   # 10 次 INCR，单命令原子，最终必为 10

• BRPOP（单命令）是原子的：3 个 worker 并发等待，3 个任务 LPUSH 后，每个任务只会被一个 worker 拿走，不会重复消费；
• INCR（单命令）也是原子的：循环 10 次，最终计数必然等于 10，不会出现「读改写」竞态导致的少加；
• 脚本用 sorted(results) 输出，说明结果与线程执行顺序无关，只保证「每个任务恰好被消费一次」。

这个脚本直接对应技术问答中「为什么多 Worker 下不会重复消费」「单命令为什么天然原子」两个问题。

15. redisjson_document.py（RedisJSON：文档存读改，需模块）

client.execute_command('JSON.SET', key, '$', '{"model":"seedance-2.0","ratio":"16:9","retry":0}')
client.execute_command('JSON.GET', key, '$')                       # 取整个文档
client.execute_command('JSON.NUMINCRBY', key, '$.retry', 1)        # 对 $.retry 原子自增

• 用 execute_command 调用原生 JSON.* 命令（需 Redis 安装 RedisJSON 模块）；
• $ 是 JsonPath 根路径，$.retry 定位到具体字段并做原子自增；
• 若模块缺失，ResponseError 被捕获并打印「模块不可用」，脚本不崩。

16. probabilistic_filters.py（概率数据结构，需 RedisBloom）

('BF.ADD', key, 'goods:1001')      # 布隆过滤器：判断元素「可能存在」
('BF.EXISTS', key, 'goods:1002')
('CF.ADD', key, 'task:6')          # Cuckoo 过滤器：可删除的布隆
('CMS.INCRBY', key, 'redis', 1)    # Count-Min Sketch：高频统计
('TOPK.ADD', key, 'video:8','video:6')  # Top-K：热门榜
('TDIGEST.ADD', key, 2.1, 2.4, 3.0)     # T-Digest：分位数

这些是 RedisBloom 模块提供的「概率数据结构」，用可控的内存和误差换取海量数据的近似统计：布隆/Cuckoo 做「去重 + 存在性」，CMS/Top-K 做「频次/热度」，T-Digest 做「分位数」。脚本用统一的 execute_optional 容错，缺模块时逐条打印跳过。

17. search_vector_timeseries.py（检索 + 时间序列，需模块）

client.hset('demo:search:task:6', mapping={'title':'海边视频生成','stage':'IMAGE','user_id':'1001'})
('FT.CREATE', 'demo:idx:tasks', 'ON','HASH','PREFIX','1','demo:search:task:',
 'SCHEMA','title','TEXT','stage','TAG','user_id','NUMERIC')   # 建索引
('FT.SEARCH', 'demo:idx:tasks', '@stage:{IMAGE} 海边', 'DIALECT','2')  # 全文+过滤检索
('TS.ADD', 'demo:ts:queue:task:length', '*', 12)   # 记录时序点
('TS.RANGE', 'demo:ts:queue:task:length', '-', '+')  # 查时间范围

• RediSearch：FT.CREATE 建立全文/标签/数值索引，FT.SEARCH 做「stage 为 IMAGE 且标题含 海边」的检索；
• TimeSeries：TS.ADD 写时间戳 *（自动当前时间）+ 值 12，TS.RANGE 查区间，用于队列长度等指标监控；
• 两个模块都通过 optional 容错，缺模块时打印跳过。

---

运行顺序建议

所有脚本彼此独立，可任意顺序运行；但建议先跑基础数据结构（2~9、11），再跑需要模块的（15~17）：

# 基础数据结构
python3 string_cache_and_lock.py
python3 hash_progress.py
python3 list_queue_and_latest.py
python3 brpop_worker_demo.py
python3 set_dedup_and_relation.py
python3 zset_ranking_and_delay.py
python3 stream_consumer_group.py
python3 bitmap_signin.py
python3 hll_uv.py
python3 geo_nearby_search.py
python3 pubsub_broadcast.py
python3 connection_pool_demo.py
python3 atomicity_demo.py

# 需要额外模块（无模块时自动降级提示）
python3 redisjson_document.py
python3 probabilistic_filters.py
python3 search_vector_timeseries.py