redis-py 集群 Keyspace Notifications 消费方案解析从设计规格到源码实现【免费下载链接】redis-pyRedis Python client项目地址: https://gitcode.com/GitHub_Trending/re/redis-py在 Redis OSS 集群模式下键空间通知Keyspace Notifications不会像普通 Pub/Sub 消息那样在节点间水平传播导致单节点订阅难以覆盖全集群事件。本文基于仓库内 specs/keyspace-notifications/SPEC.md 设计规格并结合 redis-py 的同步与异步源码实现完整讲解 keyspace / keyevent / subkeyspace 等六类通知频道的前缀约定、负载线格式、集群拓扑应对策略以及如何用现成的 API 实现跨节点、跨拓扑变化的通知消费。为什么集群模式下不能直接消费 Keyspace Notifications对于单机standaloneRedis通过常规 Pub/Sub 频道消费 keyspace notifications 非常直接客户端只需subscribe/psubscribe到对应频道即可。但在 OSS 集群模式下事情并不简单这正是本规格文档诞生的初衷Whereby it is easy to consume keyspace notifications via normal PubSub channels with a standalone Redis client, it is not straightforward to consume them in OSS cluster mode.根本原因在于普通 Pub/Sub 消息与 keyspace notifications 的传播机制不同见规格的 Cluster specifics 一节普通 Pub/Sub 消息会在节点间水平传播客户端 A 连接节点 X 并执行PSUBSCRIBE foo*客户端 B 连接集群中的另一节点 Y 并执行PUBLISH foo/bar some_payloadA 会经由 BYXA 的链路间接收到消息。Keyspace / Keyevent 消息不会经由集群总线cluster bus传播服务器只对在该节点上被操作的键发布事件只有直接订阅到该节点的客户端才能收到通知。因此规格明确指出要使用任何多键订阅基于模式匹配的__keyspace或任意__keyevent客户端必须连接到该层级的全部相关节点——通常意味着集群中所有主节点primary nodes——并且这一状态必须能经受住拓扑变化节点增删、故障转移等而持续有效。这正是 redis-py 中ClusterKeyspaceNotifications与AsyncClusterKeyspaceNotifications要解决的抽象问题。六类通知频道前缀约定与负载格式规格将通知频道划分为两大类共六种。下面逐一梳理其频道前缀与负载payload语义这些约定在 redis/keyspace_notifications.py 的解析逻辑中均有对应实现。基础两类Keyspace 与 KeyeventKeyspace前缀__keyspacedb__:监听对特定键执行的操作例如__keyspace0__:mykey通知负载返回的是对该键执行的操作事件类型。Keyevent前缀__keyeventdb__:监听特定操作事件例如__keyevent0__:del通知负载返回的是受影响的键名。即keyspace 频道按键订阅、负载是事件keyevent 频道按事件订阅、负载是键名两者互为倒置。子键级四类Subkeyspace 系列很多应用需要更精细的通知尤其是针对哈希结构。典型场景包括确定哈希中哪个字段被修改、确定哪些字段过期、进行细粒度缓存失效、在不重处理整个键的情况下响应局部对象变化。规格提出的 Redis 核心新特性在保留现有 Pub/Sub 消息信封的前提下扩展出了四种子键感知subkey-aware通知通知类型频道前缀频道含义负载格式Subkeyspace__subkeyspacedb__:监听特定键如__subkeyspace0__:myhash受影响子键集合event\|subkeysSubkeyevent__subkeyeventdb__:监听特定事件如__subkeyevent0__:del受影响子键集合 受影响键key\|subkeysSubkeyspaceitem__subkeyspaceitemdb__:监听特定键的特定子键如__subkeyspaceitem0__:myhash\nmyfield子键上执行的事件Subkeyspaceevent__subkeyspaceeventdb__:监听特定键上的特定事件如__subkeyspaceevent0__:event\|key受影响的子键集合v1 契约中的线格式Wire Format细则规格在需求清单中对各类型的解析给出了严格约束redis-py 在 KeyNotification._parse 中一一落实Subkeyspace的规范线格式为多子键长度前缀形式event|subkey_len:subkey[,subkey_len:subkey...]紧凑的单子键形式不属于受支持的 v1 契约。解析时以|切出事件类型再用_parse_length_prefixed_subkeys按长度:值序列解析子键列表。Subkeyevent必须将负载按key_len:key|subkey_len:subkey[,subkey_len:subkey...]解析按长度前缀读取键名而非按分隔符切分——这是为了避免键名中可能出现的|造成歧义。Subkeyspaceitem的频道为key\nsubkey键与子键以换行符分隔负载为event服务器仅在键不含\n时才发出该家族事件。Subkeyspaceevent的频道为event|key负载为长度前缀的子键列表。_parse_length_prefixed_subkeys的实现redis/keyspace_notifications.py从当前位置查找:读取长度随后按长度截取子键若后续为,则跳过继续从而支持含任意字符包括逗号、冒号的子键名。集群消费的三大特殊性规格的 Cluster specifics 一节归纳了三个必须处理的事实必须订阅全部主节点由于事件不跨节点传播任何多键/模式订阅都必须连接到所有相关节点通常意味着所有主节点并且必须在拓扑变化后保持该订阅状态。单键订阅可以只连正确节点对__keyspace0__:mysinglekey这类单键 keyspace 通知集群中只有持有该键槽的节点及其副本会发布消息。为降低开销通常只订阅正确的服务器即可。逻辑数据库差异Redis OSS Cluster 不支持SELECT db_index这种多逻辑库语义但独立客户端仍然可以借助SELECT使用多库。因此新抽象需要同时支持 standalone 的多库与 cluster 的仅库 0。规格同时要求为保持一致性该抽象也应面向 standalone 客户端实现——这也解释了仓库中KeyspaceNotifications与ClusterKeyspaceNotifications成对出现的原因。需求清单新抽象必须满足的能力规格 Requirements 一节将目标能力明确为通过集群客户端应当能够——通过新 API 订阅keyspace与keyevent频道跨集群消费两类基础通知通过新 API 订阅subkeyspace、subkeyevent、subkeyspaceitem、subkeyspaceevent四类子键通知遵循上文 v1 线格式契约完成解析standalone 客户端支持多逻辑库SELECT db_index集群继续仅使用库 0考虑普通发布消息与 keyspace 通知的行为差异——keyspace 通知不经集群总线在节点间传播自动响应拓扑变化节点增删、槽迁移等重新订阅相关频道。redis-py 的实现落地模块全景规格文档是设计蓝图仓库中的 redis/keyspace_notifications.py同步约 2174 行与 redis/asyncio/keyspace_notifications.py异步约 1029 行则是完整实现。测试方面tests/test_keyspace_notifications.py 与 tests/test_asyncio/test_keyspace_notifications.py 提供了大量用例覆盖。事件类型常量 EventType模块定义了EventType类汇总了常用的事件常量redis/keyspace_notifications.py涵盖字符串命令set、incr、append…、通用命令del、unlink、rename…、过期/淘汰expire、expired、evicted…、列表、集合、有序集合、哈希hset、hdel、hincrby…、流xadd、xdel…等。文档字符串明确说明这些常量仅为便捷与 IDE 自动补全而设任何字符串都可作为事件类型Redis 新增事件无需升级库即可工作。频道类六种 Channel 对象模块提供了六个频道类均实现__str__直接返回频道字符串因此可无缝用于subscribe()/psubscribe()KeyspaceChannel(key_or_pattern, db0)→__keyspace0__:user:*KeyeventChannel(event, db0)→__keyevent0__:set并提供KeyeventChannel.all_events()便捷方法生成__keyevent0__:*SubkeyspaceChannel(key_or_pattern, db0)、SubkeyeventChannel(event, db0)含all_events()SubkeyspaceitemChannel(key_or_pattern, subkey_or_pattern, db0)→__subkeyspaceitem0__:myhash\nmyfieldSubkeyspaceeventChannel(event, key_or_pattern, db0)→__subkeyspaceevent0__:hset|myhash。每个类都提供is_pattern属性用于判断是否包含通配符以决定走psubscribe还是subscribe。通配检测由_is_pattern完成redis/keyspace_notifications.py实现细节包括*与?始终是模式字符[只有在其后存在未转义的配对]构成[abc]或[a-z]括号表达式时才是模式字符孤立的[如键名my[key按字面量处理反斜杠转义字符会被跳过。消息解析KeyNotificationKeyNotification数据类redis/keyspace_notifications.py是通知的统一载体字段包括key受影响的键keyspace 通知取自频道、keyevent 通知取自消息数据event_type发生的操作类型纯字符串新事件自动兼容database事件发生的库号*通配时为 -1channel原始频道名is_keyspaceTrue 表示 keyspace 类False 表示 keyevent 类data原始负载subkeys子键字段列表普通通知为空列表。解析入口KeyNotification.from_message(message, key_prefixNone)接受 Pub/Sub 消息字典含type/channel/data键仅处理message与pmessage两种类型底层_parse用六条正则__keyspace(\d|\*)__:等逐一匹配六类频道前缀并实现了规格要求的全部线格式解析。key_prefix参数支持按前缀过滤并从 key 中剥离该前缀便于业务侧统一去前缀处理。独立版KeyspaceNotificationsKeyspaceNotifications(redis_client, key_prefixNone, ignore_subscribe_messagesTrue)redis/keyspace_notifications.py包装单个 Pub/Sub 连接subscribe(*channels, handlerNone)自动区分模式与精确频道分别调用psubscribe/subscribe传入handler时原始消息会被包装成KeyNotification后回调未传 handler 的通知由get_message()/listen()返回get_message(ignore_subscribe_messagesNone, timeout0.0)返回解析后的KeyNotificationlisten()为阻塞生成器持续产出通知run_in_thread(poll_timeout0.0, daemonFalse, exception_handlerNone)启动后台线程轮询并自动触发 handler注意源码警告poll_timeout默认 0.0 会造成 CPU 空转建议传正值如 0.1 或 1.0subscribed属性、close()与上下文管理器with齐备_validate_all_handlers会在无 handler 订阅存在时抛出RedisError。集群版ClusterKeyspaceNotificationsClusterKeyspaceNotifications(redis_cluster, ...)redis/keyspace_notifications.py是规格中跨集群 拓扑自愈需求的核心实现其关键设计每节点一个 PubSub_node_pubsubs字典按节点名维护 PubSub 实例通过cluster.get_primaries()获取全部主节点cluster.get_redis_connection(node)建立连接异步版使用_ClusterNodePoolAdapter适配。规范订阅注册表_subscribed_patterns与_subscribed_channels是唯一事实来源single source of truth。_execute_subscribe单轮遍历所有主节点完成订阅若某节点中途失败会被移出_node_pubsubs待下一轮refresh_subscriptions全量补订从而避免部分追平的不一致状态。新节点补齐历史订阅若发现全新节点不在_node_pubsubs中会先用先前已跟踪的全部 pattern/channel 订阅它再叠加本次新增订阅确保新节点不遗漏旧订阅对应的通知。拓扑刷新refresh_subscriptions在_refresh_lock保护下① 移除已不再是主节点的 PubSub② 检测并清理连接已断开的 PubSub③ 对新节点含重连节点按注册表全量重订。若仍有节点订阅失败则抛出ConnectionError。轮询消费get_message对全部节点 PubSub 做轮询round-robintimeout0时按节点均摊超时每节点至多 0.1 秒循环直至总超时遇到ConnectionError/TimeoutError/RedisError自动触发_refresh_subscriptions_on_error后继续timeout0则做一轮非阻塞全节点扫描。尽力而为的退订_execute_unsubscribe逐个节点退订单个坏连接不阻断其余节点退订后注册表同步移除确保刷新时不会重新订阅用户明确退订的频道。异步版本redis/asyncio/keyspace_notifications.py 提供了对等的异步实现AsyncKeyspaceNotificationsstandalone与AsyncClusterKeyspaceNotifications集群见 异步集群类接口方法均为async defhandler 支持同步或异步回调AsyncHandlerT Callable[[KeyNotification], None | Awaitable[None]]并提供async for notification in ksn.listen()的异步迭代消费方式模块 docstring 中给出了完整的async with用法示例。解析类KeyNotification、频道类、ChannelT直接复用同步模块保证两种模式语义一致。使用示例同步 standalonefrom redis import Redis from redis.keyspace_notifications import ( KeyspaceNotifications, KeyspaceChannel, KeyeventChannel, EventType, ) r Redis() # 注意服务器必须配置 notify-keyspace-events例如 KEA # 这是服务端配置应由基础设施/运维团队完成 ksn KeyspaceNotifications(r) # 模式订阅自动识别通配符并走 psubscribe channel KeyspaceChannel(user:*) ksn.subscribe(channel) # 便捷方法订阅特定事件类型 ksn.subscribe_keyevent(EventType.SET) for notification in ksn.listen(): print(fKey: {notification.key}, Event: {notification.event_type})同步集群from redis.cluster import RedisCluster from redis.keyspace_notifications import ( ClusterKeyspaceNotifications, KeyspaceChannel, EventType, ) rc RedisCluster(hostlocalhost, port7000) ksn ClusterKeyspaceNotifications(rc) ksn.subscribe(KeyspaceChannel(user:*)) ksn.subscribe_keyevent(EventType.SET) for notification in ksn.listen(): print(fKey: {notification.key}, Event: {notification.event_type})集群版会自动订阅全部主节点并在拓扑变化节点增删、槽迁移、故障转移或连接错误时通过refresh_subscriptions自动重订业务代码无需感知节点细节。异步集群from redis.asyncio.cluster import RedisCluster from redis.asyncio.keyspace_notifications import AsyncClusterKeyspaceNotifications from redis.keyspace_notifications import KeyspaceChannel, EventType async def main(): async with RedisCluster(hostlocalhost, port7000) as rc: async with AsyncClusterKeyspaceNotifications(rc) as ksn: await ksn.subscribe(KeyspaceChannel(user:*)) async for notification in ksn.listen(): print(fKey: {notification.key}, Event: {notification.event_type})子键通知哈希字段级from redis.keyspace_notifications import ( KeyspaceNotifications, SubkeyspaceChannel, SubkeyeventChannel, SubkeyspaceitemChannel, SubkeyspaceeventChannel, EventType, ) ksn KeyspaceNotifications(r) # 监听 myhash 上所有字段变化负载为 event|subkey_len:subkey... ksn.subscribe(SubkeyspaceChannel(myhash, db0)) # 监听所有 hdel 事件负载为 key_len:key|subkey_len:subkey... ksn.subscribe(SubkeyeventChannel(hdel, db0)) # 监听 myhash 上 myfield 字段的变化频道为 __subkeyspaceitem0__:myhash\nmyfield ksn.subscribe(SubkeyspaceitemChannel(myhash, myfield, db0)) # 监听 myhash 上的 hset 事件频道为 __subkeyspaceevent0__:hset|myhash ksn.subscribe(SubkeyspaceeventChannel(hset, myhash, db0)) for notification in ksn.listen(): print(notification.key, notification.event_type, notification.subkeys)测试验证规格用例在仓库中的落地规格 Test cases 一节列出了 11 条跨节点验收场景仓库测试对其中大部分做了镜像覆盖tests/test_keyspace_notifications.py 共 154 处def test_组织为TestEventType、TestPatternDetection、TestKeyspaceChannelClass、TestKeyNotification、TestClusterKeyspaceNotificationsMocked、TestClusterKeyspaceNotifications、TestStandaloneClientKeyspaceNotificationsMocked、TestParseLengthPrefixedSubkeys、四个子键频道类测试、TestSubkeyChannelDetection、TestSubkeyNotificationParsing、TestSubkeyNotifications等测试类集群拓扑相关用例包括test_refresh_subscriptions_recovers_broken_connections刷新自愈断连、test_node_failure_during_pattern_subscribe_does_not_lose_patterns订阅中途节点故障不丢模式、test_receives_notification_from_any_node任意节点事件均可达、test_is_pubsub_connected_returns_false_for_broken_connection等子键通知用例与规格逐条对应test_create_hash_field_subkeyspace_notification创建哈希字段收到 Subkeyspace 且含该字段、test_update_hash_field_subkeyspaceitem_notification更新字段同时收到 Subkeyspace 与 Subkeyspaceitem、test_delete_hash_field_subkeyevent_notification、test_delete_hash_field_subkeyspaceevent_notification删除字段在 Subkeyevent 与 Subkeyspaceevent 频道收到、test_multiple_hash_fields_subkeyspace_notification多字段批量通知、test_subkeyspace_pattern_subscription/test_subkeyevent_pattern_subscription模式订阅、test_subkeyspaceitem_does_not_receive_other_fields不收到其他字段事件、test_combined_keyspace_and_subkeyspace基础与子键通知共存异步侧 tests/test_asyncio/test_keyspace_notifications.py 提供对等的 async 用例覆盖精确/模式订阅、回调 handler、上下文管理器、key_prefix过滤、全节点订阅、断连恢复等场景。规格中关于槽迁移的验收点Move a bunch of slots from node 1 to node 2 ... ensure that keyspace notifications and key event notifications are received correctly before and after the migration可由refresh_subscriptions的新主节点补订历史订阅 移除失效节点 PubSub机制从原理上支撑槽迁移后目标节点成为相关键的主节点refresh_subscriptions会确保其订阅状态与注册表一致。小结围绕 specs/keyspace-notifications/SPEC.mdredis-py 给出了从设计到实现的完整闭环六类通知频道与 v1 线格式在KeyNotification中精确落地standalone 与 cluster 双形态同步 异步共四个管理器类满足了抽象一致性的规格要求集群版以每节点 PubSub 规范注册表 拓扑刷新三件套解决了事件不跨节点传播与拓扑自愈两个核心难题。对于需要在集群中做细粒度缓存失效、哈希字段级变更追踪或事件驱动集成的开发者这套 API 是可直接落地的现成方案在接入前请务必确认所有相关节点的notify-keyspace-events服务端配置已开启例如KEA这是整个机制生效的前提。【免费下载链接】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 OSS 集群模式下键空间通知Keyspace Notifications不会像普通 Pub/Sub 消息那样在节点间水平传播导致单节点订阅难以覆盖全集群事件。本文基于仓库内 specs/keyspace-notifications/SPEC.md 设计规格并结合 redis-py 的同步与异步源码实现完整讲解 keyspace / keyevent / subkeyspace 等六类通知频道的前缀约定、负载线格式、集群拓扑应对策略以及如何用现成的 API 实现跨节点、跨拓扑变化的通知消费。为什么集群模式下不能直接消费 Keyspace Notifications对于单机standaloneRedis通过常规 Pub/Sub 频道消费 keyspace notifications 非常直接客户端只需subscribe/psubscribe到对应频道即可。但在 OSS 集群模式下事情并不简单这正是本规格文档诞生的初衷Whereby it is easy to consume keyspace notifications via normal PubSub channels with a standalone Redis client, it is not straightforward to consume them in OSS cluster mode.根本原因在于普通 Pub/Sub 消息与 keyspace notifications 的传播机制不同见规格的 Cluster specifics 一节普通 Pub/Sub 消息会在节点间水平传播客户端 A 连接节点 X 并执行PSUBSCRIBE foo*客户端 B 连接集群中的另一节点 Y 并执行PUBLISH foo/bar some_payloadA 会经由 BYXA 的链路间接收到消息。Keyspace / Keyevent 消息不会经由集群总线cluster bus传播服务器只对在该节点上被操作的键发布事件只有直接订阅到该节点的客户端才能收到通知。因此规格明确指出要使用任何多键订阅基于模式匹配的__keyspace或任意__keyevent客户端必须连接到该层级的全部相关节点——通常意味着集群中所有主节点primary nodes——并且这一状态必须能经受住拓扑变化节点增删、故障转移等而持续有效。这正是 redis-py 中ClusterKeyspaceNotifications与AsyncClusterKeyspaceNotifications要解决的抽象问题。六类通知频道前缀约定与负载格式规格将通知频道划分为两大类共六种。下面逐一梳理其频道前缀与负载payload语义这些约定在 redis/keyspace_notifications.py 的解析逻辑中均有对应实现。基础两类Keyspace 与 KeyeventKeyspace前缀__keyspacedb__:监听对特定键执行的操作例如__keyspace0__:mykey通知负载返回的是对该键执行的操作事件类型。Keyevent前缀__keyeventdb__:监听特定操作事件例如__keyevent0__:del通知负载返回的是受影响的键名。即keyspace 频道按键订阅、负载是事件keyevent 频道按事件订阅、负载是键名两者互为倒置。子键级四类Subkeyspace 系列很多应用需要更精细的通知尤其是针对哈希结构。典型场景包括确定哈希中哪个字段被修改、确定哪些字段过期、进行细粒度缓存失效、在不重处理整个键的情况下响应局部对象变化。规格提出的 Redis 核心新特性在保留现有 Pub/Sub 消息信封的前提下扩展出了四种子键感知subkey-aware通知通知类型频道前缀频道含义负载格式Subkeyspace__subkeyspacedb__:监听特定键如__subkeyspace0__:myhash受影响子键集合event\|subkeysSubkeyevent__subkeyeventdb__:监听特定事件如__subkeyevent0__:del受影响子键集合 受影响键key\|subkeysSubkeyspaceitem__subkeyspaceitemdb__:监听特定键的特定子键如__subkeyspaceitem0__:myhash\nmyfield子键上执行的事件Subkeyspaceevent__subkeyspaceeventdb__:监听特定键上的特定事件如__subkeyspaceevent0__:event\|key受影响的子键集合v1 契约中的线格式Wire Format细则规格在需求清单中对各类型的解析给出了严格约束redis-py 在 KeyNotification._parse 中一一落实Subkeyspace的规范线格式为多子键长度前缀形式event|subkey_len:subkey[,subkey_len:subkey...]紧凑的单子键形式不属于受支持的 v1 契约。解析时以|切出事件类型再用_parse_length_prefixed_subkeys按长度:值序列解析子键列表。Subkeyevent必须将负载按key_len:key|subkey_len:subkey[,subkey_len:subkey...]解析按长度前缀读取键名而非按分隔符切分——这是为了避免键名中可能出现的|造成歧义。Subkeyspaceitem的频道为key\nsubkey键与子键以换行符分隔负载为event服务器仅在键不含\n时才发出该家族事件。Subkeyspaceevent的频道为event|key负载为长度前缀的子键列表。_parse_length_prefixed_subkeys的实现redis/keyspace_notifications.py从当前位置查找:读取长度随后按长度截取子键若后续为,则跳过继续从而支持含任意字符包括逗号、冒号的子键名。集群消费的三大特殊性规格的 Cluster specifics 一节归纳了三个必须处理的事实必须订阅全部主节点由于事件不跨节点传播任何多键/模式订阅都必须连接到所有相关节点通常意味着所有主节点并且必须在拓扑变化后保持该订阅状态。单键订阅可以只连正确节点对__keyspace0__:mysinglekey这类单键 keyspace 通知集群中只有持有该键槽的节点及其副本会发布消息。为降低开销通常只订阅正确的服务器即可。逻辑数据库差异Redis OSS Cluster 不支持SELECT db_index这种多逻辑库语义但独立客户端仍然可以借助SELECT使用多库。因此新抽象需要同时支持 standalone 的多库与 cluster 的仅库 0。规格同时要求为保持一致性该抽象也应面向 standalone 客户端实现——这也解释了仓库中KeyspaceNotifications与ClusterKeyspaceNotifications成对出现的原因。需求清单新抽象必须满足的能力规格 Requirements 一节将目标能力明确为通过集群客户端应当能够——通过新 API 订阅keyspace与keyevent频道跨集群消费两类基础通知通过新 API 订阅subkeyspace、subkeyevent、subkeyspaceitem、subkeyspaceevent四类子键通知遵循上文 v1 线格式契约完成解析standalone 客户端支持多逻辑库SELECT db_index集群继续仅使用库 0考虑普通发布消息与 keyspace 通知的行为差异——keyspace 通知不经集群总线在节点间传播自动响应拓扑变化节点增删、槽迁移等重新订阅相关频道。redis-py 的实现落地模块全景规格文档是设计蓝图仓库中的 redis/keyspace_notifications.py同步约 2174 行与 redis/asyncio/keyspace_notifications.py异步约 1029 行则是完整实现。测试方面tests/test_keyspace_notifications.py 与 tests/test_asyncio/test_keyspace_notifications.py 提供了大量用例覆盖。事件类型常量 EventType模块定义了EventType类汇总了常用的事件常量redis/keyspace_notifications.py涵盖字符串命令set、incr、append…、通用命令del、unlink、rename…、过期/淘汰expire、expired、evicted…、列表、集合、有序集合、哈希hset、hdel、hincrby…、流xadd、xdel…等。文档字符串明确说明这些常量仅为便捷与 IDE 自动补全而设任何字符串都可作为事件类型Redis 新增事件无需升级库即可工作。频道类六种 Channel 对象模块提供了六个频道类均实现__str__直接返回频道字符串因此可无缝用于subscribe()/psubscribe()KeyspaceChannel(key_or_pattern, db0)→__keyspace0__:user:*KeyeventChannel(event, db0)→__keyevent0__:set并提供KeyeventChannel.all_events()便捷方法生成__keyevent0__:*SubkeyspaceChannel(key_or_pattern, db0)、SubkeyeventChannel(event, db0)含all_events()SubkeyspaceitemChannel(key_or_pattern, subkey_or_pattern, db0)→__subkeyspaceitem0__:myhash\nmyfieldSubkeyspaceeventChannel(event, key_or_pattern, db0)→__subkeyspaceevent0__:hset|myhash。每个类都提供is_pattern属性用于判断是否包含通配符以决定走psubscribe还是subscribe。通配检测由_is_pattern完成redis/keyspace_notifications.py实现细节包括*与?始终是模式字符[只有在其后存在未转义的配对]构成[abc]或[a-z]括号表达式时才是模式字符孤立的[如键名my[key按字面量处理反斜杠转义字符会被跳过。消息解析KeyNotificationKeyNotification数据类redis/keyspace_notifications.py是通知的统一载体字段包括key受影响的键keyspace 通知取自频道、keyevent 通知取自消息数据event_type发生的操作类型纯字符串新事件自动兼容database事件发生的库号*通配时为 -1channel原始频道名is_keyspaceTrue 表示 keyspace 类False 表示 keyevent 类data原始负载subkeys子键字段列表普通通知为空列表。解析入口KeyNotification.from_message(message, key_prefixNone)接受 Pub/Sub 消息字典含type/channel/data键仅处理message与pmessage两种类型底层_parse用六条正则__keyspace(\d|\*)__:等逐一匹配六类频道前缀并实现了规格要求的全部线格式解析。key_prefix参数支持按前缀过滤并从 key 中剥离该前缀便于业务侧统一去前缀处理。独立版KeyspaceNotificationsKeyspaceNotifications(redis_client, key_prefixNone, ignore_subscribe_messagesTrue)redis/keyspace_notifications.py包装单个 Pub/Sub 连接subscribe(*channels, handlerNone)自动区分模式与精确频道分别调用psubscribe/subscribe传入handler时原始消息会被包装成KeyNotification后回调未传 handler 的通知由get_message()/listen()返回get_message(ignore_subscribe_messagesNone, timeout0.0)返回解析后的KeyNotificationlisten()为阻塞生成器持续产出通知run_in_thread(poll_timeout0.0, daemonFalse, exception_handlerNone)启动后台线程轮询并自动触发 handler注意源码警告poll_timeout默认 0.0 会造成 CPU 空转建议传正值如 0.1 或 1.0subscribed属性、close()与上下文管理器with齐备_validate_all_handlers会在无 handler 订阅存在时抛出RedisError。集群版ClusterKeyspaceNotificationsClusterKeyspaceNotifications(redis_cluster, ...)redis/keyspace_notifications.py是规格中跨集群 拓扑自愈需求的核心实现其关键设计每节点一个 PubSub_node_pubsubs字典按节点名维护 PubSub 实例通过cluster.get_primaries()获取全部主节点cluster.get_redis_connection(node)建立连接异步版使用_ClusterNodePoolAdapter适配。规范订阅注册表_subscribed_patterns与_subscribed_channels是唯一事实来源single source of truth。_execute_subscribe单轮遍历所有主节点完成订阅若某节点中途失败会被移出_node_pubsubs待下一轮refresh_subscriptions全量补订从而避免部分追平的不一致状态。新节点补齐历史订阅若发现全新节点不在_node_pubsubs中会先用先前已跟踪的全部 pattern/channel 订阅它再叠加本次新增订阅确保新节点不遗漏旧订阅对应的通知。拓扑刷新refresh_subscriptions在_refresh_lock保护下① 移除已不再是主节点的 PubSub② 检测并清理连接已断开的 PubSub③ 对新节点含重连节点按注册表全量重订。若仍有节点订阅失败则抛出ConnectionError。轮询消费get_message对全部节点 PubSub 做轮询round-robintimeout0时按节点均摊超时每节点至多 0.1 秒循环直至总超时遇到ConnectionError/TimeoutError/RedisError自动触发_refresh_subscriptions_on_error后继续timeout0则做一轮非阻塞全节点扫描。尽力而为的退订_execute_unsubscribe逐个节点退订单个坏连接不阻断其余节点退订后注册表同步移除确保刷新时不会重新订阅用户明确退订的频道。异步版本redis/asyncio/keyspace_notifications.py 提供了对等的异步实现AsyncKeyspaceNotificationsstandalone与AsyncClusterKeyspaceNotifications集群见 异步集群类接口方法均为async defhandler 支持同步或异步回调AsyncHandlerT Callable[[KeyNotification], None | Awaitable[None]]并提供async for notification in ksn.listen()的异步迭代消费方式模块 docstring 中给出了完整的async with用法示例。解析类KeyNotification、频道类、ChannelT直接复用同步模块保证两种模式语义一致。使用示例同步 standalonefrom redis import Redis from redis.keyspace_notifications import ( KeyspaceNotifications, KeyspaceChannel, KeyeventChannel, EventType, ) r Redis() # 注意服务器必须配置 notify-keyspace-events例如 KEA # 这是服务端配置应由基础设施/运维团队完成 ksn KeyspaceNotifications(r) # 模式订阅自动识别通配符并走 psubscribe channel KeyspaceChannel(user:*) ksn.subscribe(channel) # 便捷方法订阅特定事件类型 ksn.subscribe_keyevent(EventType.SET) for notification in ksn.listen(): print(fKey: {notification.key}, Event: {notification.event_type})同步集群from redis.cluster import RedisCluster from redis.keyspace_notifications import ( ClusterKeyspaceNotifications, KeyspaceChannel, EventType, ) rc RedisCluster(hostlocalhost, port7000) ksn ClusterKeyspaceNotifications(rc) ksn.subscribe(KeyspaceChannel(user:*)) ksn.subscribe_keyevent(EventType.SET) for notification in ksn.listen(): print(fKey: {notification.key}, Event: {notification.event_type})集群版会自动订阅全部主节点并在拓扑变化节点增删、槽迁移、故障转移或连接错误时通过refresh_subscriptions自动重订业务代码无需感知节点细节。异步集群from redis.asyncio.cluster import RedisCluster from redis.asyncio.keyspace_notifications import AsyncClusterKeyspaceNotifications from redis.keyspace_notifications import KeyspaceChannel, EventType async def main(): async with RedisCluster(hostlocalhost, port7000) as rc: async with AsyncClusterKeyspaceNotifications(rc) as ksn: await ksn.subscribe(KeyspaceChannel(user:*)) async for notification in ksn.listen(): print(fKey: {notification.key}, Event: {notification.event_type})子键通知哈希字段级from redis.keyspace_notifications import ( KeyspaceNotifications, SubkeyspaceChannel, SubkeyeventChannel, SubkeyspaceitemChannel, SubkeyspaceeventChannel, EventType, ) ksn KeyspaceNotifications(r) # 监听 myhash 上所有字段变化负载为 event|subkey_len:subkey... ksn.subscribe(SubkeyspaceChannel(myhash, db0)) # 监听所有 hdel 事件负载为 key_len:key|subkey_len:subkey... ksn.subscribe(SubkeyeventChannel(hdel, db0)) # 监听 myhash 上 myfield 字段的变化频道为 __subkeyspaceitem0__:myhash\nmyfield ksn.subscribe(SubkeyspaceitemChannel(myhash, myfield, db0)) # 监听 myhash 上的 hset 事件频道为 __subkeyspaceevent0__:hset|myhash ksn.subscribe(SubkeyspaceeventChannel(hset, myhash, db0)) for notification in ksn.listen(): print(notification.key, notification.event_type, notification.subkeys)测试验证规格用例在仓库中的落地规格 Test cases 一节列出了 11 条跨节点验收场景仓库测试对其中大部分做了镜像覆盖tests/test_keyspace_notifications.py 共 154 处def test_组织为TestEventType、TestPatternDetection、TestKeyspaceChannelClass、TestKeyNotification、TestClusterKeyspaceNotificationsMocked、TestClusterKeyspaceNotifications、TestStandaloneClientKeyspaceNotificationsMocked、TestParseLengthPrefixedSubkeys、四个子键频道类测试、TestSubkeyChannelDetection、TestSubkeyNotificationParsing、TestSubkeyNotifications等测试类集群拓扑相关用例包括test_refresh_subscriptions_recovers_broken_connections刷新自愈断连、test_node_failure_during_pattern_subscribe_does_not_lose_patterns订阅中途节点故障不丢模式、test_receives_notification_from_any_node任意节点事件均可达、test_is_pubsub_connected_returns_false_for_broken_connection等子键通知用例与规格逐条对应test_create_hash_field_subkeyspace_notification创建哈希字段收到 Subkeyspace 且含该字段、test_update_hash_field_subkeyspaceitem_notification更新字段同时收到 Subkeyspace 与 Subkeyspaceitem、test_delete_hash_field_subkeyevent_notification、test_delete_hash_field_subkeyspaceevent_notification删除字段在 Subkeyevent 与 Subkeyspaceevent 频道收到、test_multiple_hash_fields_subkeyspace_notification多字段批量通知、test_subkeyspace_pattern_subscription/test_subkeyevent_pattern_subscription模式订阅、test_subkeyspaceitem_does_not_receive_other_fields不收到其他字段事件、test_combined_keyspace_and_subkeyspace基础与子键通知共存异步侧 tests/test_asyncio/test_keyspace_notifications.py 提供对等的 async 用例覆盖精确/模式订阅、回调 handler、上下文管理器、key_prefix过滤、全节点订阅、断连恢复等场景。规格中关于槽迁移的验收点Move a bunch of slots from node 1 to node 2 ... ensure that keyspace notifications and key event notifications are received correctly before and after the migration可由refresh_subscriptions的新主节点补订历史订阅 移除失效节点 PubSub机制从原理上支撑槽迁移后目标节点成为相关键的主节点refresh_subscriptions会确保其订阅状态与注册表一致。小结围绕 specs/keyspace-notifications/SPEC.mdredis-py 给出了从设计到实现的完整闭环六类通知频道与 v1 线格式在KeyNotification中精确落地standalone 与 cluster 双形态同步 异步共四个管理器类满足了抽象一致性的规格要求集群版以每节点 PubSub 规范注册表 拓扑刷新三件套解决了事件不跨节点传播与拓扑自愈两个核心难题。对于需要在集群中做细粒度缓存失效、哈希字段级变更追踪或事件驱动集成的开发者这套 API 是可直接落地的现成方案在接入前请务必确认所有相关节点的notify-keyspace-events服务端配置已开启例如KEA这是整个机制生效的前提。【免费下载链接】redis-pyRedis Python client项目地址: https://gitcode.com/GitHub_Trending/re/redis-py创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考