Redis: Jedis 源代码剖析2- 发布者/订阅者模式剖析
2015-12-04 22:28
603 查看
在Jedis提供的发布者/订阅者模式中,最重要的就是JedisPubSub . 当订阅通道后,JedisPubSub就会轮询通道。直到通道取消订阅,才会继续执行。
因为Jedis是单线程,所以当订阅之后,线程就在循环。除非,在JedisPubSub收到消息后,新开线程继续执行任务。
在JedisPubSub 类中定义了接受消息后触发事件和轮询代码。JedisPubSub类是抽象类,所以为了实现自己的功能,必须继承该类。
public abstract class JedisPubSub {
private int subscribedChannels = 0;
private volatile Client client;
public void onMessage(String channel, String message) {
}
public void onPMessage(String pattern, String channel, String message) {
}
public void onSubscribe(String channel, int subscribedChannels) {
}
public void onUnsubscribe(String channel, int subscribedChannels) {
}
public void onPUnsubscribe(String pattern, int subscribedChannels) {
}
public void onPSubscribe(String pattern, int subscribedChannels) {
}
public void unsubscribe() {
if (client == null) {
throw new JedisConnectionException("JedisPubSub was not subscribed to a Jedis instance.");
}
client.unsubscribe();
client.flush();
}
public void unsubscribe(String... channels) {
if (client == null) {
throw new JedisConnectionException("JedisPubSub is not subscribed to a Jedis instance.");
}
client.unsubscribe(channels);
client.flush();
}
public void subscribe(String... channels) {
if (client == null) {
throw new JedisConnectionException("JedisPubSub is not subscribed to a Jedis instance.");
}
client.subscribe(channels);
client.flush();
}
public void psubscribe(String... patterns) {
if (client == null) {
throw new JedisConnectionException("JedisPubSub is not subscribed to a Jedis instance.");
}
client.psubscribe(patterns);
client.flush();
}
public void punsubscribe() {
if (client == null) {
throw new JedisConnectionException("JedisPubSub is not subscribed to a Jedis instance.");
}
client.punsubscribe();
client.flush();
}
public void punsubscribe(String... patterns) {
if (client == null) {
throw new JedisConnectionException("JedisPubSub is not subscribed to a Jedis instance.");
}
client.punsubscribe(patterns);
client.flush();
}
public boolean isSubscribed() {
return subscribedChannels > 0;
}
public void proceedWithPatterns(Client client, String... patterns) {
this.client = client;
client.psubscribe(patterns);
client.flush();
process(client);
}
public void proceed(Client client, String... channels) {
this.client = client;
client.subscribe(channels);
client.flush();
process(client);
}
//JedisPubSub执行一直轮询,直到取消订阅。
private void process(Client client) {
do {
List<Object> reply = client.getRawObjectMultiBulkReply();
final Object firstObj = reply.get(0);
if (!(firstObj instanceof byte[])) {
throw new JedisException("Unknown message type: " + firstObj);
}
final byte[] resp = (byte[]) firstObj;
if (Arrays.equals(SUBSCRIBE.raw, resp)) {
subscribedChannels = ((Long) reply.get(2)).intValue();
final byte[] bchannel = (byte[]) reply.get(1);
final String strchannel = (bchannel == null) ? null : SafeEncoder.encode(bchannel);
onSubscribe(strchannel, subscribedChannels);
} else if (Arrays.equals(UNSUBSCRIBE.raw, resp)) {
subscribedChannels = ((Long) reply.get(2)).intValue();
final byte[] bchannel = (byte[]) reply.get(1);
final String strchannel = (bchannel == null) ? null : SafeEncoder.encode(bchannel);
onUnsubscribe(strchannel, subscribedChannels);
} else if (Arrays.equals(MESSAGE.raw, resp)) {
final byte[] bchannel = (byte[]) reply.get(1);
final byte[] bmesg = (byte[]) reply.get(2);
final String strchannel = (bchannel == null) ? null : SafeEncoder.encode(bchannel);
final String strmesg = (bmesg == null) ? null : SafeEncoder.encode(bmesg);
onMessage(strchannel, strmesg);
} else if (Arrays.equals(PMESSAGE.raw, resp)) {
final byte[] bpattern = (byte[]) reply.get(1);
final byte[] bchannel = (byte[]) reply.get(2);
final byte[] bmesg = (byte[]) reply.get(3);
final String strpattern = (bpattern == null) ? null : SafeEncoder.encode(bpattern);
final String strchannel = (bchannel == null) ? null : SafeEncoder.encode(bchannel);
final String strmesg = (bmesg == null) ? null : SafeEncoder.encode(bmesg);
onPMessage(strpattern, strchannel, strmesg);
} else if (Arrays.equals(PSUBSCRIBE.raw, resp)) {
subscribedChannels = ((Long) reply.get(2)).intValue();
final byte[] bpattern = (byte[]) reply.get(1);
final String strpattern = (bpattern == null) ? null : SafeEncoder.encode(bpattern);
onPSubscribe(strpattern, subscribedChannels);
} else if (Arrays.equals(PUNSUBSCRIBE.raw, resp)) {
subscribedChannels = ((Long) reply.get(2)).intValue();
final byte[] bpattern = (byte[]) reply.get(1);
final String strpattern = (bpattern == null) ? null : SafeEncoder.encode(bpattern);
onPUnsubscribe(strpattern, subscribedChannels);
} else {
throw new JedisException("Unknown message type: " + firstObj);
}
} while (isSubscribed());
/* Invalidate instance since this thread is no longer listening */
this.client = null;
}
public int getSubscribedChannels() {
return subscribedChannels;
}
}
TestPub 为测试类,一直发布消息:
public class TestPub {
public static void main(String[] args) {
Jedis jedis = new Jedis();
while (true) {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
jedis.publish("channel", "Pub_Message");
}
}
}
TestSub消息进行订阅:
public class TestSub {
public static void main(String[] args)
{
Jedis jedis =new Jedis();
//创建一个订阅者
JedisPubSub listener =new JedisPubSubListener();
//订阅指定通道
jedis.subscribe(listener, "channel");
System.out.println("===subscribe success==");
}
}
因为Jedis是单线程,所以当订阅之后,线程就在循环。除非,在JedisPubSub收到消息后,新开线程继续执行任务。
在JedisPubSub 类中定义了接受消息后触发事件和轮询代码。JedisPubSub类是抽象类,所以为了实现自己的功能,必须继承该类。
public abstract class JedisPubSub {
private int subscribedChannels = 0;
private volatile Client client;
public void onMessage(String channel, String message) {
}
public void onPMessage(String pattern, String channel, String message) {
}
public void onSubscribe(String channel, int subscribedChannels) {
}
public void onUnsubscribe(String channel, int subscribedChannels) {
}
public void onPUnsubscribe(String pattern, int subscribedChannels) {
}
public void onPSubscribe(String pattern, int subscribedChannels) {
}
public void unsubscribe() {
if (client == null) {
throw new JedisConnectionException("JedisPubSub was not subscribed to a Jedis instance.");
}
client.unsubscribe();
client.flush();
}
public void unsubscribe(String... channels) {
if (client == null) {
throw new JedisConnectionException("JedisPubSub is not subscribed to a Jedis instance.");
}
client.unsubscribe(channels);
client.flush();
}
public void subscribe(String... channels) {
if (client == null) {
throw new JedisConnectionException("JedisPubSub is not subscribed to a Jedis instance.");
}
client.subscribe(channels);
client.flush();
}
public void psubscribe(String... patterns) {
if (client == null) {
throw new JedisConnectionException("JedisPubSub is not subscribed to a Jedis instance.");
}
client.psubscribe(patterns);
client.flush();
}
public void punsubscribe() {
if (client == null) {
throw new JedisConnectionException("JedisPubSub is not subscribed to a Jedis instance.");
}
client.punsubscribe();
client.flush();
}
public void punsubscribe(String... patterns) {
if (client == null) {
throw new JedisConnectionException("JedisPubSub is not subscribed to a Jedis instance.");
}
client.punsubscribe(patterns);
client.flush();
}
public boolean isSubscribed() {
return subscribedChannels > 0;
}
public void proceedWithPatterns(Client client, String... patterns) {
this.client = client;
client.psubscribe(patterns);
client.flush();
process(client);
}
public void proceed(Client client, String... channels) {
this.client = client;
client.subscribe(channels);
client.flush();
process(client);
}
//JedisPubSub执行一直轮询,直到取消订阅。
private void process(Client client) {
do {
List<Object> reply = client.getRawObjectMultiBulkReply();
final Object firstObj = reply.get(0);
if (!(firstObj instanceof byte[])) {
throw new JedisException("Unknown message type: " + firstObj);
}
final byte[] resp = (byte[]) firstObj;
if (Arrays.equals(SUBSCRIBE.raw, resp)) {
subscribedChannels = ((Long) reply.get(2)).intValue();
final byte[] bchannel = (byte[]) reply.get(1);
final String strchannel = (bchannel == null) ? null : SafeEncoder.encode(bchannel);
onSubscribe(strchannel, subscribedChannels);
} else if (Arrays.equals(UNSUBSCRIBE.raw, resp)) {
subscribedChannels = ((Long) reply.get(2)).intValue();
final byte[] bchannel = (byte[]) reply.get(1);
final String strchannel = (bchannel == null) ? null : SafeEncoder.encode(bchannel);
onUnsubscribe(strchannel, subscribedChannels);
} else if (Arrays.equals(MESSAGE.raw, resp)) {
final byte[] bchannel = (byte[]) reply.get(1);
final byte[] bmesg = (byte[]) reply.get(2);
final String strchannel = (bchannel == null) ? null : SafeEncoder.encode(bchannel);
final String strmesg = (bmesg == null) ? null : SafeEncoder.encode(bmesg);
onMessage(strchannel, strmesg);
} else if (Arrays.equals(PMESSAGE.raw, resp)) {
final byte[] bpattern = (byte[]) reply.get(1);
final byte[] bchannel = (byte[]) reply.get(2);
final byte[] bmesg = (byte[]) reply.get(3);
final String strpattern = (bpattern == null) ? null : SafeEncoder.encode(bpattern);
final String strchannel = (bchannel == null) ? null : SafeEncoder.encode(bchannel);
final String strmesg = (bmesg == null) ? null : SafeEncoder.encode(bmesg);
onPMessage(strpattern, strchannel, strmesg);
} else if (Arrays.equals(PSUBSCRIBE.raw, resp)) {
subscribedChannels = ((Long) reply.get(2)).intValue();
final byte[] bpattern = (byte[]) reply.get(1);
final String strpattern = (bpattern == null) ? null : SafeEncoder.encode(bpattern);
onPSubscribe(strpattern, subscribedChannels);
} else if (Arrays.equals(PUNSUBSCRIBE.raw, resp)) {
subscribedChannels = ((Long) reply.get(2)).intValue();
final byte[] bpattern = (byte[]) reply.get(1);
final String strpattern = (bpattern == null) ? null : SafeEncoder.encode(bpattern);
onPUnsubscribe(strpattern, subscribedChannels);
} else {
throw new JedisException("Unknown message type: " + firstObj);
}
} while (isSubscribed());
/* Invalidate instance since this thread is no longer listening */
this.client = null;
}
public int getSubscribedChannels() {
return subscribedChannels;
}
}
/** * Note that subscribe is a blocking operation because it will poll Redis for responses on the thread that calls subscribe. * A single JedisPubSub instance can be used to subscribe to multiple channels. * You can call subscribe or psubscribe on an existing JedisPubSub instance to change your subscriptions. * @author Administrator * */ public class JedisPubSubListener extends JedisPubSub { @Override public void onMessage(String channel, String message) { super.onMessage(channel, message); System.out.println("onMessage"); } @Override public void onPMessage(String pattern, String channel, String message) { // TODO Auto-generated method stub super.onPMessage(pattern, channel, message); System.out.println("onPMessage"); } @Override public void onSubscribe(String channel, int subscribedChannels) { // TODO Auto-generated method stub super.onSubscribe(channel, subscribedChannels); System.out.println("onSubscribe"); } @Override public void onUnsubscribe(String channel, int subscribedChannels) { // TODO Auto-generated method stub super.onUnsubscribe(channel, subscribedChannels); System.out.println("onUnsubscribe"); } @Override public void onPUnsubscribe(String pattern, int subscribedChannels) { // TODO Auto-generated method stub super.onPUnsubscribe(pattern, subscribedChannels); System.out.println("onPUnsubscribe"); } @Override public void onPSubscribe(String pattern, int subscribedChannels) { // TODO Auto-generated method stub super.onPSubscribe(pattern, subscribedChannels); System.out.println("onPSubscribe"); } }
TestPub 为测试类,一直发布消息:
public class TestPub {
public static void main(String[] args) {
Jedis jedis = new Jedis();
while (true) {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
jedis.publish("channel", "Pub_Message");
}
}
}
TestSub消息进行订阅:
public class TestSub {
public static void main(String[] args)
{
Jedis jedis =new Jedis();
//创建一个订阅者
JedisPubSub listener =new JedisPubSubListener();
//订阅指定通道
jedis.subscribe(listener, "channel");
System.out.println("===subscribe success==");
}
}
相关文章推荐
- php-redis扩展编译
- Nginx+Tomcat+Redis实现应用服务器集群负载均衡和Session共享
- Linux 安装redis 详解
- redis
- Redis数据库笔记
- redis的两种备份方式
- redis.conf 配置详解
- Ubuntu 安装 Redis
- Redis常用命令
- java对redis的基本操作
- redis命令
- yii2 redis 常用命令
- Redis 或弃用当前 VM 机制,采用新的 diskstore 模型
- redis(2)、redis数据类型
- Redis3.0.5 cluster--安装
- mac 安装phpredis扩展
- Redis总结(四)Redis 的持久化
- redis 基本 接口
- Zabbix 监控redis
- Windows下安装redis以及PHP redis扩展