Redis 异步 ClusterPubSub 首次订阅报 slots_cache KeyError

刚创建 redis.asyncio.cluster.RedisCluster，第一步就订阅频道，程序还没连到 Redis 就报 KeyError: 11554。调用栈落在 get_node_from_slot() 的 slots_cache[slot] 时，要先检查集群客户端是否完成初始化，而不是把这个数字当 Redis 返回的错误码。

这条路径在 redis-py 8.0.1 出现，正式 8.1.0 的异步 ClusterPubSub 仍会直接按频道查槽缓存。Redis 服务端健康，也不代表刚构造的 Python 对象已经拿到了槽表。

首次普通命令与首次订阅走的入口不同

集群客户端具有延迟初始化机制。普通命令会经过客户端的初始化路径；异步 ClusterPubSub 的首次 SUBSCRIBE 则可能先给频道算槽，再直接访问尚为空的缓存。这里还存在异常类型边界：8.1.0 的槽查询捕获 IndexError、TypeError，却没有把缺失字典键的 KeyError 转为 SlotNotCoveredError。

因此，“先执行过一次普通命令就正常”是有用线索，不是要求应用永远加一条无意义的业务查询。应把初始化放进真正的启动流程，并等待它完成后再启动订阅任务。

连接真实集群，等待初始化完成

以下示例需要一个已经配置好的 Redis Cluster，以及 Redis Cluster 对客户端公布的节点地址都可达。环境变量 REDIS_CLUSTER_URL 使用应用原有的连接地址与认证配置，不能填一个没有服务的占位端口来期待订阅成功。

import asyncio
import os
from redis.asyncio.cluster import RedisCluster

async def main():
    client = RedisCluster.from_url(os.environ['REDIS_CLUSTER_URL'])
    try:
        await client.initialize()
        async with client.pubsub() as subscriber:
            await subscriber.subscribe('app-events')
            async for message in subscriber.listen():
                if message['type'] == 'message':
                    print(message['data'])
    finally:
        await client.aclose()

asyncio.run(main())

app-events 是示例频道，真实应用必须与发布方保持一致；这里用 print 表示接收位置，不能把可能含敏感信息的业务消息原样新增到生产日志。示例是持续监听进程，需要沿用应用的任务取消与退出机制。

原系统若已经在启动阶段等待一次 PING，也可能借普通命令完成初始化；显式 initialize 更直接。不要仅创建协程而不 await，也不要一边初始化、一边让独立后台任务抢先 subscribe。长期运行的服务应共享自己管理的客户端生命周期，不要每收一条消息就重建连接。

缓存已经初始化，仍缺槽时要继续查集群

初始化可能因认证、TLS、连接超时或节点公布了不可达地址而失败。这时应保留原异常，处理实际部署问题；不能捕获所有异常后继续订阅，更不能把 KeyError 改名就当槽表已经完整。

后续采用其他正式版本时，需确认异步订阅入口是否已主动初始化，以及缺槽异常是否分别处理。不能只看版本号变大就删除应用需要的启动顺序。

在独立频道上核对新客户端首次订阅、收到确认与真实消息、取消监听后的资源释放，再检查断线重订阅。Redis Pub/Sub 不保存断开期间的消息；初始化恢复不提供历史补发，需要可靠投递的业务仍应保留原有持久化与确认设计。