如何用 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的结果。订阅通道按键或按事件通道类会自动生成带__keyspacedb__:前缀的完整通道名并自动识别通配符含*、?、[...]的走psubscribe精确键名走subscribe自动检测逻辑见 subscribe 实现。# 方式一监听某个键或一组键的变更keyspace消息体是事件类型 ksn.subscribe(KeyspaceChannel(user:123)) # 精确键 → __keyspace0__:user:123 ksn.subscribe(KeyspaceChannel(user:*)) # 通配 → psubscribe __keyspace0__:user:* # 方式二监听某类事件在所有键上发生keyevent消息体是键名 ksn.subscribe_keyevent(EventType.SET) # → __keyevent0__:set # 方式三用便捷方法按键订阅 ksn.subscribe_keyspace(session:*)常用事件类型常量集中在EventType类定义位置如SET、DEL、EXPIRED、HSET等文档同时说明这不是穷举列表任何字符串事件类型都可直接使用新事件无需升级库。解析并消费通知get_message(timeout)返回一个已解析的KeyNotification超时未收到则返回None。listen()是一个阻塞生成器适合一直监听的循环。KeyNotification的字段定义key受影响的键keyevent 消息中为消息体携带的键名event_type事件类型字符串如set、deldatabase事件发生的数据库编号is_keyspaceTrue 表示 keyspace 通知False 表示 keyeventsubkeyssubkey 级通知中受影响的字段列表普通通知为空列表一次性取消息r.set(user:1, alice) notification ksn.get_message(timeout2.0) if notification: print(notification.key, notification.event_type)持续监听for notification in ksn.listen(): print(fKey: {notification.key}, Event: {notification.event_type})另一种消费方式是订阅时注册 handler由 Pub/Sub 直接回调此时get_message对该消息返回None所有订阅都带 handler 时还可以用ksn.run_in_thread(poll_timeout0.1, daemonTrue)启动后台轮询线程默认poll_timeout0.0会形成 CPU 空转文档建议传正值结束时对返回的线程调用stop()实现。验证写入后断言收到事件redis-py 的集成测试 tests/test_keyspace_notifications.py 给出了可直接参照的验证模式开启KEA后订阅写入数据再用带超时的get_message取回并断言字段。以 subkey 级通知为例测试文件 L1576-L1595notifications r.keyspace_notifications() notifications.subscribe_subkeyspace(test:hash1) r.hset(test:hash1, field1, value1) msg notifications.get_message(timeout2.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 与 通道实现方法通道消息内容subscribe_subkeyspace(key)__subkeyspacedb__:key事件 受影响字段列表subscribe_subkeyevent(event)__subkeyeventdb__:event受影响的键 字段列表subscribe_subkeyspaceitem(key, subkey)__subkeyspaceitemdb__:key\nsubkey事件类型subscribe_subkeyspaceevent(event, key)__subkeyspaceeventdb__:event\|key受影响字段列表前提是服务端版本支持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(hostlocalhost, port7000) 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 集群只有逻辑数据库 0db参数保持默认值即可。收尾与清理停止监听后调用ksn.close()关闭底层 Pub/Sub 连接上下文管理器with块退出时自动调用只想退订部分通道用ksn.unsubscribe(...)。最后别忘记把服务端的notify-keyspace-events恢复为原值避免通知流量影响同实例上的其他应用。更多细节可查 完整模块文档与示例、功能规格 和 配套测试。【免费下载链接】redis-pyRedis Python client项目地址: https://gitcode.com/GitHub_Trending/re/redis-py创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 如何用 redis-py 订阅并解析 keyspace notifications 监听键变更事件
如何用 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的结果。订阅通道按键或按事件通道类会自动生成带__keyspacedb__:前缀的完整通道名并自动识别通配符含*、?、[...]的走psubscribe精确键名走subscribe自动检测逻辑见 subscribe 实现。# 方式一监听某个键或一组键的变更keyspace消息体是事件类型 ksn.subscribe(KeyspaceChannel(user:123)) # 精确键 → __keyspace0__:user:123 ksn.subscribe(KeyspaceChannel(user:*)) # 通配 → psubscribe __keyspace0__:user:* # 方式二监听某类事件在所有键上发生keyevent消息体是键名 ksn.subscribe_keyevent(EventType.SET) # → __keyevent0__:set # 方式三用便捷方法按键订阅 ksn.subscribe_keyspace(session:*)常用事件类型常量集中在EventType类定义位置如SET、DEL、EXPIRED、HSET等文档同时说明这不是穷举列表任何字符串事件类型都可直接使用新事件无需升级库。解析并消费通知get_message(timeout)返回一个已解析的KeyNotification超时未收到则返回None。listen()是一个阻塞生成器适合一直监听的循环。KeyNotification的字段定义key受影响的键keyevent 消息中为消息体携带的键名event_type事件类型字符串如set、deldatabase事件发生的数据库编号is_keyspaceTrue 表示 keyspace 通知False 表示 keyeventsubkeyssubkey 级通知中受影响的字段列表普通通知为空列表一次性取消息r.set(user:1, alice) notification ksn.get_message(timeout2.0) if notification: print(notification.key, notification.event_type)持续监听for notification in ksn.listen(): print(fKey: {notification.key}, Event: {notification.event_type})另一种消费方式是订阅时注册 handler由 Pub/Sub 直接回调此时get_message对该消息返回None所有订阅都带 handler 时还可以用ksn.run_in_thread(poll_timeout0.1, daemonTrue)启动后台轮询线程默认poll_timeout0.0会形成 CPU 空转文档建议传正值结束时对返回的线程调用stop()实现。验证写入后断言收到事件redis-py 的集成测试 tests/test_keyspace_notifications.py 给出了可直接参照的验证模式开启KEA后订阅写入数据再用带超时的get_message取回并断言字段。以 subkey 级通知为例测试文件 L1576-L1595notifications r.keyspace_notifications() notifications.subscribe_subkeyspace(test:hash1) r.hset(test:hash1, field1, value1) msg notifications.get_message(timeout2.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 与 通道实现方法通道消息内容subscribe_subkeyspace(key)__subkeyspacedb__:key事件 受影响字段列表subscribe_subkeyevent(event)__subkeyeventdb__:event受影响的键 字段列表subscribe_subkeyspaceitem(key, subkey)__subkeyspaceitemdb__:key\nsubkey事件类型subscribe_subkeyspaceevent(event, key)__subkeyspaceeventdb__:event\|key受影响字段列表前提是服务端版本支持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(hostlocalhost, port7000) 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 集群只有逻辑数据库 0db参数保持默认值即可。收尾与清理停止监听后调用ksn.close()关闭底层 Pub/Sub 连接上下文管理器with块退出时自动调用只想退订部分通道用ksn.unsubscribe(...)。最后别忘记把服务端的notify-keyspace-events恢复为原值避免通知流量影响同实例上的其他应用。更多细节可查 完整模块文档与示例、功能规格 和 配套测试。【免费下载链接】redis-pyRedis Python client项目地址: https://gitcode.com/GitHub_Trending/re/redis-py创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考