比如，如果有一个新的客户端 Client 08 要订阅 run 渠道，那么上图就会变成
如果 Client 08 要订阅一个新的渠道 new_sport ，那么就会变成
image-20191222161558999
整个订阅的过程可以采用下面伪代码来实现
Map<String, List<Object>> pubsub_channels = new HashMap<>(); public void subscribe(String[] subscribeList, Object client) { //遍历所有订阅的 channel，检查是否在 pubsub_channels 中，不在则创建新的 key 和空链表 for (int i = 0; i < subscribeList.length; i++) { if (!pubsub_channels.containsKey(subscribeList[i])) { pubsub_channels.put(subscribeList[i], new ArrayList<>()); } pubsub_channels.get(subscribeList[i]).add(client); } }
取消订阅
上面介绍的是单个 Channel 的订阅，相反的如果一个客户端要取消订阅相关 Channel，则无非是找到对应的 Channel 的链表，从中删除对应的客户端，如果该客户端已经是最后一个了，则将对应 Channel 也删除。
public void unSubscribe(String[] subscribeList, Object client) { //遍历所有订阅的 channel，依次删除 for (int i = 0; i < subscribeList.length; i++) { pubsub_channels.get(subscribeList[i]).remove(client); //如果长度为 0 则清楚 channel if (pubsub_channels.get(subscribeList[i]).size() == 0) { remove(subscribeList[i]); } } }
模式订阅结构
模式渠道的订阅与单个渠道的订阅类似，不过服务器是将所有模式的订阅关系都保存在服务器状态的pubsub_patterns 属性里面。
struct redisServer{ //保存所有模式订阅关系 list *pubsub_patterns; }
与订阅单个 Channel 不同的是，pubsub_patterns 属性是一个链表，不是字典。节点的结构如下：
struct pubsubPattern{ //订阅模式的客户端 redisClient *client; //被订阅的模式 robj *pattern; } pubsubPattern;
其实 client 属性是用来存放对应客户端信息，pattern 是用来存放客户端对应的匹配模式。
所以对应上面的 Client-06 模式匹配的结构存储如下
image-20191222174528367
在pubsub_patterns链表中有一个节点，对应的客户端是 Client-06，对应的匹配模式是run*。
订阅模式
当某个客户端通过命令psubscribe 订阅对应模式的 Channel 时候，服务器会创建一个节点，并将 Client 属性设置为对应的客户端，pattern 属性设置成对应的模式规则，然后添加到链表尾部。
对应的伪代码如下：
List<PubSubPattern> pubsub_patterns = new ArrayList<>(); public void psubscribe(String[] subscribeList, Object client) { //遍历所有订阅的 channel，创建节点 for (int i = 0; i < subscribeList.length; i++) { PubSubPattern pubSubPattern = new PubSubPattern(); pubSubPattern.client = client; pubSubPattern.pattern = subscribeList[i]; pubsub_patterns.add(pubSubPattern); } }
- 创建新节点；
- 给节点的属性赋值；
- 将节点添加到链表的尾部；
退订模式
退订模式的命令是punsubscribe，客户端使用这个命令来退订一个或者多个模式 Channel。服务器接收到该命令后，会遍历pubsub_patterns链表，将匹配到的 client 和 pattern 属性的节点给删掉。这里需要判断 client 属性和 pattern 属性都合法的时候再进行删除。
伪代码如下：
public void punsubscribe(String[] subscribeList, Object client) { //遍历所有订阅的 channel 相同 client 和 pattern 属性的节点会删除 for (int i = 0; i < subscribeList.length; i++) { for (int j = 0; j < pubsub_patterns.size(); j++) { if (pubsub_patterns.get(j).client == client && pubsub_patterns.get(j).pattern == subscribeList[i]) { remove(pubsub_patterns); } } } }
遍历所有的节点，当匹配到相同 client 属性和 pattern 属性的时候就进行节点删除。
发布消息
发布消息比较好容易理解，当一个客户端执行了publish channelName message 命令的时候，服务器会从pubsub_channels和pubsub_patterns 两个结构中找到符合channelName 的所有 Channel，进行消息的发送。在 pubsub_channels 中只要找到对应的 Channel 的 key 然后向对应的 value 链表中的客户端发送消息就好。
Redis 的持久化你了解吗
持久化是将程序数据在持久状态和瞬时状态间转换的机制。通俗的讲，就是瞬时数据（比如内存中的数据，是不能永久保存的）持久化为持久数据（比如持久化至数据库中，能够长久保存）。另外我们使用的 Redis 之所以快就是因为数据都存储在内存当中，为了保证在服务器出现异常过后还能恢复数据，所以就有了 Redis 的持久化，Redis 的持久化有两种方式，一种是快照形式 RDB，另一种是增量文件 AOF。
RDB
RDB 持久化方式是会在一个特定的时间间隔里面保存某个时间点的数据快照，我们拿到这个数据快照过后就可以根据这个快照完整的复制出数据。这种方式我们可以用来备份数据，把快照文件备份起来，传送到其他服务器就可以直接恢复数据。但是这只是某个时间点的全部数据，如果我们想要最新的数据，就只能定期的去生成快照文件。
RDB 的实现主要是通过创建一个子进程来实现 RDB 文件的快照生成，通过子进程来实现备份功能，不会影响主进程的性能。同时上面也提到 RDB 的快照文件是保存一定时间间隔的数据的，这就会导致如果时间间隔过长，服务器出现异常还没来得及生成快照的时候就会丢失这个间隔时间的所有数据；那有同学就会说，我们可以把时间间隔设置的短一点，适当的缩短是可以的，但是如果间隔时间段设置短一点频繁的生成快照对系统还是会有影响的，特别是在数据量大的情况下，高性能的环境下是不允许这种情况出现的。
我们可以在 redis.conf 进行 RDB 的相关配置，配置生成快照的策略，以及日志文件的路径和名称。还有定时备份规则，如下图所示，里面的注释写的很清楚，简单说就是在多少时间以内多少个 key 变化了就会触发快照。如save 300 10 表示在 5 分钟内如果有 10 个 key 发生了变化就会触发生产快照，其他的同理。
