/**
* Created by orchid0809 on 15/7/9.
*/
@Component
public class JedisClient {
private static final Logger log = LoggerFactory.getLogger(JedisClient.class);
/**
* 冲突延时 1000ms
*/
private static final int MUTEX_EXP = 1000;
private static final List<Class> SIMPLE_CLASS_OBJ = Lists.newArrayList();
static {
SIMPLE_CLASS_OBJ.add(Number.class);
SIMPLE_CLASS_OBJ.add(String.class);
SIMPLE_CLASS_OBJ.add(Boolean.class);
}
private ShardedJedisPool jedisPool;
private boolean connected = false;
private int maxTotal = 500;//最大活动的对象个数
private int maxIdle = 1000 * 60;//对象最大空闲实例
private int maxWaitMillis = 1000 * 60;//获取对象时最大等待时间
@Value("${saint.session.redis.config}")
private String redisConfig;
private static final String REDIS_CONFIG_SPLIT = "\\|";
// public JedisClient(){
// init();
// }
@SuppressWarnings("unchecked")
private static boolean isSimpleObj(Class classObj) {
for (Class c : SIMPLE_CLASS_OBJ) {
if (c.isAssignableFrom(classObj)){
return true;
}
}
return false;
}
@PostConstruct
public void init() {
if (connected) {
return;
}
if (StringUtils.isBlank(redisConfig)){
throw new RuntimeException("redis config can not be null.");
}
JedisPoolConfig config = new JedisPoolConfig();//Jedis池配置
config.setMaxTotal(maxTotal);
config.setMaxIdle(maxIdle);
config.setMaxWaitMillis(maxWaitMillis);
config.setTestOnBorrow(true);
String[] tempArray = redisConfig.split(REDIS_CONFIG_SPLIT);
List<JedisShardInfo> jdsInfoList = new ArrayList<>(tempArray.length);
for (String str : tempArray) {
if (StringUtils.isBlank(str)){
throw new RuntimeException("redis config format error. ");
}
// String[] s = str.split("\\:");
// if (s.length < 3)
// throw new RuntimeException("redis config format error. ");
JedisShardInfo jsInfo = new JedisShardInfo(str);
// jsInfo.setPassword(s[2]);
jdsInfoList.add(jsInfo);
}
jedisPool = new ShardedJedisPool(config, jdsInfoList, Hashing.MURMUR_HASH, Sharded.DEFAULT_KEY_TAG_PATTERN);
Object testResult = runTask(new Callback() {
@Override
public Object onTask(ShardedJedis jedis) {
return jedis != null;
}
}, true);
if (testResult == null || testResult.equals(false)) {
destroy();
} else {
connected = true;
}
}
private void destroy() {
jedisPool.destroy();
connected = false;
}
public boolean set(final String key, final int seconds, final Object value) {
if (key == null || value == null) {
return false;
}
Object succeed = runTask(new Callback() {
@Override
public Object onTask(ShardedJedis jedis) {
String ret;
if (isSimpleObj(value.getClass())) {
if (seconds == JedisManager.CACHE_EXP_FOREVER) {
ret = jedis.set(key, value.toString());
} else {
ret = jedis.setex(key, seconds, value.toString());
}
} else {
byte[] bKey = SafeEncoder.encode(key);
byte[] bValue = serialize(value);
if (seconds == JedisManager.CACHE_EXP_FOREVER) {
ret = jedis.set(bKey, bValue);
} else {
ret = jedis.setex(bKey, seconds, bValue);
}
}
return "OK".equals(ret);
}
});
return succeed != null && (boolean) succeed;
}
/**
* 设定一个hash对象
*
* @param key 哈希表中的key
* @param field 域
* @param value 值
* @return 如果是第一次创建，则返回true，否则为false
*/
public boolean setHash(final String key, final int seconds, final String field, final Object value) {
if (key == null || field == null || value == null) {
return false;
}
Object succeed = runTask(new Callback() {
@Override
public Object onTask(ShardedJedis jedis) {
Long ret;
if (isSimpleObj(value.getClass())) {
ret = jedis.hset(key, field, value.toString());
} else {
byte[] bKey = SafeEncoder.encode(key);
byte[] bField = SafeEncoder.encode(field);
byte[] bValue = serialize(value);
ret = jedis.hset(bKey, bField, bValue);
}
if (seconds != JedisManager.CACHE_EXP_FOREVER) {
jedis.expire(key, seconds);
}
return ret != null && ret == 1;
}
});
return succeed != null && (boolean) succeed;
}
@SuppressWarnings("unchecked")
public <T> T getHash(final String key, final String field, final Class<T> cls) {
if (field == null){
throw new IllegalArgumentException("field can not null");
}
Object ret = runTask(new Callback() {
@Override
public Object onTask(ShardedJedis jedis) {
Object obj = null;
if (isSimpleObj(cls)) {
String str = jedis.hget(key, field);
if (str != null){
obj = createSimpleObj(str, cls);
}
} else {
byte[] bs = jedis.hget(SafeEncoder.encode(key), SafeEncoder.encode(field));
if (bs != null) {
obj = deserialize(bs);
}
}
return obj;
}
});
return ret == null ? null : (T) ret;
}
@SuppressWarnings("unchecked")
public <T> List<T> getHash(final String key, final Class<T> cls, final String... fields) {
Object ret = runTask(new Callback() {
@Override
public Object onTask(ShardedJedis jedis) {
final byte[][] bfields = new byte[fields.length][];
for (int i = 0; i < bfields.length; i++) {
bfields[i] = SafeEncoder.encode(fields[i]);
}
List<byte[]> bytes = jedis.hmget(SafeEncoder.encode(key), bfields);
if (bytes == null) {
return null;
}
List<T> retList = Lists.newArrayList();
boolean isSimple = isSimpleObj(cls);
for (byte[] e : bytes) {
if (e == null) {
retList.add(null);
continue;
}
retList.add(isSimple ? createSimpleObj(SafeEncoder.encode(e), cls) : (T) deserialize(e));
}
return retList;
}
});
return ret == null ? null : (List<T>) ret;
}
/**
* 获取list的size
*/
public long getListSize(final String key) {
Object ret = runTask(new Callback() {
@Override
public Object onTask(ShardedJedis jedis) {
return jedis.llen(key);
}
});
return ret == null ? 0 : (long) ret;
}
/**
* 获取list所有元素
*/
public List<String> getListRange(final String key) {
Object ret = runTask(new Callback() {
@Override
public Object onTask(ShardedJedis jedis) {
return jedis.lrange(key, 0, jedis.llen(key) - 1);
}
});
return ret == null ? Collections.<String>emptyList() : (List<String>) ret;
}
/**
* 获取指定指定index的list存储对象
*/
@SuppressWarnings("unchecked")
public <T> T getList(final String key, final int index, final Class<T> cls) {
Object ret = runTask(new Callback() {
@Override
public Object onTask(ShardedJedis jedis) {
Object obj = null;
if (isSimpleObj(cls)) {
String str = jedis.lindex(key, index);
if (str != null){
obj = createSimpleObj(str, cls);
}
} else {
byte[] bs = jedis.lindex(SafeEncoder.encode(key), index);
if (bs != null) {
obj = deserialize(bs);
}
}
return obj;
}
});
return ret == null ? null : (T) ret;
}
/**
* 讲一批对象推送到list中
*/
public boolean pushList(final String key, final int seconds, final Object... values) {
if (key == null || values == null || values.length == 0) {
return false;
}
Object succeed = runTask(new Callback() {
@Override
public Object onTask(ShardedJedis jedis) {
Long ret;
if (isSimpleObj(values.getClass())) {
String[] array = new String[values.length];
for (int i = 0; i < values.length; i++) {
array[i] = values[i].toString();
}
ret = jedis.lpush(key, array);
} else {
byte[] bKey = SafeEncoder.encode(key);
byte[][] array = new byte[values.length][];
for (int i = 0; i < values.length; i++) {
array[i] = serialize(values[i]);
}
ret = jedis.lpush(bKey, array);
}
if (seconds != JedisManager.CACHE_EXP_FOREVER) {
jedis.expire(key, seconds);
}
return ret != null && ret == 1;
}
});
return succeed != null && (boolean) succeed;
}
/**
* 对hash表中的某个元素执行增加操作，如果操作的field非数字，则结果返回null
*
* @param value 需要增加的值
* @return 增加之后的值，如果操作的field非数字，则结果返回null
*/
public Long incrHash(final String key, final String field, final long value) {
Object ret = runTask(new Callback() {
@Override
public Object onTask(ShardedJedis jedis) {
return jedis.hincrBy(key, field, value);
}
});
return ret == null ? null : (Long) ret;
}
public Double incrHash(final String key, final String field, final double value) {
Object ret = runTask(new Callback() {
@Override
public Object onTask(ShardedJedis jedis) {
return jedis.hincrByFloat(key, field, value);
}
});
return ret == null ? null : (Double) ret;
}
@SuppressWarnings("unchecked")
public <T> T get(final String key, final Class<T> cls) {
Object ret = runTask(new Callback() {
@Override
public Object onTask(ShardedJedis jedis) {
Object obj = null;
if (isSimpleObj(cls)) {
String str = jedis.get(key);
if (str != null){
obj = createSimpleObj(str, cls);
}
} else {
byte[] bs = jedis.get(SafeEncoder.encode(key));
if (bs != null) {
obj = deserialize(bs);
}
}
return obj;
}
});
return ret == null ? null : (T) ret;
}
@SuppressWarnings("unchecked")
private <T> T createSimpleObj(String arg, Class<T> cls) {
T ret = null;
Constructor[] constructors = cls.getDeclaredConstructors();
for (Constructor t : constructors) {
Class[] parameterTypes = t.getParameterTypes();
if (parameterTypes.length == 1 && parameterTypes[0].equals(String.class)) {
try {
ret = (T) t.newInstance(arg);
} catch (Exception e) {
log.error("create simple obj error: " + e.getMessage(), e);
}
break;
}
}
return ret;
}
public Long delete(final String key) {
Object ret = runTask(new Callback() {
@Override
public Object onTask(ShardedJedis jedis) {
return jedis.del(key);
}
});
return ret == null ? 0L : (long) ret;
}
public double incrBy(final String key, final double value) {
Object ret = runTask(new Callback() {
@Override
public Object onTask(ShardedJedis jedis) {
return jedis.incrByFloat(key, value);
}
});
return ret == null ? 0 : (double) ret;
}
//+=values
public long incrBy(final String key, final long value) {
Object ret = runTask(new Callback() {
@Override
public Object onTask(ShardedJedis jedis) {
return jedis.incrBy(key, value);
}
});
return ret == null ? 0L : (long) ret;
}
//i++
public long incr(final String key) {
Object ret = runTask(new Callback() {
@Override
public Object onTask(ShardedJedis jedis) {
return jedis.incr(key);
}
});
return ret == null ? 0L : (long) ret;
}
// public boolean isConnected() {
// Object obj = runTask(new Callback() {
// @Override
// public Object onTask(ShardedJedis jedis) {
// return jedis.
// }
// });
// return obj != null && (boolean) obj;
// }
//
// public void shutdown() {
// runTask(new Callback() {
// @Override
// public Object onTask(ShardedJedis jedis) {
// return jedis.shutdown();
// }
// });
// }
/**
* 获取资源key的锁，超过指定的时间无效，0表示无时间限制
* 获取lock失败返回null
*
* @param key 资源name
* @param timeout 超时时间, 单位ms
* @return 返回与该锁配对的值，unlock时必须在key和该返回的value匹配上之后才可以unlock
* 返回null表示lock已经被占用
*/
public String lock(final String key, final long timeout) {
Object value = runTask(new Callback() {
@Override
public Object onTask(ShardedJedis jedis) {
String retValue = String.valueOf((long) (Math.random() * Long.MAX_VALUE));
long lastTimeout = MUTEX_EXP;
if (timeout > 0) {
lastTimeout = timeout;
}
String ret = jedis.set(key, retValue, "NX", "PX", lastTimeout);
return "OK".equals(ret) ? retValue : null;
}
});
return value == null ? null : (String) value;
}
public void unlock(final String key, final String lockVal) {
runTask(new Callback() {
@Override
public Object onTask(ShardedJedis jedis) {
if (lockVal.equals(jedis.get(key))) {
jedis.del(key);
}
return null;
}
});
}
private Object runTask(Callback callback) {
return runTask(callback, false);
}
/**
* @param isInit 在init中调用该方法s时，isInit必须为ture，否则会无限递归调用，那就废了，项目运行不起来了就~~~~
*/
private Object runTask(Callback callback, boolean isInit) {
if (!isInit) {
init();
}
ShardedJedis jedis = null;
try {
jedis = jedisPool.getResource();
return callback.onTask(jedis);
} catch (Exception e) {
log.error("Redis runTask error: ", e);
jedisPool.returnBrokenResource(jedis);
jedis = null;
if (!isInit && e.getCause() != null && e.getCause().getMessage().contains("Connection refused")) {
destroy();
}
} finally {
if (jedis != null){
jedisPool.returnResource(jedis);
}
}
return null;
}
private byte[] serialize(Object object) {
if (!(object instanceof Serializable)) {
throw new IllegalArgumentException("设定缓存的对象：" + object.getClass() + "无法序列化，确保 implements Serializable");
}
ObjectOutputStream objOS = null;
ByteArrayOutputStream byteAOS = new ByteArrayOutputStream();
try {
objOS = new ObjectOutputStream(byteAOS);
objOS.writeObject(object);
return byteAOS.toByteArray();
} catch (Exception e) {
log.error("serialize error: " + e.getMessage());
} finally {
try {
if (objOS != null) {
objOS.close();
}
byteAOS.close();
} catch (IOException e) {
log.error("serialize close error : " + e.getMessage());
}
}
return null;
}
private Object deserialize(byte[] bytes) {
ByteArrayInputStream byteAIS = new ByteArrayInputStream(bytes);
ObjectInputStream objIS = null;
try {
// objIS = new ObjectInputStream(byteAIS);
objIS = new ObjectInputStream(byteAIS);
return objIS.readObject();
} catch (Exception e) {
log.error("deserialize error: " + e.getMessage());
} finally {
try {
byteAIS.close();
if (objIS != null) {
objIS.close();
}
} catch (IOException e) {
log.error("deserialize close error: " + e.getMessage());
}
}
return null;
}
/**
* The TTL command returns the remaining time to live in seconds of a key
* that has an {expire(String, int) EXPIRE} set. This introspection
* capability allows a Redis client to check how many seconds a given key
* will continue to be part of the dataset.
*
* @param key key
* @return Integer reply, returns the remaining time to live in seconds of a
* key that has an EXPIRE. In Redis 2.6 or older, if the Key does
* not exists or does not have an associated expire, -1 is returned.
* In Redis 2.8 or newer, if the Key does not have an associated
* expire, -1 is returned or if the Key does not exists, -2 is
* returned.
*/
public Long ttl(final String key) {
Object ret = runTask(new Callback() {
@Override
public Object onTask(ShardedJedis jedis) {
return jedis.ttl(key);
}
});
return ret == null ? 0L : (long) ret;
}
interface Callback {
Object onTask(ShardedJedis jedis);
}
}
Processed:
0.015, SQL:
8
