Redis实现发布订阅模式-CFANZ编程社区

介绍

Redis安装和常用命令

发布订阅（又称pub/sub）的特点是是订阅者（listener）负责订阅频道（channel），发送者（publisher）负责向频道发送二进制字符串消息。每当有消息被发送至给定频道时，频道的所有订阅者都会收到消息。

订阅者可以同时订阅多个频道，发送者可以再多个频道发送消息。

相关的命令如下：

命令	用例和描述
PUBLISH	PUBLISH channel message，向给定频道发送消息
SUBSCRIBE	SUBSCRIBE channel [channel …]，订阅给定的一个或多个频道
UNSUBSCRIBE	UNSUBSCRIBE [channel [channel …]]，退订给定的一个或多个频道，如果执行时没有给定任何频道，那么退订所有频道
PSUBSCRIBE	PSUBSCRIBE pattern [pattern …]，订阅与给定模式相匹配的所有频道
PUNSUBSCRIBE	PUNSUBSCRIBE [pattern [pattern …]]，退订给定的模式，如果执行时没有给定任何模式，那么退订所有模式
PUBSUB	PUBSUB subcommand [argument [argument …]]，

PUBLISH，SUBSCRIBE ，UNSUBSCRIBE命令应该都比较好理解。
PSUBSCRIBE ，PUNSUBSCRIBE是用了类似正则表达式的方式

# it*匹配所有以it开头的频道，如it.news，it.blog

命令演示

打开第一个客户端，订阅一个channel

SUBSCRIBE container

打开第二个客户端，往channel中发送消息

PUBLISH container msg1
PUBLISH container msg2

此时第一个客户端输出第二个客户端发送的文字

第一个客户端

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

第二个客户端

127.0.0.1:6379> PUBLISH container msg1
(integer) 1
127.0.0.1:6379> PUBLISH container msg2
(integer) 1
127.0.0.1:6379>

代码实现

JedisUtil工具类

public class JedisUtil {
  private static volatile JedisUtil jedisUtil;
  private JedisPool jedisPool = this.initConfig();

  public JedisPool initConfig() {
    JedisPoolConfig config = new JedisPoolConfig();
    config.setMaxIdle(20);
    config.setMaxTotal(256);
    return new JedisPool(config,"127.0.0.1", 6379);
  }

  public Jedis getRedisClient() {
    return jedisPool.getResource();
  }

  public static JedisUtil getInstance() {
    if (jedisUtil == null) {
      synchronized (JedisUtil.class) {
        if (jedisUtil == null) {
          jedisUtil = new JedisUtil();
        }
      }
    }
    return jedisUtil;
  }
}

发布者

public class Publisher {

  public void publishMsg(String channel, String msg) {
    Jedis jedis = JedisUtil.getInstance().getRedisClient();
    jedis.publish(channel, msg);
  }
}

订阅者

public class Listener extends JedisPubSub {

  @Override
  public void onMessage(String channel, String message) {
    // 在这里输出channel名字和message名字
    System.out.println(channel + " " + message);
  }

  public void subscribe(String channel) {
    Jedis jedis = JedisUtil.getInstance().getRedisClient();
    new Thread(() -> {
      // 这是一个阻塞方法,所以在线程内订阅
      jedis.subscribe(this, channel);
    }).start();
  }
}

测试类

public class Main {

  public static final String CHANNEL_NAME = "demo_channel";

  public static void main(String[] args) {
    Listener listener = new Listener();
    listener.subscribe(CHANNEL_NAME);
    Publisher publisher = new Publisher();
    publisher.publishMsg(CHANNEL_NAME, "hello world");
  }
}