第0章：简介

（1）maven配置：

<dependency>
 <groupId>redis.clients</groupId>
 <artifactId>jedis</artifactId>
 <version>2.1.0</version>
 <type>jar</type>
 <scope>compile</scope>
 </dependency>

（2）简单测试

Jedis jedis = new Jedis("localhost");
jedis.set("foo","bar");
String value = jedis.get("foo");

（3）参考网站

Jedis官网：https://github.com/xetorthio/jedis

Jedis开源中国：http://www.oschina.net/p/jedis

大V博客：http://snowolf.iteye.com/blog/1633196

（4）札记

1）目前Redis大概有3中基于Java语言的Client，分别是Jredis，Jedis，Redis4J，这里只说Jedis，因为它是官方提供的唯一Redis Client For Java Provider。

第1章：Jedis实践

这里假设Redis已经正常运行，redis的具体安装可以参考“Redis基础”章节。下面就2.1.0版jedis作为redis的java客户端进行封装和测试，具体如下：

（1）Redis客户端（RedisClient.java）

package com.mcc.core.redis;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import redis.clients.jedis.*;
import redis.clients.jedis.exceptions.JedisConnectionException;
import redis.clients.util.Slowlog;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/**
* Redis客户端，基于Jedis,用到jedis包
* 单列模式, 建议应用与Redis集群部署在同一台服务器上，以获得最佳性能
*
* @author <a href="mailto:417877417@qq.com">menergy</a>
* DateTime: 13-12-21 下午2:09
*/
public class RedisClient {
 // members
 private static Logger logger = LoggerFactory.getLogger(RedisClient.class);
 // 单例
 private static RedisClient instance;
 // 默认超时时间
 private int defaultTimeout = 10000;
 // Redis集群，写节点，Master
 private JedisPool masterPool;
 // 默认Master IP
 private String masterHost = "127.0.0.1";
 // 默认端口
 private int masterPort = 6379;
 // Redis集群, 读节点，Slaves
 private Map<Integer, JedisPool> slavePools;
 //
 private static int poolIndex = 1;
 // 连接密码
 private String password;
 // 当前数据库
 private int database;
 // Redis Master Server 状态 -1 : 连接失败，0 : 未连接，1 : 连接成功
 private int writeStatus = 0;
 // Redis slave Server 状态 -1 : 连接失败，0 : 未连接，1 : 连接成功
 private int readStatus = 0;
 //
 private int delayRetry = 10;
 //
 private AliveEventListener aliveEventListener;
 //
 private ExecutorService executor;

 // static block

 // constructors

 /**
 * 默认构造器 
 * 当应用与Redis集群为同一台服务器时使用
 */
 private RedisClient() {
 this("127.0.0.1", 6379);
 }

 /**
 * 指定Master实例地址端口 
 * 当应用与Redis集群为不同服务器时使用，如开发测试时
 *
 * @param host Master IP
 * @param port Master PORT
 */
 private RedisClient(String host, int port) {
 this(host, port, null, 0);
 }

 /**
 * 指定Master实例地址端口，密码，数据库 
 * 适用一套Redis集群支撑多套应用
 *
 * @param host Master IP
 * @param port Master PORT
 * @param password Redis密码
 * @param database 数据库
 */
 private RedisClient(String host, int port, String password, Integer database) {
 this.masterHost = host;
 this.masterPort = port;
 this.password = password != null && password.isEmpty() ? null : password;
 this.database = database == null ? 0 : database;
 // 连接集群
 this.connectRedis();
 this.executor = Executors.newFixedThreadPool(2, new SubscribeThreadFactory());
 // 检查集群是否在线
 this.checkAlive();
 }

 // properties

 public int getDatabase() {
 return database;
 }

 /**
 * 是否可以写入状态
 *
 * @return 写入状态
 */
 public int getWriteStatus() {
 return writeStatus;
 }

 /**
 * 是否可以读入状态
 *
 * @return 读入状态
 */
 public int getReadStatus() {
 return readStatus;
 }

 /**
 * 存活事件
 *
 * @param aliveEventListener
 */
 public void setAliveEventListener(AliveEventListener aliveEventListener) {
 this.aliveEventListener = aliveEventListener;
 }

 // public methods

 /**
 * 获得取实例
 *
 * @return 返回一个Client实例
 */
 public static RedisClient getInstance() {
 if (instance == null) {
 instance = new RedisClient();
 }
 return instance;
 }

 /**
 * 获得取实例
 *
 * @param host Master IP
 * @param port Master PORT
 * @return 返回一个Client实例
 */
 public static RedisClient getInstance(String host, int port) {
 if (instance == null) {
 instance = new RedisClient(host, port);
 }
 return instance;
 }

 /**
 * 获得取实例
 *
 * @param host Master IP
 * @param port Master PORT
 * @param password 连接密码, 允许空
 * @return 返回一个Client实例
 */
 public static RedisClient getInstance(String host, int port, String password) {
 if (instance == null) {
 instance = new RedisClient(host, port, password, 0);
 }
 return instance;
 }

 /**
 * 获得取实例
 *
 * @param host Master IP
 * @param port Master PORT
 * @param password 连接密码, 允许空
 * @param database 数据库
 * @return 返回一个Client实例
 */
 public static RedisClient getInstance(String host, Integer port, String password, Integer database) {
 if (instance == null) {
 instance = new RedisClient(host, port, password, database);
 }
 return instance;
 }

 /**
 * 获得SlavePool
 *
 * @return 返回slave连接池
 */
 public JedisPool getSlavePool() {
 JedisPool pool = slavePools.get(poolIndex);
 poolIndex += 1;
 if (poolIndex > slavePools.size()) poolIndex = 1;
 return pool;
 }

 /**
 * 检查给定 key 是否存在
 *
 * @param key 键
 * @return 若 key 存在，返回 1 ，否则返回 0
 */
 public boolean exists(final String key) {
 Object actual = slave(new Action() {
 @Override
 public Object execute(final Jedis jedis) {
 return jedis.exists(key);
 }
 });
 return actual != null ? (Boolean) actual : false;
 }

 /**
 * 查看哈希表 key 中，给定域 field 是否存在
 *
 * @param key 键
 * @param field 域
 * @return 如果哈希表含有给定域，返回 1, 若不含有给定域，或 key 不存在，返回 0
 */
 public boolean hexists(final String key, final String field) {
 Object actual = slave(new Action() {
 @Override
 public Object execute(final Jedis jedis) {
 return jedis.hexists(key, field);
 }
 });
 return actual != null ? (Boolean) actual : false;
 }

 /**
 * 返回哈希表 key 中给定域 field 的值
 *
 * @param key 键
 * @param field 域
 * @return 给定域的值, 当给定域不存在或是给定 key 不存在时，返回 nil
 */
 public String hget(final String key, final String field) {
 Object actual = slave(new Action() {
 @Override
 public Object execute(final Jedis jedis) {
 return jedis.hget(key, field);
 }
 });
 return actual != null ? actual.toString() : null;
 }

 /**
 * 返回哈希表 key 中域的数量
 *
 * @param key 键
 * @return 哈希表中域的数量, 当 key 不存在时，返回 0
 */
 public Long hlen(final String key) {
 Object actual = slave(new Action() {
 @Override
 public Object execute(final Jedis jedis) {
 return jedis.hlen(key);
 }
 });
 return actual != null ? Long.valueOf(actual.toString()) : null;
 }

 /**
 * 返回哈希表 key 中的所有域
 *
 * @param key 键
 * @return 一个包含哈希表中所有域的表, 当 key 不存在时，返回一个空表
 */
 public Set<String> hkeys(final String key) {
 Object actual = slave(new Action() {
 @Override
 public Object execute(final Jedis jedis) {
 return jedis.hkeys(key);
 }
 });
 return actual != null ? (Set<String>) actual : null;
 }

 /**
 * 返回哈希表 key 中所有域的值
 *
 * @param key 键
 * @return 一个包含哈希表中所有值的表, 当 key 不存在时，返回一个空表
 */
 public List<String> hvals(final String key) {
 Object actual = slave(new Action() {
 @Override
 public Object execute(final Jedis jedis) {
 return jedis.hvals(key);
 }
 });
 return actual != null ? (List<String>) actual : null;
 }

 /**
 * 返回哈希表 key 中，所有的域和值
 *
 * @param key 键
 * @return 以列表形式返回哈希表的域和域的值, 若 key 不存在，返回空列表
 */
 public Map<String, String> hgetAll(final String key) {
 Object actual = slave(new Action() {
 @Override
 public Object execute(final Jedis jedis) {
 return jedis.hgetAll(key);
 }
 });
 return actual != null ? (Map<String, String>) actual : null;
 }

 /**
 * 查找所有符合给定模式 pattern 的 key 
 * <pre>
 * KEYS * 匹配数据库中所有 key 。
 * KEYS h?llo 匹配 hello ， hallo 和 hxllo 等。
 * KEYS h*llo 匹配 hllo 和 heeeeello 等。
 * KEYS h[ae]llo 匹配 hello 和 hallo ，但不匹配 hillo
 * </pre>
 *
 * @param pattern 模式
 * @return 符合给定模式的 key 列表
 */
 public Set<String> keys(final String pattern) {
 Object actual = master(new Action() {
 @Override
 public Object execute(final Jedis jedis) {
 return jedis.keys(pattern);
 }
 });
 return actual != null ? (Set<String>) actual : null;
 }

 /**
 * 删除给定的一个或多个 key 
 * 不存在的 key 会被忽略
 *
 * @param keys 缓存名称
 * @return 被删除 key 的数量
 */
 public Long del(final String... keys) {
 Object actual = master(new Action() {
 @Override
 public Object execute(final Jedis jedis) {
 return jedis.del(keys);
 }
 });
 return actual != null ? (Long) actual : null;
 }

 /**
 * 将信息 message 发送到指定的频道 channel
 *
 * @param channel 频道
 * @param command 指令
 * @param message 指令内容
 * @return
 */
 public Long publish(final String channel, final String command, final String message) {
 long actual = 0;
 if (command != null && message != null) {
 actual = publish(channel, String.format("%s %s", command, message));
 }
 return actual;
 }

 /**
 * 将信息 message 发送到指定的频道 channel
 *
 * @param channel 频道
 * @param message 信息
 * @return
 */
 public Long publish(final String channel, final String message) {
 Object actual = master(new Action() {
 @Override
 public Object execute(final Jedis jedis) {
 return jedis.publish(channel, message);
 }
 });
 return actual != null ? (Long) actual : 0;
 }

 /**
 * 从频道channels订阅信息message 
 * 订阅会阻塞进程，因此开一个新线程来执行, 注意不要频繁调用该方法
 *
 * @param handler 消息处理器
 * @param channels 频道
 */
 public void subscribe(final JedisPubSub handler, final String... channels) {
 // 订阅会阻塞进程，因此开一个新线程来执行
 executor.execute(new Runnable() {
 @Override
 public void run() {
 int delayRetry = 10;
 JedisShardInfo info = new JedisShardInfo(masterHost, masterPort, defaultTimeout);
 if (password != null) {
 info.setPassword(password);
 }
 // jedis.select(database); 每个Redis Server只有一套订阅，与数据库无关，不需要关注select 那个database
 while (true) {
 try {
 Jedis jedis = new Jedis(info);
 jedis.subscribe(handler, channels);
 } catch (JedisConnectionException e) {
 writeStatus = -1;
 if (logger.isErrorEnabled()) {
 logger.error("redis db connection reset ! please check it , {} second after reconnect !", delayRetry);
 }
 sleep(delayRetry);
 }
 }
 }
 });
 }

 /**
 * 查看Redis数据库连接是否正常
 *
 * @return 返回是否正常
 */
 public boolean pingMaster() {
 Object actual = master(new Action() {
 @Override
 public Object execute(final Jedis jedis) {
 String pong = null;
 try {
 pong = jedis.ping();
 } catch (Exception ignored) {
 }
 return pong;
 }
 });
 return true;
 }

 /**
 * 查看Redis数据库连接是否正常
 *
 * @return 返回是否正常
 */
 public boolean pingSlave() {
 Object actual = slave(new Action() {
 @Override
 public Object execute(final Jedis jedis) {
 String pong = null;
 try {
 pong = jedis.ping();
 } catch (Exception ignored) {
 }
 return pong;
 }
 });
 return true;
 }

 /**
 * 返回当前数据库的 key 的数量
 *
 * @return 当前数据库的 key 的数量
 */
 public Long dbSize() {
 Object actual = master(new Action() {
 @Override
 public Object execute(final Jedis jedis) {
 return jedis.dbSize();
 }
 });

 return actual != null ? (Long) actual : null;
 }

 /**
 * 执行一个 AOF文件重写操作
 *
 * @return Status code reply
 */
 public String bgrewriteaof() {
 Object actual = master(new Action() {
 @Override
 public Object execute(final Jedis jedis) {
 return jedis.bgrewriteaof();
 }
 });
 return actual != null ? (String) actual : null;
 }

 /**
 * 保存当前数据库的数据到磁盘
 *
 * @return Status code reply
 */
 public String bgsave() {
 Object actual = master(new Action() {
 @Override
 public Object execute(final Jedis jedis) {
 return jedis.bgsave();
 }
 });
 return actual != null ? (String) actual : null;
 }

 /**
 * 将当前 Redis 实例的所有数据快照(snapshot)以 RDB 文件的形式保存到硬盘(采用同步方式，一般不用)
 *
 * @return Status code reply
 */
 public String save() {
 Object actual = master(new Action() {
 @Override
 public Object execute(final Jedis jedis) {
 return jedis.save();
 }
 });
 return actual != null ? (String) actual : null;
 }

 /**
 * 获得最近一次 Redis 成功将数据保存到磁盘的时间
 *
 * @return Status code reply
 */
 public Long lastSave() {
 Object actual = master(new Action() {
 @Override
 public Object execute(final Jedis jedis) {
 return jedis.lastsave();
 }
 });
 return actual != null ? (Long) actual : null;
 }

 /**
 * 将当前服务器转变为指定服务器的从属服务器(slave server)
 *
 * @param host 从属服务器ip
 * @param port 从属服务器端口
 * @return Status code reply
 */
 public String slaveof(final String host, final Integer port) {
 Object actual = slave(new Action() {
 @Override
 public Object execute(final Jedis jedis) {
 return jedis.slaveof(host, port);
 }
 });
 return actual != null ? (String) actual : null;
 }

 /**
 * 关闭从属服务器的复制功能
 *
 * @return Status code reply
 */
 public String slaveofNoOne() {
 Object actual = slave(new Action() {
 @Override
 public Object execute(final Jedis jedis) {
 return jedis.slaveofNoOne();
 }
 });
 return actual != null ? (String) actual : null;
 }

 /**
 * 清空整个 Redis 服务器的数据(删除所有数据库的所有 key ,此命令从不失败)
 *
 * @return Status code reply
 */
 public String flushAll() {
 Object actual = master(new Action() {
 @Override
 public Object execute(final Jedis jedis) {
 return jedis.flushAll();
 }
 });
 return actual != null ? (String) actual : null;
 }

 /**
 * 清空当前数据库中的所有 key(此命令从不失败)
 *
 * @return Status code reply
 */
 public String flushDB() {
 Object actual = master(new Action() {
 @Override
 public Object execute(final Jedis jedis) {
 return jedis.flushDB();
 }
 });
 return actual != null ? (String) actual : null;
 }

 /**
 * 停止所有客户端
 *
 * @return Status code reply
 */
 public String shutdown() {
 Object actual = master(new Action() {
 @Override
 public Object execute(final Jedis jedis) {
 return jedis.shutdown();
 }
 });
 return actual != null ? (String) actual : null;
 }

 /**
 * 获取查询日志
 *
 * @return Status code reply
 */
 public List<Slowlog> slowlogGet() {
 Object actual = slave(new Action() {
 @Override
 public Object execute(final Jedis jedis) {
 return jedis.slowlogGet();
 }
 });
 return actual != null ? (List<Slowlog>) actual : null;
 }

 /**
 * 获取指定数量的查询日志
 *
 * @param entries 日志数量
 * @return 日志
 */
 public List<Slowlog> slowlogGet(final Long entries) {
 Object actual = slave(new Action() {
 @Override
 public Object execute(final Jedis jedis) {
 return jedis.slowlogGet(entries);
 }
 });
 return actual != null ? (List<Slowlog>) actual : null;
 }

 /**
 * 查看当前日志的数量
 *
 * @return 当前日志的数量
 */
 public Long slowlogLen() {
 Object actual = slave(new Action() {
 @Override
 public Object execute(final Jedis jedis) {
 return jedis.slowlogLen();
 }
 });
 return actual != null ? (Long) actual : null;
 }

 /**
 * 清空日志
 *
 * @return 清空日志
 */
 public byte[] slowlogReset() {
 Object actual = master(new Action() {
 @Override
 public Object execute(final Jedis jedis) {
 return jedis.slowlogReset();
 }
 });
 return actual != null ? (byte[]) actual : null;
 }

 /**
 * 返回关于 Redis 服务器的各种信息和统计值
 *
 * @return Redis 服务器的各种信息和统计值
 */
 public String info() {
 Object actual = master(new Action() {
 @Override
 public Object execute(final Jedis jedis) {
 return jedis.info();
 }
 });
 return actual != null ? (String) actual : null;
 }

 /**
 * 获取服务器的配置参数
 *
 * @param pattern 参数
 * @return 参数表
 */
 public List<String> configGet(final String pattern) {
 Object actual = slave(new Action() {
 @Override
 public Object execute(final Jedis jedis) {
 return jedis.configGet(pattern);
 }
 });
 return actual != null ? (List<String>) actual : null;
 }

 /**
 * 动态改变服务器参数（无需重启，即刻生效）
 *
 * @param parameter 参数名
 * @param value 参数值
 * @return Status code reply
 */
 public String configSet(final String parameter, final String value) {
 Object actual = master(new Action() {
 @Override
 public Object execute(final Jedis jedis) {
 return jedis.configSet(parameter, value);
 }
 });
 return actual != null ? (String) actual : null;
 }

 /**
 * 重置 INFO 命令中的某些统计数据
 *
 * @return Status code reply
 */
 public String configResetstart() {
 Object actual = master(new Action() {
 @Override
 public Object execute(final Jedis jedis) {
 return jedis.configResetStat();
 }
 });
 return actual != null ? (String) actual : null;
 }

 /**
 * Redis查询，保证只从Slave查询
 *
 * @param action 操作
 * @return 返回查询结果
 */
 public Object slave(Action action) {
 Object actual = null;
 if (slaveOnline()) {
 JedisPool pool = null;
 Jedis jedis = null;
 try {
 pool = this.getSlavePool();
 if (pool != null) {
 jedis = pool.getResource();
 actual = action.execute(jedis);
 } else {
 this.readStatus = -1;
 }
 } catch (JedisConnectionException e) {
 int osize = slavePools.size();
 initSlaves(getDefaultConfig());
 int nsize = slavePools.size();
 this.readStatus = slavePools.isEmpty() ? -1 : 1;
 if (osize != nsize && logger.isErrorEnabled()) {
 logger.error("slave instance size changed! {} to {}", osize, nsize);
 }
 } finally {
 if (pool != null) {
 pool.returnResource(jedis);
 }
 }
 }
 return actual;
 }

 /**
 * 更新或插入数据，只从Master写入数据 
 * 若要使用方式Jedis的select方法临时切换数据库，请使用该方法
 *
 * @param action 操作
 * @return 返回操作结果
 */
 public Object master(Action action) {
 Object actual = null;
 if (masterOnline()) {
 JedisPool pool = null;
 Jedis jedis = null;
 try {
 pool = this.getMasterPool();
 jedis = pool.getResource();
 actual = action.execute(jedis);
 } catch (JedisConnectionException e) {
 this.writeStatus = -1;
 if (logger.isErrorEnabled()) {
 logger.error("redis db connection reset ! please checkAlive it !");
 }
 } finally {
 if (pool != null) {
 pool.returnResource(jedis);
 }
 }
 }
 return actual;
 }

 // protected methods

 /**
 * 获得MasterPool
 *
 * @return
 */
 public JedisPool getMasterPool() {
 return masterPool;
 }

 // friendly methods

 // private methods

 /**
 * master配置
 *
 * @param config 配置
 */
 private void initMaster(JedisPoolConfig config) {
 try {
 masterPool = new JedisPool(config, this.masterHost, this.masterPort, this.defaultTimeout, this.password, this.database);
 Jedis jedis = masterPool.getResource();
 jedis.ping();
 masterPool.returnResource(jedis);
 writeStatus = 1;
 if (logger.isInfoEnabled()) {
 logger.info("redis master connected !");
 }
 } catch (JedisConnectionException e) {
 writeStatus = -1;
 if (logger.isErrorEnabled()) {
 logger.error("redis master connection reset ! please check it , {} second after reconnect !", delayRetry);
 }
 }
 }

 /**
 * slave配置
 *
 * @param config 配置
 */
 private void initSlaves(final JedisPoolConfig config) {
 if (masterPool != null) {
 slavePools = new HashMap<Integer, JedisPool>();
 master(new Action() {
 @Override
 public Object execute(Jedis jedis) {
 String info = jedis.info();
 String[] lines = info.split("\r\n");
 for (String line : lines) {
 if (line.contains("slave") && line.contains("online")) {
 String[] online = line.split(":")[1].split(",");
 String host = online[0];
 String port = online[1];
 // 处理当应用与redis不在同一台服务器上的情况
 if("127.0.0.1".equalsIgnoreCase(host)){
 host = masterHost;
 }
 jedis.hset("redis.slave", host, port);
 JedisPool pool = null;
 Jedis slave = null;
 try {
 pool = new JedisPool(config, host, Integer.valueOf(port), defaultTimeout, password, database);
 slave = pool.getResource();
 slavePools.put(slavePools.size() + 1, pool);
 } catch (JedisConnectionException e) {
 if (logger.isErrorEnabled()) {
 logger.error("slave: {}", e.getMessage());
 }
 } finally {
 if (pool != null && slave != null) {
 pool.returnResource(slave);
 }
 }
 }
 }
 // 更新slave节点状态
 readStatus = !slavePools.isEmpty() ? 1 : -1;
 switch (readStatus) {
 case -1:
 if (logger.isErrorEnabled()) {
 logger.error("redis slave connection reset ! please check it , {} second after reconnect !", delayRetry);
 }
 break;
 case 1:
 if (logger.isInfoEnabled()) {
 logger.info("redis slave connected!");
 }
 break;
 }
 return readStatus;
 }
 });
 }
 }

 /**
 * 返回一个默认的JedisPoolConfig
 *
 * @return 返回一个默认的JedisPoolConfig
 */
 private JedisPoolConfig getDefaultConfig() {
 // 连接配置
 JedisPoolConfig config = new JedisPoolConfig();
 config.setMaxActive(100);
 config.setMaxIdle(20);
 config.setMaxWait(10000);
 // 设定在借出对象时是否进行有效性检查
 config.setTestOnBorrow(true);
 return config;
 }

 /**
 * 连接Redis Cluster
 */
 private void connectRedis() {
 if (writeStatus != 1) {
 initMaster(getDefaultConfig());
 initSlaves(getDefaultConfig());
 }
 if (readStatus != 1) {
 initSlaves(getDefaultConfig());
 }
 // 解发事件
 if (this.aliveEventListener != null) {
 this.aliveEventListener.onEvents(masterOnline(), slaveOnline());
 }
 }

 /**
 * 检查集群是否存活
 *
 * @return
 */
 private void checkAlive() {
 executor.execute(new Runnable() {
 @Override
 public void run() {
 while (true) {
 try {
 connectRedis();
 sleep(delayRetry);
 } catch (JedisConnectionException e) {
 if (logger.isErrorEnabled()) {
 logger.error("redis db connection reset ! please check it , {} second after reconnect !", delayRetry);
 }
 sleep(delayRetry);
 }
 }
 }
 });
 }

 /**
 * Master连接池是否在线
 *
 * @return 若没初化成功则返回 false, 若成功则返回 true
 */
 private boolean masterOnline() {
 return this.writeStatus == 1;
 }

 /**
 * Slave连接池是否在线
 *
 * @return 若没初化成功则返回 false, 若成功则返回 true
 */
 private boolean slaveOnline() {
 return this.readStatus == 1;
 }

 /**
 * 休眠指定时间
 *
 * @param second 休眠秒数
 */
 private void sleep(int second) {
 try {
 TimeUnit.SECONDS.sleep(second);
 } catch (InterruptedException ignored) {
 }
 }

 // inner class

 /**
 * Redis Cluster 存活状态发生变化事件监听器
 */
 public interface AliveEventListener {
 /**
 * Redis Cluster 存活状态发生变化事件 
 * 每十秒触发一次
 *
 * @param masterAlive Redis Master Server是否存活
 * @param slaveAlive Redis Slave Server 是否存活
 */
 void onEvents(boolean masterAlive, boolean slaveAlive);
 }
 // test main

}

（2）Jedis查询接口（Action.java）

package com.mcc.core.redis;

import redis.clients.jedis.Jedis;

/**
* Jedis查询接口
*
* @author <a href="mailto:417877417@qq.com">menergy</a>
* DateTime: 13-12-21 下午2:28
*/
public interface Action {
 /**
 * Redis 通用查询
 *
 * @param jedis redis客户端
 * @return
 */
 Object execute(final Jedis jedis);
}

（3）Redis命令的发布订阅（CommandHandler.java）

package com.mcc.core.redis;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import redis.clients.jedis.JedisPubSub;

/**
* 通过Redis的发布订阅机制实现实时指令控制台
*
* @author <a href="mailto:417877417@qq.com">menergy</a>
* DateTime: 13-12-21 下午2:34
*/
public class CommandHandler extends JedisPubSub {

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

 //频道，也可以用配置文件配置
 public static final String COMMAND_CHANNEL = "execute";
 //命令，可以自定义命令
 public static final String COMMAND = "loadall";

 /**
 * 指令频道名称
 */
 private String channel;

 /**
 * 指令消息
 */
 private String message;

 public CommandHandler() {

 }

 /**
 * 处理指令消息事件
 *
 * @param channel 指令频道名称
 * @param message 指令消息
 */
 @Override
 public void onMessage(String channel, String message) {
 this.channel = channel;
 this.message = message;
 try {
 if (channel != null && channel.equalsIgnoreCase(COMMAND_CHANNEL)) {
 if (message != null && !message.isEmpty()) {
 String cmd = message, content = "";
 int length = message.length(), delimiter = message.indexOf(" ");
 if (delimiter != -1) {
 int start = delimiter + 1 > length ? length : delimiter + 1;
 cmd = message.substring(0, delimiter); // 指令
 content = message.substring(start, length); // 指令内容
 }
 // 更新全部缓存
 if (cmd.equalsIgnoreCase(COMMAND)) {
 logger.info("该命令下的操作逻辑");
 //执行该命令下的相应操作，具体这里暂不实现
 }
 }
 }
 } catch (Exception e) {
 if (logger.isErrorEnabled()) {
 logger.error("failure to execute command '{}' , cause: {}!", message, e.getMessage());
 }
 }
 }

 @Override
 public void onPMessage(String pattern, String channel, String message) {
 // TODO Auto-generated method stub

 }

 @Override
 public void onPSubscribe(String pattern, int subscribedChannels) {
 // TODO Auto-generated method stub

 }

 @Override
 public void onPUnsubscribe(String pattern, int subscribedChannels) {
 // TODO Auto-generated method stub

 }

 @Override
 public void onSubscribe(String channel, int subscribedChannels) {
 // TODO Auto-generated method stub

 }

 @Override
 public void onUnsubscribe(String channel, int subscribedChannels) {
 // TODO Auto-generated method stub

 }

 public String getChannel() {
 return channel;
 }

 public void setChannel(String channel) {
 this.channel = channel;
 }

 public String getMessage() {
 return message;
 }

 public void setMessage(String message) {
 this.message = message;
 }
}

（4）异常处理类（SubscribeThreadFactory.java）

package com.mcc.core.redis;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.ThreadFactory;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

/**
* 带异常处处理器的ThreaFactory
*
* @author <a href="mailto:417877417@qq.com">menergy</a>
* DateTime: 13-12-21 下午2:19
*/
public class SubscribeThreadFactory implements ThreadFactory {

 // members
 private static Logger logger = LoggerFactory.getLogger(SubscribeThreadFactory.class);
 static final AtomicInteger poolCounter = new AtomicInteger(1);
 final AtomicLong threadCounter = new AtomicLong(1);
 final ThreadGroup group;
 final String namingPattern;
 // static block

 // constructors
 public SubscribeThreadFactory() {
 SecurityManager securityManager = System.getSecurityManager();
 group = (securityManager != null) ? securityManager.getThreadGroup() : Thread.currentThread().getThreadGroup();
 namingPattern = "redis-" + poolCounter.getAndIncrement() + "-subscribe-thread-";
 }
 // properties

 // public methods

 @Override
 public Thread newThread(Runnable r) {
 Thread worker = new Thread(group, r, namingPattern + threadCounter.getAndIncrement(), 0);
 worker.setUncaughtExceptionHandler(new Thread.UncaughtExceptionHandler() {
 @Override
 public void uncaughtException(Thread t, Throwable e) {
 if (logger.isErrorEnabled()) {
 logger.error(e.getMessage(), e);
 }
 }
 });
 if (worker.isDaemon()) {
 worker.setDaemon(false);
 }
 if (worker.getPriority() != Thread.NORM_PRIORITY) {
 worker.setPriority(Thread.NORM_PRIORITY);
 }
 return worker;
 }

 // protected methods

 // friendly methods

 // private methods

 // inner class

 // test main
}

（5）Jedis客户端测试类（RedisTest.java）

package com.mcc.core.test;

import com.mcc.core.redis.CommandHandler;
import com.mcc.core.redis.RedisClient;

import java.util.concurrent.atomic.AtomicBoolean;

/**
* redis测试类
*
* @author <a href="mailto:417877417@qq.com">menergy</a>
* DateTime: 13-12-21 下午2:51
*/
public class RedisTest {

 public static void main() {
 final RedisClient redisClient = RedisClient.getInstance("192.168.8.69", 6379, "", 1);

 //
 final AtomicBoolean needLoad = new AtomicBoolean(true);
 // 监听Redis Cluster 存活状态
 redisClient.setAliveEventListener(new RedisClient.AliveEventListener() {
 /**
 * Redis Cluster 存活状态发生变化事件 
 * 每十秒触发一次
 *
 * @param masterAlive Redis Master Server是否存活
 * @param slaveAlive Redis Slave Server 是否存活
 */
 @Override
 public void onEvents(boolean masterAlive, boolean slaveAlive) {
 if (masterAlive) {
 // 确认Redis Cluster 后初始化配置
 if (needLoad.get()) {
 // 通过Redis的发布订阅实现指令系统
 CommandHandler handler = new CommandHandler();
 // 指令信道名称
 String channel = CommandHandler.COMMAND_CHANNEL;
 // 注册指令信道服务
 redisClient.publish(channel, "");
 // 订阅并监听指令信道
 redisClient.subscribe(handler, channel);
 // 标记为已配置
 needLoad.set(false);
 }
 } else {
 // 继续等待Redis Cluster存活通知
 }
 }
 });
 }
}

秒客网

Redis的java客户端Jedis