我们为了自己实现一个MQ功能，就要深入底层挖掘现有开源产品的实现过程。

Redis 发布订阅底层结构解析

Redis 不存储消息，仅作为“实时中转”；只有订阅者在线时才能收到消息；消息是广播给所有订阅此频道的客户端。

1. 核心数据结构：哈希表（dict）+ 链表（链式客户端列表）

1.1 pubsub_channels：频道订阅表（dict）

Redis 使用一个全局字典（哈希表）结构来维护频道与订阅者之间的映射关系：

dict *pubsub_channels; // key: channel name (sds)，value: list of clients

• Key 是频道名称（channel），类型为 Redis 字符串（SDS）
• Value 是一个客户端链表（list *），保存所有订阅该频道的客户端连接指针

1.2 客户端链表（client list）

对于每个频道，对应一个链表或列表结构（底层用的是 Redis 自身的 list 类型）：

typedef struct client {...int fd;                // 客户端 socket fdlist *subscribed_channels; // 客户端自身也记录了订阅了哪些频道...
} client;

这个链表中每个节点是一个 client * 指针，表示一个订阅该频道的活跃客户端连接。

2. 消息发布流程

当执行 PUBLISH channel message 时，Redis 会：

1. 查找 pubsub_channels[channel]，得到订阅该频道的客户端链表
2. 遍历该链表，对每个客户端执行 addReply(client, message)
3. 消息会直接写入客户端输出缓冲区，等待发送至客户端 socket

如果客户端网络异常导致写失败，可能会断开连接并清理其订阅记录。

3. 订阅流程

订阅（SUBSCRIBE）：

• Redis 检查频道是否存在于 pubsub_channels：
  • 如果存在，将当前客户端加入该频道对应的客户端链表
  • 如果不存在，新建一个键值对：channel -> list，然后加入客户端
    

4、取消订阅流程（UNSUBSCRIBE）：

• Redis 从 pubsub_channels[channel] 的链表中移除该客户端
• 如果链表为空，则从字典中移除该频道
• 同时更新客户端的 subscribed_channels 属性

---

5、 客户端断开连接时的清理机制

当客户端断开连接，Redis 会调用清理逻辑：

pubsubUnsubscribeAllChannels(client *c)

该函数会遍历客户端 subscribed_channels，从所有频道对应的客户端链表中移除该客户端，并删除空频道。

• SUBSCRIBE：注册订阅关系。
• PUBLISH：消息发布，立即广播给在线订阅者。
• 掉线（连接断开）：Redis 会检测到客户端断连，并自动移除其订阅关系。
• UNSUBSCRIBE：客户端主动取消订阅，Redis 也会移除其订阅记录。

实践过程：