Redis队列（二）：消息的发布与订阅PubSub的使用

一、概述

消息发布订阅模式

• 在之前的文章中介绍了使用Redis列表这种数据类型来实现一个轻量级的消息队列，不过使用列表实现的消息队列存在一个缺陷就是由于是基于列表实现，所以消息出队列之后则不再存在，所以只能被一个消费者消费一次，不支持多个不同的消费者各消费一次，即不支持消息广播。
• 为了实现消息队列常见的消息发布订阅PubSub模式，在Redis中提供了消息的发布与订阅实现，即消息生产者客户端可以往某个指定的频道channel或者模式pattern发布一个消息，然后将这个消息广播给多个订阅了这个频道channel的客户端或者广播给订阅了该消息匹配的某个模式pattern的客户端。
• 所以Redis提供的是消息的发布与订阅，不是传统的消息队列实现，发布的消息并不会被存储，如Redis基于列表实现的消息队列会在消费之前存放在列表的链表数据结构里。Redis提供的消息订阅发布是实时的消息发布和订阅接收，如果消息所发往的频道channel或者模式pattern没有订阅者，则该条消息不会传给任何其他客户端，直接过掉或者说丢弃掉。

轻量级实现

• Redis所提供的消息订阅发布，可以理解为是一个轻量级的消息订阅发布实现，所谓轻量级是相对于RabbitMQ和Kafka这种专业的消息队列所提供的消息订阅发布而言的，即RabbitMQ和Kafka需要在服务器单独配置和启动服务端Broker进程来接收客户端的消息写入和消息读取消费，而Redis的消息订阅发布由于是Redis内置的，由于在项目中通常会使用Redis作为分布式缓存实现，所以不需要进行其他额外的配置和部署。
• 所以如果项目中刚开始没有使用RabbitMQ这种专业的队列，而项目后期又需要对项目进行解耦，需要用到消息的订阅发布功能，同时不想额外在生产服务器申请资源来部署RabbitMQ或者Kafka，则可以直接使用作为缓存的Redis所提供的消息发布订阅功能。
• 其中消息的发布订阅模式的一个应用场景为一个服务的多个部署实例使用本地缓存来缓存数据（不常更新），由于客户端可能连接任意一个部署实例，并对这个缓存进行更新，所以为了实现不同部署实例间的本地缓存数据同步，则这多个部署可以订阅同一个channel，当某个实例接收到客户端的更新请求时，更新后发布到该channel，从而通知到订阅了这个channel的其他部署实例。

二、使用方法

• 在使用层面，主要包括对频道channel的精确发布与订阅和对模式pattern的模式匹配发布与订阅。

基于Redis命令行使用

• 以下基于Redis的命令行对消息的订阅与发布功能进行演示，其中包含订阅频道channel和订阅模式pattern两种用法。

订阅频道channel

1. 如下首先在一个命令行订阅名为testChannel的频道，阻塞等待：

   127.0.0.1:6379> SUBSCRIBE testChannel
   Reading messages... (press Ctrl-C to quit)
   1) "subscribe"
   2) "testChannel"
   3) (integer) 1
   
   

2. 然后在另外一个命令行往testChannel频道发布一个消息：

   127.0.0.1:6379> PUBLISH testChannel "hello"
   (integer) 1
   

3. 在之前的订阅命令行接收到了这个发布的消息，并且继续阻塞等待：

   127.0.0.1:6379> SUBSCRIBE testChannel
   Reading messages... (press Ctrl-C to quit)
   1) "subscribe"
   2) "testChannel"
   3) (integer) 1
   1) "message"
   2) "testChannel"
   3) "hello"
   
   

订阅模式pattern

• 订阅模式主要是模糊匹配，如模式test* 则匹配所有以test开头的模式和频道channel的消息发布，即如果某个客户端订阅了模式 test*，则当另外一个客户端往testChannel这个频道发布了一条消息或者往test* 这个模式发布了一条消息，该客户端会收到消息，如下订阅test* 这个模式pattern，然后使用上面的命令往testChannel这个频道发布一条消息hello，则该客户端会收到消息，如下：

  127.0.0.1:6379> PSUBSCRIBE test*
  Reading messages... (press Ctrl-C to quit)
  1) "psubscribe"
  2) "test*"
  3) (integer) 1
  1) "pmessage"
  2) "test*"
  3) "testChannel"
  4) "hello"
  

基于Java客户端Jedis使用

• 在Java编程中，可以基于Redis的Java客户端Jedis来对消息订阅与发布功能进行使用，如果项目使用了spring-data-redis包提供的封装类来使用。以下使用Jedis进行一个简单演示：

  public void testReidsSub() {
      Jedis jedis = new Jedis("127.0.0.1", 6379);
      JedisPubSub pubSub = new JedisPubSub() {
          // 接收往频道channel发布的消息
          @Override
          public void onMessage(String channel, String message) {
              System.out.println("onMessage: " + message);
          }
  
          // 接收往模式pattern发布的消息
          @Override
          public void onPMessage(String pattern, String channel, String message) {
              System.out.println("onPMessage: " + message);
          }
  
          @Override
          public void onSubscribe(String channel, int subscribedChannels) {
  
          }
  
          @Override
          public void onUnsubscribe(String channel, int subscribedChannels) {
  
          }
  
          @Override
          public void onPUnsubscribe(String pattern, int subscribedChannels) {
  
          }
  
          @Override
          public void onPSubscribe(String pattern, int subscribedChannels) {
  
          }
      };
      // subscribe和psubscribe都是阻塞等待，故以下只有subscribe接收到消息，
      // 即只会在onMessage方法得到回调。
      
      // 订阅频道channel
      jedis.subscribe(pubSub, "testPubSub");
      // 订阅模式pattern
      jedis.psubscribe(pubSub, "test*");
      System.out.println("end...");
  }
  

• 当发布消息到某个频道channel时，由于模式pattern是模糊匹配，所以如果存在匹配的pattern，则订阅了这个pattern的客户端也会收到消息。由于Jedis的频道接收subscribe和模式接收psubscribe都是阻塞方法，所以只有一个会收到并在JedisPubSub方法的回调方法打印，所以在实际编程中，由于是阻塞方法，通常在不同的线程进行分别接收。