redis 发布 订阅

1.服务端
package com.dz.im.tools;

import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;

/**
* 消息发布端
* @author David
*
*/
public class PubClient {

// Redis服务器IP
private static String ADDR = "192.168.0.11";

// Redis的端口号
private static int PORT = 6379;

// 如果赋值为-1，则表示不限制；如果pool已经分配了maxActive个jedis实例，则此时pool的状态为exhausted(耗尽)。
private static int MAX_ACTIVE = 500;

// 控制一个pool最多有多少个状态为idle(空闲的)的jedis实例，默认值也是8。
private static int MAX_IDLE = 100;

// 等待可用连接的最大时间，单位毫秒，默认值为-1，表示永不超时。如果超过等待时间，则直接抛出JedisConnectionException；
private static int MAX_WAIT = 1000;

// 当调用borrow Object方法时，是否进行有效性检查
private static Boolean TEST_ON_BORROW = false;

// 当调用return Object方法时，是否进行有效性检查
private static Boolean TEST_ON_RETURN = false;

// 超时时间
private static int TIMEOUT = 1000000;

private static JedisPool jedisPool = null;

/**
* 初始化Redis连接池
*/
private static void initialPool() {
try {
JedisPoolConfig config = new JedisPoolConfig();
config.setMaxActive(MAX_ACTIVE);
config.setMaxIdle(MAX_IDLE);
config.setMaxWait(MAX_WAIT);
config.setTestOnBorrow(TEST_ON_BORROW);
config.setTestOnReturn(TEST_ON_RETURN);
jedisPool = new JedisPool(config, ADDR, PORT, TIMEOUT);
} catch (Exception e) {
e.printStackTrace();
LoggerUtil.MyLogger.error("init jedispoll error ：%s", e);
}
}

/**
* 在多线程环境同步初始化
*/
private static synchronized void poolInit() {
if (jedisPool == null) {
initialPool();
}
}

/**
* 释放jedis资源
*
* @param jedis
*/
private static void returnResource(final Jedis jedis) {
if (jedis != null && jedisPool != null) {
jedisPool.returnResource(jedis);
}
}

/**
* 销毁连接
*
* @param shardedJedis
*/
private static void returnBrokenResource(final Jedis jedis) {
try {
jedisPool.returnBrokenResource(jedis);
} catch (Exception e) {
LoggerUtil.MyLogger.error("returnBrokenResource error", e);
}
}

/**
* 推入消息到redis消息通道
*
* @param channel
* @param message
*/
public void publish(String channel, String message) {
Jedis jedis = null;
if (jedisPool == null) {
poolInit();
}
try {
if (jedisPool != null) {
jedis = jedisPool.getResource();
jedis.publish(channel, message);
}
} catch (Exception ex) {
ex.printStackTrace();
returnBrokenResource(jedis);
} finally {
returnResource(jedis);
}
}

public void close(String channel){ 
Jedis jedis = null;
if (jedisPool == null) {
poolInit();
}
try {
if (jedisPool != null) {
jedis = jedisPool.getResource();
jedis.publish(channel, "quit");
jedis.del(channel);
}
} catch (Exception ex) {
ex.printStackTrace();
returnBrokenResource(jedis);
} finally {
returnResource(jedis);
}
} 

}

2.客户端
package com.dz.im.tools;

import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;
/**
* 消息订阅端
*
* @author David
*
*/
public class SubClient {

// Redis服务器IP
private static String ADDR = "192.168.0.11";

// Redis的端口号
private static int PORT = 6379;

// 如果赋值为-1，则表示不限制；如果pool已经分配了maxActive个jedis实例，则此时pool的状态为exhausted(耗尽)。
private static int MAX_ACTIVE = 500;

// 控制一个pool最多有多少个状态为idle(空闲的)的jedis实例，默认值也是8。
private static int MAX_IDLE = 100;

// 等待可用连接的最大时间，单位毫秒，默认值为-1，表示永不超时。如果超过等待时间，则直接抛出JedisConnectionException；
private static int MAX_WAIT = 1000;

// 当调用borrow Object方法时，是否进行有效性检查
private static Boolean TEST_ON_BORROW = false;

// 当调用return Object方法时，是否进行有效性检查
private static Boolean TEST_ON_RETURN = false;

// 超时时间
private static int TIMEOUT = 1000000;

private static JedisPool jedisPool = null;

/**
* 初始化Redis连接池
*/
private static void initialPool() {
try {
JedisPoolConfig config = new JedisPoolConfig();
config.setMaxActive(MAX_ACTIVE);
config.setMaxIdle(MAX_IDLE);
config.setMaxWait(MAX_WAIT);
config.setTestOnBorrow(TEST_ON_BORROW);
config.setTestOnReturn(TEST_ON_RETURN);
jedisPool = new JedisPool(config, ADDR, PORT, TIMEOUT);
} catch (Exception e) {
e.printStackTrace();
LoggerUtil.MyLogger.error("init jedispoll error ：%s", e);
}
}

/**
* 在多线程环境同步初始化
*/
private static synchronized void poolInit() {
if (jedisPool == null) {
initialPool();
}
}

/**
* 释放jedis资源
*
* @param jedis
*/
private static void returnResource(final Jedis jedis) {
if (jedis != null && jedisPool != null) {
jedisPool.returnResource(jedis);
}
}

/**
* 销毁连接
*
* @param shardedJedis
*/
private static void returnBrokenResource(final Jedis jedis) {
try {
jedisPool.returnBrokenResource(jedis);
} catch (Exception e) {
LoggerUtil.MyLogger.error("returnBrokenResource error", e);
}
}

/**
* 推入消息到redis消息通道
*
* @param channel
* @param message
*/
public void subscribe(JedisPubSub listener, String channel) {
Jedis jedis = null;
if (jedisPool == null) {
poolInit();
}
try {
if (jedisPool != null) {
jedis = jedisPool.getResource();
jedis.subscribe(listener, channel);
}
} catch (Exception ex) {
ex.printStackTrace();
returnBrokenResource(jedis);
} finally {
returnResource(jedis);
}
}
}

3.客户端监听
package com.dz.im.tools;

import java.text.SimpleDateFormat;
import java.util.Date;

import redis.clients.jedis.JedisPubSub;

/**
* 订阅者消息处理器
* @author David
*
*/
public class PrintListener extends JedisPubSub{

@Override
public void onMessage(String channel, String message) {
Date curDate=new Date();
SimpleDateFormat df=new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
String time = df.format(curDate);
        System.out.println("message receive:" + message + ",channel:" + channel + "..." + time); 
        //此处我们可以取消订阅 
        if(message.equalsIgnoreCase("quit")){ 
            this.unsubscribe(channel); 
        } 
}

@Override
public void onPMessage(String arg0, String arg1, String arg2) {

}

@Override
public void onPSubscribe(String arg0, int arg1) {

}

@Override
public void onPUnsubscribe(String arg0, int arg1) {

}

@Override
public void onSubscribe(String arg0, int arg1) {

}

@Override
public void onUnsubscribe(String arg0, int arg1) {

}
}
4.服务端测试类
package com.dz.im.tools;

import org.apache.commons.lang.RandomStringUtils;

public class PubSubTestMain {

public static void main(String[] args) throws Exception {
PubClient pubClient = new PubClient();
String channel = "public-test";
int i = 0;
while (i < 200) {
String message = "测试的所得税法大的"+RandomStringUtils.random(6, true, true);
pubClient.publish(channel, message);
i++;
Thread.sleep(1000);
}
}
}

 

5.客户端测试类

package com.dz.im.tools;

 

import redis.clients.jedis.JedisPubSub;

 

public class Demo {

 

private static final String channel="public-test";

 

public static void main(String[] args) {

try {

while(true){

SubClient subClient = new SubClient();

JedisPubSub listener = new PrintListener();

subClient.subscribe(listener, channel);

}

} catch (Exception e) {

e.printStackTrace();

}

}

}

 

参考http://shift-alt-ctrl.iteye.com/blog/1867454 这篇文章