Redis Pubsub命令用法

Posted 2020-08-27 Ruthless

一.什么是pub/sub及实现
Pub/Sub功能（means Publish, Subscribe）即发布及订阅功能。

Redis通过publish和subscribe命令实现订阅和发布的功能。

订阅者通过subscribe向redis server订阅自己感兴趣的消息类型。redis将信息类型称为通道(channel)。
当发布者通过publish命令向redis server发送特定类型的信息时，订阅该消息类型的全部订阅者都会收到此消息。

客户端1订阅CCTV1:

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

客户端2订阅CCTV1和CCTV2:

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

此时这两个客户端分别监听指定的频道。现在另一个客户端向服务器推送了这两个频道的信息。

127.0.0.1:6379> publish CCTV1 "cctv1 is good"
(integer) 2 //返回2表示两个客户端接收了该消息。

被接收到消息的客户端如下所示。
客户端1：

1) "message"
2) "CCTV1"
3) "cctv1 is good"

客户端2：

1) "message"
2) "CCTV1"
3) "cctv1 is good"

如上的订阅/发布也称订阅发布到频道(使用publish与subscribe命令)，此外还有订阅发布到模式(使用psubscribe来订阅一个模式)

订阅CCTV的全部频道

127.0.0.1:6379> psubscribe CCTV*
Reading messages... (press Ctrl-C to quit)
1) "psubscribe"
2) "CCTV*"
3) (integer) 1

当依然先如上推送一个CCTV1的消息时，该客户端正常接收。

二、Pub/Sub在Java中的实现
导入Redis驱动:

dependencies {
compile ‘redis.clients:jedis:2.4.2‘
}

Redis驱动包提供了一个抽象类:JedisPubSub…继承这个类就完成了对客户端对订阅的监听。示例代码:

package com.ljq.durian.test;

import org.apache.log4j.Logger;

import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPubSub;

/**
 * 客户端订阅监听类
 * 
 * @author jqlin
 *
 */
public class PubSubService extends JedisPubSub {
    private static final Logger logger = Logger.getLogger(PubSubService.class);
    
    /**
     * 监听到订阅频道接收到消息
     */
    @Override
    public void onMessage(String channel, String message) {
        logger.info(String.format("onSubscribe: channel[%s], " + "message[%s]", channel, message));
    }

    /**
     * 监听到订阅模式接收到消息
     */
    @Override
    public void onPMessage(String pattern, String channel, String message) {
        logger.info(String.format("onPMessage: pattern[%s], channel[%s], message[%s]", pattern, channel, message));
    }

    /**
     * 订阅频道时的回调
     * 
     */
    @Override
    public void onSubscribe(String channel, int subscribedChannels) {
        logger.info(String.format("onSubscribe: channel[%s], " + "subscribedChannels[%s]", channel, subscribedChannels));
    }

    /**
     * 取消订阅频道时的回调
     */
    @Override
    public void onUnsubscribe(String channel, int subscribedChannels) {
        logger.info(String.format("onUnsubscribe: channel[%s], " + "subscribedChannels[%s]", channel, subscribedChannels));
    }

    /**
     * 取消订阅模式时的回调
     */
    @Override
    public void onPUnsubscribe(String pattern, int subscribedChannels) {
        logger.info(String.format("onPUnsubscribe: pattern[%s], " + "subscribedChannels[%s]", pattern, subscribedChannels));
    }

    /**
     * 订阅频道模式时的回调
     */
    @Override
    public void onPSubscribe(String pattern, int subscribedChannels) {
        logger.info(String.format("onPSubscribe: pattern[%s], " + "subscribedChannels[%s]", pattern, subscribedChannels));
    }
    
    public static void main(String[] args) {
        Jedis jedis = null;
        try {
            jedis = new Jedis("127.0.0.1", 6379, 0);// redis服务地址和端口号
            PubSubService pubSub = new PubSubService();
            jedis.subscribe(pubSub, "news.share", "news.blog");
            
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            if (jedis != null) {
                jedis.disconnect();
            }
        }
    }
}

从代码中我们不难看出，我们声明的一个redis链接在设置监听后就可以执行一些操作，例如发布消息，订阅消息等。。。

当运行上述代码后会在控制台输出:

onSubscribe: channel[news.share],subscribedChannels[1]
onSubscribe: channel[news.blog],subscribedChannels[2]
//onSubscribe方法成功运行

此时当在有客户端向new.share或者new.blog通道publish消息时，onMessage方法即可被相应。(jedis.publish(channel, message))。