Redis 发布订阅（完整教程）

Redis 发布订阅：让消息在系统间“自由流动”

在现代软件架构中，服务之间的通信方式越来越多样化。传统的请求-响应模式虽然稳定，但在某些场景下显得笨重。比如，当多个模块需要实时感知某个事件发生时，如果每个模块都去轮询状态，不仅浪费资源，还容易导致延迟。这时候，Redis 发布订阅机制就成为了一个优雅的解决方案。

想象一下，你在一个大型社区里，每个居民都对特定的消息感兴趣，比如“小区停电通知”或“快递到货提醒”。如果每个居民都去物业办公室每天问一次，那显然效率极低。但如果有广播系统，物业一旦发布通知，所有关心的人立刻就能收到，这就是“发布订阅”思想的体现。Redis 的发布订阅功能，正是为这类场景量身打造的。

Redis 发布订阅的基本原理

Redis 提供了一种基于主题（channel）的消息传递机制，允许一个或多个客户端订阅特定的频道，当有消息发布到该频道时，所有订阅者都会收到通知。整个过程是异步的，不依赖于客户端的主动轮询。

这个机制的核心角色有三个：

• 发布者（Publisher）：负责向某个频道发送消息。
• 订阅者（Subscriber）：监听一个或多个频道，等待接收消息。
• 频道（Channel）：消息的传输通道，可以理解为一个“消息主题”。

Redis 本身不保存消息内容，一旦消息被发布出去，订阅者收到后就不再保留。这使得它特别适合用于实时通知、日志广播、事件驱动架构等场景。

举个生活化的例子：你订阅了“天气预报”频道，每天早上 7 点，气象局（发布者）会向该频道推送当天天气信息，你的手机（订阅者）会自动收到提醒，无需你手动查天气。

如何使用 Redis 发布订阅？

我们先用 Redis 命令行工具来演示最基础的使用方式。确保你已经安装并运行了 Redis 服务。

启动订阅者

打开两个终端窗口，一个作为订阅者，另一个作为发布者。

在第一个终端中执行以下命令：

SUBSCRIBE news

此时，终端会进入等待状态，显示：

Reading messages... (press Ctrl+C to quit)
1) "subscribe"
2) "news"
3) (integer) 1

表示你已成功订阅“news”频道，接下来所有发往该频道的消息都会被接收。

发布消息

在第二个终端中执行：

PUBLISH news "今天天气晴朗，适合出门散步"

返回值是 (integer) 1，表示有一名订阅者接收了消息。

回到第一个终端，你会看到：

1) "message"
2) "news"
3) "今天天气晴朗，适合出门散步"

说明消息已成功送达。

✅ 注意：SUBSCRIBE 是阻塞命令，必须保持连接。如果断开，就无法再接收消息。这是 Redis 发布订阅的一个特点，适合长期运行的监听服务。

实际应用：实时日志监控系统

下面我们用一个更贴近实际开发的案例，展示如何在项目中使用 Redis 发布订阅。

假设你正在开发一个 Web 应用，需要监控用户登录失败的次数。当某个用户的登录失败次数超过 5 次时，系统需要立即通知安全模块进行封禁。

我们可以这样设计：

1. 安全模块作为订阅者

import redis
import json

r = redis.Redis(host='localhost', port=6379, db=0)

pubsub = r.pubsub()
pubsub.subscribe('security_alert')

print("正在监听安全告警消息...")

for message in pubsub.listen():
    if message['type'] == 'message':
        alert_data = json.loads(message['data'])
        print(f"【安全告警】用户 {alert_data['user_id']} 登录失败次数过多！")
        # 可在此处执行封禁逻辑

✅ 注释：pubsub.listen() 会持续返回消息，我们通过判断 type 为 'message' 来提取有效数据。json.loads 用于解析结构化消息。

2. 登录服务作为发布者

import redis
import time

r = redis.Redis(host='localhost', port=6379, db=0)

def login_failed(user_id):
    # 模拟失败次数统计（实际应使用 Redis 计数器）
    fail_count = 3  # 假设当前失败 3 次
    
    # 如果超过阈值，发布告警
    if fail_count >= 5:
        alert_data = {
            "user_id": user_id,
            "event": "login_failed_threshold_exceeded",
            "timestamp": time.time()
        }
        # 发布到安全告警频道
        r.publish('security_alert', json.dumps(alert_data))
        print(f"已发布安全告警：用户 {user_id} 失败次数超限")

✅ 注释：这里我们用 json.dumps 将字典序列化为字符串，确保消息结构清晰。publish 是非阻塞的，调用后立即返回。

3. 运行效果

启动安全模块（订阅者），然后运行登录服务，模拟多次失败登录。当失败次数达到 5 次时，控制台会立即输出告警信息。

这种设计的优势在于：登录服务无需关心“谁要接收告警”，只需发布；而安全模块只需监听指定频道，实现解耦。

多频道订阅与模式匹配

Redis 不仅支持精确频道订阅，还支持模式订阅（pattern subscription），即可以监听符合某种规则的多个频道。

使用 PSUBSCRIBE 进行模式匹配

假设你有一个系统，不同地区有各自的监控频道，如：

• region.north.login
• region.south.login
• region.east.login

你想监听所有区域的登录事件，就可以使用模式匹配：

PSUBSCRIBE "region.*.login"

当有消息发布到 region.north.login 时，订阅者也会收到。

模式订阅的 Python 实现

import redis

r = redis.Redis(host='localhost', port=6379, db=0)
pubsub = r.pubsub()
pubsub.psubscribe('region.*.login')

print("正在监听区域登录事件...")

for message in pubsub.listen():
    if message['type'] == 'pmessage':
        print(f"【模式匹配】收到消息：频道={message['channel'].decode()}, 内容={message['data'].decode()}")

✅ 注释：pmessage 类型表示来自模式订阅的消息。decode() 用于将字节数据转为字符串。

⚠️ 注意：PSUBSCRIBE 不会自动重连，如果连接中断，需手动恢复。

优缺点与使用建议

优点

• 低延迟：消息几乎实时送达，适合事件驱动场景。
• 解耦：发布者和订阅者无需知道彼此的存在，系统更灵活。
• 轻量级：Redis 本身性能高，适合高频消息传递。

缺点

• 无持久化：消息一旦发出，未被接收就丢失。不适合重要消息。
• 无确认机制：无法保证订阅者一定收到。
• 连接依赖：订阅者必须保持连接，否则会漏消息。

使用建议

• 适合：实时通知、日志广播、事件总线、Webhook 通知。
• 不适合：需要持久化、重试、确认机制的业务消息（如订单支付成功）。

总结

Redis 发布订阅是一种简单却强大的通信模式，特别适合构建松耦合、高响应的系统。它像一个公共广播站，发布者只需“喊一声”，所有关心的订阅者都会立刻听见。

通过本文的讲解与代码示例，你应该已经掌握了如何在实际项目中使用 Redis 发布订阅。无论是实现日志监控、安全告警，还是构建事件驱动架构，它都能发挥重要作用。

记住：真正的系统设计，不在于“我做了多少”，而在于“谁听到了”。 Redis 发布订阅，正是让系统“听得清”的关键一环。