SUBSCRIBE study

PUBLISH study "message1-go go go"

PSUBSCRIBE study*

PUBLISH study "message2"

PUBLISH study:java "I hate java forever"

127.0.0.1:6379> PUBSUB channels

1) "study"

private Logger logger = LoggerFactory.getLogger(PublishMessage.class);

@Resource
private JedisCluster jedisCluster;

/**
 * 发布消息
 *
 * @param channel 频道
 * @param message 信息
 */
public void sendMessage(final String channel, final String message) {
  Thread thread = new Thread(() -> {
    Long publish = jedisCluster.publish(channel, message);
    logger.info("服务器在: {} 频道发布消息{} - {}", channel, message, publish);
  });
  logger.info("发布线程启动:");
  thread.setName("publishThread");
  thread.start();
}

private Logger logger = LoggerFactory.getLogger(ChatSubscribe.class);

// 取得订阅的消息后的处理
@Override
public void onMessage(String channel, String message) {
  logger.info("订阅成功,接收到的消息为:频道-{},消息-{}", channel, message);
  RedisString.message = message;
}

// 取得按表达式的方式订阅的消息后的处理
@Override
public void onPMessage(String pattern, String channel, String message) {
  System.out.println("-----取得按表达式的方式订阅的消息后的处理-----");
  System.out.println(pattern + "=" + channel + "=" + message);
}

// 初始化按表达式的方式订阅时候的处理
@Override
public void onPSubscribe(String pattern, int subscribedChannels) {
  System.out.println("-----初始化按表达式的方式订阅时候的处理-----");
  System.out.println(pattern + "=" + subscribedChannels);
}

// 取消按表达式的方式订阅时候的处理
@Override
public void onPUnsubscribe(String pattern, int subscribedChannels) {
  System.out.println("-----取消按表达式的方式订阅时候的处理-----");
  System.out.println(pattern + "=" + subscribedChannels);
}

@Override
public void onPong(String pattern) {
  super.onPong(pattern);
}

// 初始化订阅时候的处理
@Override
public void onSubscribe(String channel, int subscribedChannels) {
  logger.info("初始化订阅信息:频道-{},订阅频道-{}", channel, subscribedChannels);
}

// 取消订阅时候的处理
@Override
public void onUnsubscribe(String channel, int subscribedChannels) {
  logger.info("已取消订阅频道{}", channel);
}

private Logger logger = LoggerFactory.getLogger(SubScribeMessage.class);

private ExecutorService cachedThreadPool = Executors.newCachedThreadPool();

@Resource
private JedisCluster jedisCluster;
/**
   * 订阅频道
   *
   * @param channel     频道
   * @param roomSubListerner
   */
  public void subscribeChannel(final String channel, final ChatSubscribe roomSubListerner) {

    cachedThreadPool.execute(new Runnable() {
      @Override
      public void run() {
        jedisCluster.subscribe(roomSubListerner, channel);
      }
    });
  }

jedisCluster是否封装工具类，取自各位看官，核心代码已给出，请各位看官根据自身业务与逻辑，自行更改与优化代码。

本次示例程序采用tomcat 9.0 + spring + springmvc

使用了诸如：@RestController,@GetMapping等相关注解，便于开发，有兴趣可自行查阅spring相关资料。