分布式缓存Redis之发布/订阅（Pub/Sub）

写在前面

  本学习教程所有示例代码见GitHub：https://github.com/selfconzrr/Redis_Learning

一、简介

SUBSCRIBE、UNSUBSCRIBE和PUBLISH 三个命令实现了发布与订阅信息泛型（Publish/Subscribe messaging paradigm），在这个实现中， 发送者（发送信息的客户端）不是将信息直接发送给特定的接收者（接收信息的客户端）， 而是将信息发送给频道（channel）， 然后由频道将信息转发给所有对这个频道感兴趣的订阅者。也就是说发送者无须知道任何关于订阅者的信息， 而订阅者也无须知道是那个客户端给它发送信息， 它只要关注自己感兴趣的频道即可。

  对发布者和订阅者进行解构（decoupling），可以极大地提高系统的扩展性（scalability），并得到一个更动态的网络拓扑（network topology）。

  Redis 客户端可以订阅任意数量的频道。

  下图展示了频道 channel1，以及订阅这个频道的三个客户端 —— client2、client5和 client1之间的关系：

  当有新消息通过 PUBLISH 命令发送给频道 channel1 时， 这个消息就会被发送给订阅它的三个客户端：

二、原理

  RedisServer包含两个重要的结构：

• channels：实际上就是一个key-value的Map结构,key为订阅地频道，value为Client的List
• patterns:存放模式+client地址的列表

  **流程：**从pubsub_channels中找出跟publish中channel相符的clients-list，然后再去pubsub_patterns中找出每一个相符的pattern和client。向这些客户端发送publish的消息。

三、信息格式：

  频道转发的每条信息都是一条带有三个元素的多条批量回复（multi-bulk reply）。信息的第一个元素标识了信息的类型：

• subscribe ： 表示当前客户端成功地订阅了第二个元素所指示的频道，而信息的第三个元素则记录了目前客户端已订阅频道的总数。
• unsubscribe ： 表示当前客户端成功地退订了第二个元素所指示的频道，信息的第三个元素记录了客户端目前仍在订阅的频道数量。当客户端订阅的频道数量降为 0 时， 客户端不再订阅任何频道， 它可以像往常一样， 执行任何 Redis 命令。
• message ： 表示这条信息是由某个客户端执行 PUBLISH 命令所发送的真正的信息。 信息的第二个元素是信息来源的频道， 而第三个元素则是信息的内容。

  当然，Redis 的发布与订阅实现也支持模式匹配（pattern matching）： 客户端可以订阅一个带 * 号的模式， 如果某个/某些频道的名字和这个模式匹配， 那么当有信息发送给这个/这些频道的时候， 客户端也会收到这个/这些频道的信息。

redis > PSUBSCRIBE news.*

  客户端将收到来自 news.art.figurative 、 news.music.jazz 等频道的信息。

四、实例

  以下实例演示了发布订阅是如何工作的。在我们实例中我们订阅的频道为 redisChat 在客户端1执行

  然后重新开启个 redis 客户端，然后在同一个频道 redisChat 发布两次消息，订阅者就能接收到消息（在客户端1），返回成功发送到订阅者的数目：

再次切回客户端1：

退订频道：

  上面的代码简单的演示了订阅信道、向指定的信道发布消息、然后消息推送到订阅者以及取消订阅。

  Jedis中提供了JedisPubSub抽象类来提供发布/订阅的机制，在实际应用中需要实现JedisPubSub类。至于发布/订阅的java实现，且看我编写的实例代码：包含详细的注释

  https://github.com/selfconzrr/Redis_Learning

@Test
	public void testSubscribe() throws Exception{
		Jedis jedis = new Jedis("192.168.65.130", 6379); 
		jedis.auth("redis");
        RedisMsgPubSubListener listener = new RedisMsgPubSubListener();  
        jedis.subscribe(listener, "redisChatTest");  
        // other code
	}

@Test
	public void testPublish() throws Exception {
		Jedis jedis = new Jedis("192.168.65.130", 6379);
		jedis.auth("redis");
        jedis.publish("redisChatTest", "我是天才");  
        Thread.sleep(5000);  
        jedis.publish("redisChatTest", "我牛逼");  
        Thread.sleep(5000);  
        jedis.publish("redisChatTest", "哈哈");
	}

  运行成功后，在redis客户端执行pubsub channels查看当前活跃频道，即可看到在代码中订阅的频道“redisChatTest”

五、注意：

1、通过pattern模式而接收到的信息的类型为 pmessage ：

2、因为所有接收到的信息都会包含一个信息来源：当信息来自频道时，来源是某个频道；当信息来自模式时，来源是某个模式。因此， 客户端可以用一个哈希表，将特定来源和处理该来源的回调函数关联起来。 当有新信息到达时， 程序就可以根据信息的来源， 在 O(1) 复杂度内， 将信息交给正确的回调函数来处理。比如

  那么当有信息发送到频道 foo 时， 客户端将收到两条信息： 一条来自频道 foo ，信息类型为 message ； 另一条来自模式 f* ，信息类型为 pmessage 。

3、要在单独的线程中订阅，因为subscribe会阻塞当前线程的执行。你可以使用一个PubSub实例来订阅多个Channel。

4、一旦客户端进入订阅状态，客户端就只可接受订阅相关的命令SUBSCRIBE、PSUBSCRIBE、UNSUBSCRIBE和PUNSUBSCRIBE除了这些命令，其他命令一律失效。

5、使用PUNSUBSCRIBE命令只能退订通过PSUBSCRIBE命令订阅的规则，不会影响SUBSCRIBE订阅的频道。