redis发布/订阅模式

大家都知道Redis中的list结构可以作为队列来满足一些生产消费的业务场景。实际上Redis还提供了发布/订阅（publish/subscribe）模式来实现类似的生产消费的功能。

list与发布/订阅的不同

1. list中的任务或消息无法被重复消费，消息被一个消费者pop 掉以后，其他消费者就获取不到了这个消息了。而发布/订阅模式中可以有多个订阅者消费同一个消息。
2. list可以保存任务或消息，直到客户端连接之后才消费掉。但发布/订阅模式中订阅者无法获取到订阅之前的历史消息，由于这个缺陷，在一些严格的生产消费场景下，建议还是用MQ的主题模式。

发布/订阅的使用

发布/订阅模式模式包含两种角色：发布者和订阅者。订阅者可以订阅一个或多个通道(channel)，而发布者可以向指定的通道(channel)中发送消息，所有此通道的订阅者都会收到消息。拿一个现实生活中的例子来比喻的话，发布者就像一个电台广播员，订阅者就像听众，广播员将消息通过将消息发送到一个频道中，而这个频道的听众就会听到播音员的声音。

下面通过几个命令来演示一下发布/订阅的效果

订阅通道

由于redis的订阅者无法获取到订阅之前通道中的消息，所以我们先让一个客户端订阅一个通道
订阅通道的命令：subscribe [channel...] 如果要订阅多个通道，就用空格分开

执行上面的命令客户端会进入订阅状态，进入订阅状态后客户端就会首先收到三个回复：
1) "subscribe"。 表示订阅成功的反馈信息。
2) "demoChannel"。 订阅成功的通道名称
3) "(integer) 1"。当前客户端订阅的通道数量

发布消息

我们再新开一个客户端，然后输入发布通知的命令：
publish

返回值(integer) 1 表示收到这条消息的订阅者数量。
然后我们再来看订阅者的客户端

此时我们发现界面上又多出了三行记录
1) "message"。表示接收到消息
2) "demoChannel"。表示收到消息的通道名称
3) "hello"。 表示消息的内容

通配符订阅

除了可以使用subscribe命令订阅通道之外，redis还支持通配符订阅。
命令的格式为：psubscribe [pattern…] 如果要订阅多个通道就用空格分开
通配符中?表示一个占位符，*表示任意个占位符（包括0个），?*表示一个以上占位符
我们在一个客户端中输入：
psubscribe d?mo czx* test?*

然后我们再去发布者的客户端中发布消息

上面的返回值(integer) 1 表示每个消息都有一个订阅者接收了。
再来看下订阅者的客户端

订阅者成功收到消息。

通配符订阅的注意事项

使用psubscribe命令可以重复订阅同一个通道，比如客户端执行了psubscribe test? test*，这时向 test1通道发布消息，订阅者就会收到两条消息，且publish返回值是2。

发布订阅的几个实际应用

Redis Sentinel 节点发现

Redis Sentinel（哨兵）是Redis官方推荐的一套高可用方案，在Redis主从同步的场景下，Redis Sentinel作为一个独立运行的进程，监控多个master-slave集群，当主节点故障的时候，自动将从节点提升为主节点，从而避免大面积瘫痪。
这里我们不详细研究Redis Sentinel 的原理，主要来看下Redis Sentinel如何使用发布订阅模式。
Redis Sentinel使用发布订阅模式，实现新节点的发现，以及交换主节点之间的状态。

如上图所示，每一个Sentinel会定时的向_sentinel_:hello 通道中发送信息，该信息包含Sentinel自己的IP、端口、ID等内容，以此来向其他Sentinel宣告自己的存在。同时，Sentinel也会通过订阅接收其他Sentinel的"HELLO"信息，以此来发现监视同一个主从集群的其他Sentinel。

Redission分布式锁

Redission开源框架提供了一些便捷的操作Redis的方法，其中比较出名的是基于Redis的分布式锁。
首先我们来看一下Redission加锁的用法

...
RLock redissonLock = redisson.getLock("xxxx");
redissonLock.lock();

RLock 继承自Java标准的Lock接口，调用lock()方法，就会判断xxxx这把锁是否被其他客户端获取到，如果是的话，就会线程阻塞并等待锁释放。
这里就有一个问题了，阻塞的线程如何接收到锁释放的通知呢？
没错，答案你们一定猜到了，就是利用了Redis的发布订阅模式。
当获取锁失败后，线程就会订阅redission_lock_channel_xxxx (xxxx代表锁的名称)通道，使用异步线程监听消息，然后利用Java中的SemaPhore使当前线程进入阻塞。
当锁被释放的时候，redission就会向redission_lock_channel_xxxx 这个通道中发布解锁的通知，异步线程收到消息，就会调用SemaPhore释放信号量，从而唤醒当前阻塞的线程去抢占锁。