redis-py的pubsub默认非阻塞，真正阻塞的是listen()方法；需用get_message()轮询或线程/async方案实现非阻塞监听，注意连接隔离、重连及版本兼容性。
redis-py 的 pubsub 对象默认就是非阻塞的，但监听方式决定是否“卡住”
很多人误以为 pubsub 一创建就自动监听，其实它只是个消息通道管理器。真正阻塞的是 listen() 方法——它内部是个无限迭代器，调用即阻塞主线程。要非阻塞，就不能直接调 listen()，而得手动轮询或结合事件循环。
常见错误现象：sub.listen() 一执行，后续代码全不运行；或者用 threading 硬套却没处理好连接复用，导致订阅失效或报 ConnectionError。
-
pubsub实例不能跨线程共享，每个线程需独立connection_pool或显式connection - 若用
threading.Thread启动监听，务必在子线程里调subscribe()+listen()，且避免主线程关闭Redis客户端时连带断开子线程连接 - 更稳妥的做法是用
get_message(ignore_subscribe_messages=True)主动轮询，配合time.sleep(0.01)控制 CPU 占用
用 get_message() 实现轻量级非阻塞监听
这是最可控的方式：不依赖线程/协程，适合嵌入已有主循环（如 Flask 后台任务、CLI 工具的 while 循环）。它返回 None 表示当前无消息，不会挂起。
注意点：
-
ignore_subscribe_messages=False（默认）时，首次subscribe()会收到{'type': 'subscribe', ...}类型消息，容易被当有效数据误处理 - 必须在调
get_message()前完成subscribe()，否则返回None且无提示 - 如果 Redis 连接中断，
get_message()可能抛ConnectionError，需在外层捕获并重连
import redis
r = redis.Redis()
ps = r.pubsub()
ps.subscribe('channel_a')
<p>while True:
msg = ps.get_message(ignore_subscribe_messages=True, timeout=0.1)
if msg and msg['type'] == 'message':
print(f"Received: {msg['data']}")</p><h1>其他业务逻辑可在此处执行</h1><p></p>
用 threading 实现后台持续监听时的关键约束
适用于需要长期守候、且不希望主逻辑被轮询干扰的场景。但线程内必须持有独立的 Redis 连接实例，不能复用主线程的客户端对象。
Python 3.14.2是Python编程语言在2025年12月5日发布的稳定版本，属于3.14系列的第二个维护更新。该版本包含了18项修复，重点解决了多进程、数据类及正则表达式等模块的回归问题，并修复了CVE-2025-12084等安全漏洞。此版本标志着自由线程模式（移除GIL）正式获得官方支持，是Python发展的重要里程碑。
典型坑：
- 主线程的
Redis实例被close()或 GC 回收后，子线程里的pubsub会静默失效，不再收消息 -
pubsub不自动重连，网络抖动后需手动调ps.reset()并重新subscribe() - Python 的 GIL 让多线程对 I/O 密集型任务影响不大，但若监听多个频道又做复杂解析，建议加
queue.Queue解耦消费
异步场景下用 aioredis（非 redis-py）才真正支持 async/await
redis-py 本身是同步库，所谓“async 支持”仅指兼容 asyncio 的连接池（如 ConnectionPool.from_url(..., decode_responses=True)），但 pubsub 模块至今（v4.6+）仍无原生 async 接口。
如果项目已用 asyncio，硬套 redis-py 的 pubsub 会导致整个 event loop 被阻塞。此时应切换到 aioredis（v2.x）或 redis-py 的继任者 redis（v4.5+ 中的 AsyncRedis + AsyncPubSub）。
例如：
import asyncio
from redis import Redis
r = Redis.from_url("redis://localhost")
ps = r.pubsub()
await ps.subscribe("channel_a") # 注意：这行在 v4.5+ 才可用
async for msg in ps.listen():
if msg["type"] == "message":
print(msg["data"])
这个 async for 形式才是真正的非阻塞监听，但前提是你的 redis 版本 ≥ 4.5 且服务端支持 RESP3。
旧版本强行用线程跑 listen() 再通过 asyncio.to_thread() 包装，属于绕路方案，延迟和资源开销都更高。
Python免费学习笔记（深入）：立即使用
在学习笔记中，你将探索 Python 的核心概念和高级技巧！
