Java 类redis.clients.jedis.Jedis 实例源码

/** {@inheritDoc} */
@Override
public Double zScore(String key, Object data) {
    int tries = 0;
    boolean sucess = false;
    Double retVal = null;
    do {
        tries++;
        try {
            Jedis jedis = pool.getResource();
            retVal = jedis.zscore(key, SerializationUtil.serialize(data));
            jedis.close();
            sucess = true;
        } catch (JedisConnectionException ex) {
            log.error(RedisConstants.CONN_FAILED_RETRY_MSG + tries);
            if (tries == numRetries) {
                throw ex;
            }
        }
    } while (!sucess && tries <= numRetries);
    return retVal;
}

@Override
public <T> Map<String, T> hgetAllObjects(final String bizkey, final String nameSpace, final TypeReference<T> type,
        final GetDataCallBack<T> gbs) {
    final String key = CacheUtils.getKeyByNamespace(bizkey, nameSpace);
    return this.performFunction(key, new CallBack<Map<String, T>>() {
        public Map<String, T> invoke(Jedis jedis) {
            try {
                Map<String, String> all = jedis.hgetAll(key);
                Map<String, T> allObjs = new HashMap<String, T>();
                for (Entry<String, String> item : all.entrySet()) {
                    String _key = item.getKey();
                    T _value = CacheUtils.parseObject(key, item.getValue(), type);
                    allObjs.put(_key, _value);
                }
                return allObjs;
            } catch (Exception e) {
                logger.error("key:" + key + "hgetAllObjects Exception：" + e.getMessage());
            }
            return null;
        }
    });
}

/**
 * 向List缓存中添加值
 * @param key 键
 * @param value 值
 * @return
 */
public static long listObjectAdd(String key, Object... value) {
    long result = 0;
    Jedis jedis = null;
    try {
        jedis = getResource();
        List<byte[]> list = Lists.newArrayList();
        for (Object o : value){
            list.add(toBytes(o));
        }
        result = jedis.rpush(getBytesKey(key), (byte[][])list.toArray());
        logger.debug("listObjectAdd {} = {}", key, value);
    } catch (Exception e) {
        logger.warn("listObjectAdd {} = {}", key, value, e);
    } finally {
        returnResource(jedis);
    }
    return result;
}

/**
 * 获取Set长度
 * @param setKey
 * @return
 */
public Long getLenBySet(String setKey) throws Exception {
    Jedis jds = null;
    try {
        jds = getJedis();
        jds.select(0);
        Long result = jds.scard(setKey);
        return result;
    } catch (Exception e) {
        throw e;
    } finally {
        if(jds != null)
        {
            jds.close();
        }
    }
}

@Override
public boolean expireat(String key, String val, int cacheTime) {
    Jedis jedis = getJedisPool().getResource();
    try {
        jedis.set(key, val);
        if (cacheTime > 0) {
            jedis.expire(key, cacheTime);
        }
        return true;
    } catch (Exception e) {
        logger.error("expireat ", e);
        return false;
    } finally {
        getJedisPool().returnResource(jedis);
    }
}

public String hget(String key, String field) {
    Jedis jedis = null;
    boolean success = true;
    String ret = null;
    try {
        jedis = jedisPool.getResource();
        if (jedis == null) {
            success = false;
            return ret;
        }
        ret = jedis.hget(key, field);
    } catch (Exception e) {
        success = false;
        returnBrokenResource(jedis, "hmgetString" + key, e);
    } finally {
        releaseReidsSource(success, jedis);
    }
    return ret;
}

/**
 * 获取会话列表
 * @param offset
 * @param limit
 * @return
 */
public Map getActiveSessions(int offset, int limit) {
    Map sessions = new HashMap();
    Jedis jedis = RedisUtil.getJedis();
    // 获取在线会话总数
    long total = jedis.llen(ZHENG_UPMS_SERVER_SESSION_IDS);
    // 获取当前页会话详情
    List<String> ids = jedis.lrange(ZHENG_UPMS_SERVER_SESSION_IDS, offset, (offset + limit - 1));
    List<Session> rows = new ArrayList<>();
    for (String id : ids) {
        String session = RedisUtil.get(ZHENG_UPMS_SHIRO_SESSION_ID + "_" + id);
        // 过滤redis过期session
        if (null == session) {
            RedisUtil.lrem(ZHENG_UPMS_SERVER_SESSION_IDS, 1, id);
            total = total - 1;
            continue;
        }
         rows.add(SerializableUtil.deserialize(session));
    }
    jedis.close();
    sessions.put("total", total);
    sessions.put("rows", rows);
    return sessions;
}

@Override
protected void pushWhenNoRepeat(Task task, Request request) {
    Jedis jedis = jedisPool.getResource();
    try {
        String content = serializer.serialize(request);
        if(request.getPriority() == 0) {
            jedis.rpush(RedisKeys.getQueueNoPriorityKey(task), content);
        } else if (request.getPriority() > 0) {
            jedis.zadd(RedisKeys.getZsetPlusPriorityKey(task), request.getPriority(), content);
        } else {
            jedis.zadd(RedisKeys.getZsetMinusPriorityKey(task), request.getPriority(), content);
        }

        jedis.sadd(RedisKeys.getSetKey(task), request.key());
    } finally {
        jedis.close();
    }
}

private static void publish(int id, String string)
{
    Jedis jedis = null;
    try
    {
        jedis = SamaGamesAPI.get().getBungeeResource();
        if (jedis != null)
            jedis.publish("tsbot", SamaGamesAPI.get().getServerName() + "/" + id + ":" + string);
    }
    catch (Exception exception)
    {
        SamaGamesAPI.get().getPlugin().getLogger().log(Level.SEVERE, "Jedis error", exception);
    }
    finally
    {
        if (jedis != null)
            jedis.close();
    }
}

public void zrem(String key, String member) {
    Jedis jedis = null;
    try {
        jedis = getJedis();
        if (jedis == null) {
            logger.error("get jedis fail");
        }
        jedis.zrem(key, member);
    } catch (JedisConnectionException e) {
        if (jedis != null) {
            jedis.close();
        }
    } finally {
        returnJedisResource(jedis);
    }
}

public String lpop(String key) {
    Jedis jedis = null;
    boolean success = true;
    String ret = null;
    try {
        jedis = jedisPool.getResource();
        if (jedis == null) {
            success = false;
            return ret;
        }
        ret = jedis.lpop(key);
    } catch (Exception e) {
        success = false;
        returnBrokenResource(jedis, "lpop key:" + key, e);
    } finally {
        releaseReidsSource(success, jedis);
    }
    return ret;
}

public Long ttl(String key) {
    log.trace("get set expire " + key);
    Jedis jedis = null;
    try {
        jedis = redisConnection.getJedis();
        jedis.select(dbIndex);
        return jedis.ttl(key);
    } catch (Exception e) {
        log.warn(e.getMessage(), e);
    } finally {
        if (jedis != null) {
            jedis.close();
        }
    }
    return -2L;
}

/**
 * If key holds a hash, retrieve the value associated to the specified field.
 * <p>
 * If the field is not found or the key does not exist, a special 'nil' value is returned.
 * <p>
 * <b>Time complexity:</b> O(1)
 * 
 * @param key specified key
 * @param field hash field
 * @return Bulk reply
 */
@Override
public String hget(final String bizkey,final String nameSpace, final String field,final GetDataCallBack<String> gbs) {
    final String key = CacheUtils.getKeyByNamespace(bizkey,nameSpace);
    return this.performFunction(key, new CallBack<String>() {
        public String invoke(Jedis jedis) {
            String res = jedis.hget(key, field);
            if(StringUtils.isEmpty(res)){
                if(null!=gbs){
                    res = gbs.invoke();
                    if(StringUtils.isNotEmpty(res)){
                        hset(bizkey,nameSpace,field,res);
                    }
                }
            }
            return res;
        }
    });
}

/**
 * Create the index definition in redis
 * @param schema a schema definition, see {@link Schema}
 * @param options index option flags, see {@link IndexOptions}
 * @return true if successful
 */
public boolean createIndex(Schema schema, IndexOptions options) {
    Jedis conn = _conn();

    ArrayList<String> args = new ArrayList<>();

    args.add(indexName);

    options.serializeRedisArgs(args);

    args.add("SCHEMA");

    for (Schema.Field f : schema.fields) {
        f.serializeRedisArgs(args);
    }

    String rep = conn.getClient()
            .sendCommand(commands.getCreateCommand(),
                         args.toArray(new String[args.size()]))
            .getStatusCodeReply();
    conn.close();
    return rep.equals("OK");

}

@Override
public Long hlen(String key) {
    Jedis jedis = null;
    Long res = null;
    try {
        jedis = pool.getResource();
        res = jedis.hlen(key);
    } catch (Exception e) {

        LOGGER.error(e.getMessage());
    } finally {
        returnResource(pool, jedis);
    }
    return res;

}

/**
 * 获取List缓存
 *
 * @param key 键
 * @return 值
 */
public static List<String> getList(String key) {
    List<String> value = null;
    Jedis jedis = null;
    try {
        jedis = getResource();
        if (jedis.exists(key)) {
            value = jedis.lrange(key, 0, -1);
            logger.debug("getList {} = {}", key, value);
        }
    } catch (Exception e) {
        logger.warn("getList {} = {}", key, value, e);
    } finally {
        returnResource(jedis);
    }
    return value;
}

@Test
public void sentinelSet() {
  Jedis j = new Jedis(sentinel.getHost(), sentinel.getPort());

  try {
    Map<String, String> parameterMap = new HashMap<String, String>();
    parameterMap.put("down-after-milliseconds", String.valueOf(1234));
    parameterMap.put("parallel-syncs", String.valueOf(3));
    parameterMap.put("quorum", String.valueOf(2));
    j.sentinelSet(MASTER_NAME, parameterMap);

    List<Map<String, String>> masters = j.sentinelMasters();
    for (Map<String, String> master : masters) {
      if (master.get("name").equals(MASTER_NAME)) {
        assertEquals(1234, Integer.parseInt(master.get("down-after-milliseconds")));
        assertEquals(3, Integer.parseInt(master.get("parallel-syncs")));
        assertEquals(2, Integer.parseInt(master.get("quorum")));
      }
    }

    parameterMap.put("quorum", String.valueOf(1));
    j.sentinelSet(MASTER_NAME, parameterMap);
  } finally {
    j.close();
  }
}

/**
 * Set the the respective keys to the respective values. MSET will replace old values with new values, while
 * {@link #msetnx(String...) MSETNX} will not perform any operation at all even if just a single key already exists.
 * <p>
 * Because of this semantic MSETNX can be used in order to set different keys representing different fields of an
 * unique logic object in a way that ensures that either all the fields or none at all are set.
 * <p>
 * Both MSET and MSETNX are atomic operations. This means that for instance if the keys A and B are modified,
 * another client talking to Redis can either see the changes to both A and B at once, or no modification at all.
 * 
 * @see #msetnx(String...)
 * 
 * @param keyValues respective keys
 * @return Status code reply Basically +OK as MSET can't fail
 */
@Override
public String mset(final Map<String, String> keyValues) {
    if (isSharding()) {
        throw new RedisClientException(UNSUPPORT);
    }
    return this.performFunction("", new CallBack<String>() {
        /**
         * invoke
         * @param jedis jedis
         * @return invoke results
         */
        public String invoke(Jedis jedis) {
            return jedis.mset(CacheUtils.smapToArray(keyValues));
        }
    });
}

public void zadd(String key, double score, String member) {
    Jedis jedis = null;
    try {
        jedis = getJedis();
        if (jedis == null) {
            logger.error("get jedis fail");
        }
        jedis.zadd(key, score, member);
    } catch (JedisConnectionException e) {
        if (jedis != null) {
            jedis.close();
        }
    } finally {
        returnJedisResource(jedis);
    }
}

/**
 * 获取key的剩余时间
 * 
 * @param key
 * @return
 */
public long getKeyTTL(String key) {
    Jedis jedis = null;
    long ret = 0;
    // Set<String> keys = null;
    boolean success = true;
    try {
        jedis = jedisPool.getResource();
        if (jedis == null) {
            success = false;
            return ret;
        }
        ret = jedis.ttl(key);
    } catch (Exception e) {
        success = false;
        returnBrokenResource(jedis, "key ttl", e);
    } finally {
        releaseReidsSource(success, jedis);
    }
    return ret;
}

/**
 * 向有序集合添加元素
 *
 * @param key key
 * @param scoreMembers Map
 * @param seconds seconds
 */
public boolean zadd(String key, Map<String, Double> scoreMembers, int seconds) {
    Jedis jedis = null;
    try {
        jedis = getConnect();
        jedis.zadd(key, scoreMembers);
        if (seconds > 0) {
            jedis.expire(key, seconds);
        }
    } catch (Exception e) {
        logger.error("redis zadd data failed!", e);
        return false;
    } finally {
        close(jedis);
    }
    return true;
}

public boolean setString(String key, String value, int seconds) {
    Jedis jedis = null;
    boolean success = true;
    boolean ret = false;
    try {
        jedis = jedisPool.getResource();
        if (jedis == null) {
            success = false;
            return ret;
        }
        ret = (jedis.set(key, value) != null);
        if (seconds >= 0) {
            jedis.expire(key, seconds);
        }
    } catch (Exception e) {
        success = false;
        releaseBrokenReidsSource(jedis, key, "setString", e, true);
    } finally {
        releaseReidsSource(success, jedis);
    }
    return ret;
}

public static void main(String[] args) throws InterruptedException {
    Bench<JedisPool> bench = new JedisBench() {
        @Override
        public void executeOperation(String data, JedisPool benchInstance, int threadNumber, int iteration,
                MetricRegistry metrics) {
            Jedis jedis = benchInstance.getResource();

            Timer.Context time = metrics.timer("list").time();
            String key = "list_" + threadNumber;
            jedis.rpush(key, data);
            time.stop();

            jedis.close();
        }
    };

    Benchmark benchmark = new Benchmark(bench);
    benchmark.run(args);
}

@Override
public Double zincrby(String key, double score, String member) {
    Jedis jedis = null;
    Double res = null;
    try {
        jedis = pool.getResource();
        res = jedis.zincrby(key, score, member);
    } catch (Exception e) {

        LOGGER.error(e.getMessage());
    } finally {
        returnResource(pool, jedis);
    }
    return res;
}

@Override
public void put(String cacheKey, Object obj, int expiry) {
  Jedis jedis = null;
  try {
    jedis = pool.getResource();
    jedis.setex(cacheKey.getBytes(), expiry, getBytes(obj));
    pool.returnResource(jedis);
  } catch (Exception e) {
    LOGGER.warn("Failed to put key in cache {0}", cacheKey, e);
    pool.returnBrokenResource(jedis);
  }
}

public void set(byte[] key, byte[] value){

    Jedis jedis = null;
    try{
        jedis = get();
        if (jedis == null)
            return;
        jedis.set(key,value);
        pool.returnResource(jedis);
    }catch(Exception e){
        pool.returnBrokenResource(jedis);
    }

}

/**
 * Acqurired lock release.
 */
public synchronized void release(Jedis jedis) {
    if (locked) {
        jedis.del(lockKey);
        locked = !locked;
    }
}

private void putRedisActorPath(String identify, String identifier, String areaManagerKey) {
    Jedis jedis = getJedis();

    MapStructure<String> mapStructure = RedisStrutureBuilder.ofMap(jedis, String.class).withNameSpace(nameSpace).build();
    Map<String, String> map = mapStructure.get(areaManagerKey);

    map.put(identify, identifier);
    if (jedis != null) {
        jedis.close();
    }
}

/**
 * Tests opening an SSL/TLS connection to redis with an empty certificate
 * trust store. This test should fail because there is no trust anchor for the
 * redis server certificate.
 * 
 * @throws Exception
 */
@Test
public void connectWithShardInfoAndEmptyTrustStore() throws Exception {

  final URI uri = URI.create("rediss://localhost:6390");
  final SSLSocketFactory sslSocketFactory = createTrustNoOneSslSocketFactory();

  JedisShardInfo shardInfo = new JedisShardInfo(uri, sslSocketFactory, null, null);
  shardInfo.setPassword("foobared");

  Jedis jedis = new Jedis(shardInfo);
  try {
    jedis.get("foo");
    Assert.fail("The code did not throw the expected JedisConnectionException.");
  } catch (JedisConnectionException e) {
    Assert.assertEquals("Unexpected first inner exception.",
        SSLException.class, e.getCause().getClass());
    Assert.assertEquals("Unexpected second inner exception.",
        RuntimeException.class, e.getCause().getCause().getClass());
    Assert.assertEquals("Unexpected third inner exception.",
        InvalidAlgorithmParameterException.class, e.getCause().getCause().getCause().getClass());
  }

  try {
    jedis.close();
  } catch (Throwable e1) {
    // Expected.
  }
}

@Override
public Observable<Set<String>> listByType(String typez){

    try (Jedis jedis = pool.getResource()) {

        Set<String> result = null;
        try{
            result = jedis.hkeys(typez);
            return Observable.just(result);
        }catch(JedisDataException jde){
            return Observable.just(new HashSet<>());
        }
    }
}

/**
 * 获取指定索引的字符串
 * 
 * @param key
 *            键值
 * @param index
 *            索引
 * @return 指定索引的字符串
 */
@Override
public String lindex(final String bizkey, final String nameSpace, final long index) {
    final String key = CacheUtils.getKeyByNamespace(bizkey, nameSpace);
    return this.performFunction(key, new CallBack<String>() {
        public String invoke(Jedis jedis) {
            return jedis.lindex(key, index);
        }
    });
}

@Override
public List<String> mget(String[] bizkeys,String nameSpace) {
    final String[] key = CacheUtils.getKeyByNamespace(bizkeys,nameSpace);
       return this.performFunction("", new CallBack<List<String>>() {
           public List<String> invoke(Jedis jedis) {
               return jedis.mget(key);
           }
       });
}

/**
     * 获取资源
     * @return
     * @throws JedisException
     */
    public static Jedis getResource() throws JedisException {
        Jedis jedis = null;
        try {
            jedis = jedisPool.getResource();
//          logger.debug("getResource.", jedis);
        } catch (JedisException e) {
            logger.warn("getResource.", e);
            returnBrokenResource(jedis);
            throw e;
        }
        return jedis;
    }

/**
 * setrange
 * 
 * @param key specified key
 * @param offset offset
 * @param value string value
 * @return result
 * 
 */
@Override
public Long setrange(final String bizkey,final String nameSpace, final long offset, final String value) {
    final String key = CacheUtils.getKeyByNamespace(bizkey,nameSpace);
    return this.performFunction(key, new CallBack<Long>() {
        public Long invoke(Jedis jedis) {
            return jedis.setrange(key, offset, value);
        }
    });
}

public Long setnx(final String key, final String value) {
    return execute(new JedisExecutor<Long>() {
        @Override
        Long doInJedis(Jedis jedis) {
            return jedis.setnx(key, value);
        }
    });
}

public void release() {
    Jedis jedis = context.get();
    if (jedis != null) {
        context.remove();
        jedis.close();
        if (logger.isTraceEnabled()) {
            logger.trace("<<release a redis conn[{}]", jedis.toString());
        }
    }
}

public long ttl(String key) {
    Jedis jedis = jedisPool.getResource();
    Long result = null;
    try {
        result = jedis.ttl(key);
    } catch (Exception e) {
        jedisPool.returnBrokenResource(jedis);
    } finally {
        jedisPool.returnResource(jedis);
    }
    return result;
}

@Override
public String get(String key) {
    Jedis jedis = jedisPool.getResource();
    String result = jedis.get(key);
    jedis.close();
    return result;
}

@Override
public Request doPoll(Task task) {
    Jedis jedis = jedisPool.getResource();
    try {
        return serializer.deserialize(jedis.lpop(RedisKeys.getQueueKey(task)));
    } finally {
        jedis.close();
    }
}

@Test
public void testReturnBrokenResource() throws Exception {
  Jedis jedis = pool.getResource();

  Snapshot getSnapshot = getTracker.snapshot();
  Snapshot relSnapshot = releaseTracker.snapshot();
  relSnapshot.increment();
  pool.returnBrokenResource(jedis);
  relSnapshot.validate();
  getSnapshot.validate();
}