redis pub/sub适合实时日志广播但不可靠，因其纯内存、无持久化、无ack、不保证顺序且消息易丢失；应作为入口网关，后接kafka/elasticsearch等持久化层，并规范json序列化与频道命名。
PUB/SUB 是 Redis 实现轻量级分布式日志收集最直接的路径，但它不是万能方案——它适合实时汇聚、低延迟分发，不适合日志持久化或高可靠性场景。
为什么不能只靠 PUBLISH 和 SUBSCRIBE 存日志？
Redis 的 PUB/SUB 是纯内存、无持久化的消息通道：一旦订阅者断连，离线期间所有消息彻底丢失；没有 ACK 机制，无法确认消费成功；也没有重试或回溯能力。这意味着：
- 日志中心进程重启后，会漏掉重启窗口内的全部日志
- 网络抖动或临时故障会导致日志“静默丢失”，且无迹可查
- 多个订阅者同时监听同一频道，每条消息会被广播给所有人（非队列语义），不适合做负载分摊
r.publish() 发送日志时必须序列化，且格式要统一
发送端不处理结构化，接收端就很难解析。常见错误是直接传 raw string 或拼接字符串，导致 JSON 解析失败或字段错位。
建议始终用 JSON 序列化，并包含必要元信息：
import json
import redis
<p>r = redis.Redis()
log_entry = {
"service": "auth-service",
"host": "node-03",
"level": "ERROR",
"timestamp": "2026-07-13T12:40:22.123Z",
"message": "Failed to validate token"
}
r.publish("logs", json.dumps(log_entry))</p>
关键点：
- 不要用 str(dict) 或 f-string 拼 JSON，必须用 json.dumps()
- 字段名保持小写、下划线风格，避免大小写混用引发解析歧义
- timestamp 必须 ISO 8601 格式（带时区），否则跨节点时间对齐困难
- 避免在消息体里塞二进制或未编码的特殊字符（如换行、控制符）
pubsub.listen() 接收端需处理连接中断与消息乱序
pubsub.listen() 返回的是一个阻塞迭代器，底层依赖 socket 连接。实际部署中常见问题包括：
- Redis 服务临时不可达 → 迭代器抛 ConnectionError，但默认不重连
- 网络延迟导致多条消息抵达顺序与发布顺序不一致（PUB/SUB 不保证顺序）
- item['type'] == 'message' 之外还有 'subscribe'、'unsubscribe' 类型，不判断会 crash
健壮写法示例：
Redis 缓存和数据结构管理技能。通过自然语言操作 Redis，支持 String、Hash、List、Set、ZSet、Stream 等数据结构操作。当用户提到 Redis、缓存、消息队列、会话存储时使用此技能。
def consume_logs():
r = redis.Redis(retry_on_timeout=True)
pubsub = r.pubsub()
pubsub.subscribe('logs')
<pre class="brush:php;toolbar:false;">while True:
try:
for item in pubsub.listen():
if item['type'] != 'message':
continue
try:
log_data = json.loads(item['data'].decode('utf-8'))
# 写入 Kafka / ES / 文件，或转发给下游处理器
process_log(log_data)
except (json.JSONDecodeError, UnicodeDecodeError):
# 跳过损坏消息，记录告警但不停止消费
print(f"Invalid log message: {item['data'][:50]}")
except redis.ConnectionError:
print("Redis connection lost, reconnecting...")
time.sleep(1)
pubsub.close()
pubsub = r.pubsub()
pubsub.subscribe('logs')
真正落地时，PUB/SUB 只应作为“第一跳”，后面必须接持久化层
把 PUB/SUB 当作日志系统的“入口网关”，而非存储终点。典型组合：
- 发布端 → Redis PUB/SUB → 消费端（常驻进程）→ Kafka 或 Logstash → Elasticsearch
- 或：发布端 → Redis PUB/SUB → 消费端 → r.lpush('log_queue', ...) → 后台 worker 拉取并落盘
这样既保留了 PUB/SUB 的低延迟广播优势，又通过后续环节补足了可靠性、追溯性、批量写入和水平扩展能力。
最容易被忽略的一点：频道名别硬编码成 'logs'。生产环境应按环境隔离（如 'logs-prod'、'logs-staging'），否则测试流量会污染线上日志流。
