企业官网建设流程全解析

如何用 redis-py 订阅并解析 keyspace notifications 监听键变更事件

【免费下载链接】redis-pyRedis Python client项目地址: https://gitcode.com/GitHub_Trending/re/redis-py

当需要感知 Redis 中键的创建、修改和删除（比如做缓存失效或审计）时，可以直接轮询，也可以用 Redis 原生的 keyspace notifications。redis-py 在 redis/keyspace_notifications.py 中封装了这一能力：KeyspaceNotifications（单机）和ClusterKeyspaceNotifications（集群）负责订阅 Pub/Sub 通道，并把原始消息解析成带key、event_type、subkeys等字段的KeyNotification对象。本文以单机 Redis 为主路径演示"订阅 → 解析 → 验证收到事件"的完整流程，集群模式和 subkey 级通知作为可选分支。

前提条件：服务端开启通知

keyspace notifications 由 Redis 服务端控制，默认不发送任何事件。redis-py 的文档明确要求：必须通过服务端的notify-keyspace-events配置项开启（见 模块说明），这是基础设施侧的配置。

redis-py 的测试夹具 tests/conftest.py 给出了两种可核对的开启方式：

r = Redis() # 记录原值，便于测完恢复 original = r.config_get("notify-keyspace-events").get("notify-keyspace-events") # KEA：同时开启 keyspace(K)、keyevent(E) 和所有事件类型(A) r.config_set("notify-keyspace-events", "KEA") # ... 测试完成后恢复原配置 # r.config_set("notify-keyspace-events", original)

两点需要注意：

• 使用config set会立即修改运行中服务端的配置，测试结束后应按上面注释恢复原值（这正是 conftest 夹具的做法）。
• 如果后面要用 subkey 级通知（KEASTIV中的 S/T/I/V 标志），测试文件注明这些标志仅存在于 Redis >= 8.7.2（见 tests/conftest.py）。

创建通知管理器

对单机Redis客户端，可以直接用客户端上的便捷方法（redis/client.py），也可以显式构造：

from redis import Redis from redis.keyspace_notifications import ( KeyspaceChannel, KeyeventChannel, EventType, ) r = Redis() ksn = r.keyspace_notifications() # 等价于 KeyspaceNotifications(r)

keyspace_notifications()支持key_prefix参数（只接收以该前缀开头的键的通知，并把前缀从key中剥离），以及ignore_subscribe_messages（默认 True，订阅确认消息不会进入get_message/listen的结果）。

订阅通道：按键或按事件

通道类会自动生成带__keyspace@<db>__:前缀的完整通道名，并自动识别通配符：含*、?、[...]的走psubscribe，精确键名走subscribe（自动检测逻辑见 subscribe 实现）。

# 方式一：监听某个键或一组键的变更（keyspace，消息体是事件类型） ksn.subscribe(KeyspaceChannel("user:123")) # 精确键 → __keyspace@0__:user:123 ksn.subscribe(KeyspaceChannel("user:*")) # 通配 → psubscribe __keyspace@0__:user:* # 方式二：监听某类事件在所有键上发生（keyevent，消息体是键名） ksn.subscribe_keyevent(EventType.SET) # → __keyevent@0__:set # 方式三：用便捷方法按键订阅 ksn.subscribe_keyspace("session:*")

常用事件类型常量集中在EventType类（定义位置），如SET、DEL、EXPIRED、HSET等；文档同时说明这不是穷举列表，任何字符串事件类型都可直接使用，新事件无需升级库。

解析并消费通知

get_message(timeout)返回一个已解析的KeyNotification，超时未收到则返回None。listen()是一个阻塞生成器，适合"一直监听"的循环。KeyNotification的字段（定义）：

• key：受影响的键（keyevent 消息中为消息体携带的键名）
• event_type：事件类型字符串，如"set"、"del"
• database：事件发生的数据库编号
• is_keyspace：True 表示 keyspace 通知，False 表示 keyevent
• subkeys：subkey 级通知中受影响的字段列表，普通通知为空列表

一次性取消息：

r.set("user:1", "alice") notification = ksn.get_message(timeout=2.0) if notification: print(notification.key, notification.event_type)

持续监听：

for notification in ksn.listen(): print(f"Key: {notification.key}, Event: {notification.event_type}")

另一种消费方式是订阅时注册 handler，由 Pub/Sub 直接回调，此时get_message对该消息返回None；所有订阅都带 handler 时还可以用ksn.run_in_thread(poll_timeout=0.1, daemon=True)启动后台轮询线程（默认poll_timeout=0.0会形成 CPU 空转，文档建议传正值），结束时对返回的线程调用stop()（实现）。

验证：写入后断言收到事件

redis-py 的集成测试 tests/test_keyspace_notifications.py 给出了可直接参照的验证模式：开启KEA后订阅，写入数据，再用带超时的get_message取回并断言字段。以 subkey 级通知为例（测试文件 L1576-L1595）：

notifications = r.keyspace_notifications() notifications.subscribe_subkeyspace("test:hash1") r.hset("test:hash1", "field1", "value1") msg = notifications.get_message(timeout=2.0) assert msg is not None assert msg.key == "test:hash1" assert msg.event_type == "hset" assert "field1" in msg.subkeys notifications.close()

判断标准就是这三条断言：msg非空、key等于被写入的键、event_type等于触发的命令名（subkey 级通知还会校验subkeys包含被修改的字段）。get_message返回None表示超时未收到——先回到"前提条件"确认notify-keyspace-events确实已设置。

可选分支一：subkey 级通知（Redis >= 8.7.2）

如果只关心 hash 内部哪些字段变了（缓存细粒度失效等场景），Redis 提供了四种 subkey 通道，redis-py 对应四个通道类和订阅方法（SPEC 与 通道实现）：

前提是服务端版本支持（Redis >= 8.7.2）并用notify-keyspace-events加上 S/T/I/V 标志（conftest 使用KEASTIV）。注意SubkeyspaceitemChannel在服务端仅在键名不含\n时才发出该族通知。

可选分支二：Redis Cluster

集群模式下有一个关键差异（SPEC 说明）：keyspace/keyevent 消息不会像普通 Pub/Sub 那样在集群内横向传播——每个节点只对本地拥有的键发出事件，只连一个节点会漏掉其他节点上的变更。ClusterKeyspaceNotifications自动解决这个问题：对每个 primary 节点建立独立 Pub/Sub 订阅，并在拓扑变化（节点增删、故障转移）时通过refresh_subscriptions()重新订阅；连接报错时get_message内部也会自动触发刷新（实现）。

from redis.cluster import RedisCluster from redis.keyspace_notifications import KeyspaceChannel rc = RedisCluster(host="localhost", port=7000) ksn = rc.keyspace_notifications() ksn.subscribe(KeyspaceChannel("user:*")) for notification in ksn.listen(): print(notification.key, notification.event_type)

注意集群节点上同样要开启notify-keyspace-events——测试夹具对集群是逐 primary 节点config_set的（conftest），且 Redis OSS 集群只有逻辑数据库 0，db参数保持默认值即可。

收尾与清理

停止监听后调用ksn.close()关闭底层 Pub/Sub 连接（上下文管理器with块退出时自动调用）；只想退订部分通道用ksn.unsubscribe(...)。最后别忘记把服务端的notify-keyspace-events恢复为原值，避免通知流量影响同实例上的其他应用。

更多细节可查 完整模块文档与示例、功能规格 和 配套测试。

【免费下载链接】redis-pyRedis Python client项目地址: https://gitcode.com/GitHub_Trending/re/redis-py