redis的pub和sub

redis的pub和sub功能实现

顾名思义，可以分成pub和sub两个部分，先来说说sub

要实现redis的sub，首先必须新建一个实现了abstract 类 JedisPubSub的功能类

public class RedisTopicListener extends JedisPubSub {

private static Logger logger = LoggerFactory.getLogger(RedisTopicListener.class);

private RedisTopicProcessor redisMsgProcessor;

@Override

public void onMessage(String channel, String message) {

logger.trace("channel:{}, msg:{}", channel, message);

redisMsgProcessor.process(message); //这里调用了另一个类的process方法来实现收到channel的message消息时所做的操作

}

@Override

public void onPMessage(String pattern, String channel, String message) {

logger.trace("pattern:{}, channel:{}, msg:{}", new Object[]{pattern, channel, message});

}

@Override

public void onSubscribe(String channel, int subscribedChannels) {

logger.info("RedisTopicListener onSubscribe, channel:{}, subscribedChannels:{}",

new Object[]{channel, subscribedChannels});

}

@Override

public void onUnsubscribe(String channel, int subscribedChannels) {

logger.info("RedisTopicListener onUnsubscribe, channel:{}, subscribedChannels:{}",

new Object[]{channel, subscribedChannels});

}

@Override

public void onPUnsubscribe(String pattern, int subscribedChannels) {

logger.info("RedisTopicListener onPUnsubscribe, pattern:{}, subscribedChannels:{}",

new Object[]{pattern, subscribedChannels});

}

@Override

public void onPSubscribe(String pattern, int subscribedChannels) {

logger.info("RedisTopicListener onPSubscribe, pattern:{}, subscribedChannels:{}",

new Object[]{pattern, subscribedChannels});

}

public RedisTopicProcessor getRedisMsgProcessor() {

return redisMsgProcessor;

}

public void setRedisMsgProcessor(RedisTopicProcessor redisMsgProcessor) {

this.redisMsgProcessor = redisMsgProcessor;

}

}

在实现了该JedisPubSub虚拟类的时候，需要在另外的subscribe方法中调用该jedisPubSub类的实例，具体代码如下：

public Class classA{

private JedisPool jedisPool;

private JedisPubSub jedisPubSub;

private String lockFileName;

private String topicName;

public static void start(){

logger.info("start to transmit message from topic:{}", topicName);

Jedis jedis = jedisPool.getResource();

logger.info("subscribe topic:{}", topicName);

jedis.subscribe(jedisPubSub, topicName); 调用subscribe方法，从topic中读取数据

注意这里的jedisPubSub所注入的 是RedisTopicListener的实例。

}

}

public class classB{

public static void main(String[] args){

ClassA clsa = new ClassA();

clsa.start(); //在启动线程时，自动调用start方法

System.exit(0);

}

}

//如下为pub 部分

public class ClassC {

JedisPool jedisPool; //注意这里注入的jedisPool一定要和sub端监听的jedisPool一样

public void pub(){

Jedis jedis = jedisPool.getResource();

jedis.public("listen-Key", "1");

jedis.public("listen-Key", "2");

jedis.public("listen-Key", "3");

}

}

public class ClassD{

public static void main(String[] args){

ClassC clsc = new ClassC();

clsc.pub();

}

}