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

项目：RedisDirectory 文件：ShardedJedisPoolStream.java

/**
 * Use transactions to add index file and then delete the old one
 *
 * @param fileLengthKey the key using for hash file length
 * @param fileDataKey   the key using for hash file data
 * @param oldField      the old hash field
 * @param newField      the new hash field
 * @param values        the data values of the old hash field
 * @param fileLength    the data length of the old hash field
 */
@Override
public void rename(String fileLengthKey, String fileDataKey, String oldField, String newField, List<byte[]> values, long
        fileLength) {
    ShardedJedis shardedJedis = getShardedJedis();
    ShardedJedisPipeline pipelined = shardedJedis.pipelined();
    //add new file length
    pipelined.hset(fileLengthKey.getBytes(), newField.getBytes(), Longs.toByteArray(fileLength));
    //add new file content
    Long blockSize = getBlockSize(fileLength);
    for (int i = 0; i < blockSize; i++) {
        pipelined.hset(fileDataKey.getBytes(), getBlockName(newField, i), compressFilter(values.get(i)));
    }
    pipelined.sync();
    shardedJedis.close();
    values.clear();
    deleteFile(fileLengthKey, fileDataKey, oldField, blockSize);
}

项目：RedisDirectory 文件：ShardedJedisPoolStream.java

@Override
public void saveFile(String fileLengthKey, String fileDataKey, String fileName, List<byte[]> values, long fileLength) {
    ShardedJedis shardedJedis = getShardedJedis();
    ShardedJedisPipeline pipelined = shardedJedis.pipelined();
    pipelined.hset(fileLengthKey.getBytes(), fileName.getBytes(), Longs.toByteArray(fileLength));
    Long blockSize = getBlockSize(fileLength);
    for (int i = 0; i < blockSize; i++) {
        pipelined.hset(fileDataKey.getBytes(), getBlockName(fileName, i), compressFilter(values.get(i)));
        if (i % Constants.SYNC_COUNT == 0) {
            pipelined.sync();
            pipelined = shardedJedis.pipelined();
        }
    }
    pipelined.sync();
    shardedJedis.close();
    values.clear();
}

项目：RedisDirectory 文件：ShardedJedisPoolStream.java

/**
 * Use transactions to add index file and then delete the old one
 *
 * @param fileLengthKey the key using for hash file length
 * @param fileDataKey   the key using for hash file data
 * @param oldField      the old hash field
 * @param newField      the new hash field
 * @param values        the data values of the old hash field
 * @param fileLength    the data length of the old hash field
 */
@Override
public void rename(String fileLengthKey, String fileDataKey, String oldField, String newField, List<byte[]> values, long
        fileLength) {
    ShardedJedis shardedJedis = getShardedJedis();
    ShardedJedisPipeline pipelined = shardedJedis.pipelined();
    //add new file length
    pipelined.hset(fileLengthKey.getBytes(), newField.getBytes(), Longs.toByteArray(fileLength));
    //add new file content
    Long blockSize = getBlockSize(fileLength);
    for (int i = 0; i < blockSize; i++) {
        pipelined.hset(fileDataKey.getBytes(), getBlockName(newField, i), compressFilter(values.get(i)));
    }
    pipelined.sync();
    shardedJedis.close();
    values.clear();
    deleteFile(fileLengthKey, fileDataKey, oldField, blockSize);
}

项目：RedisDirectory 文件：ShardedJedisPoolStream.java

@Override
public void saveFile(String fileLengthKey, String fileDataKey, String fileName, List<byte[]> values, long fileLength) {
    ShardedJedis shardedJedis = getShardedJedis();
    ShardedJedisPipeline pipelined = shardedJedis.pipelined();
    pipelined.hset(fileLengthKey.getBytes(), fileName.getBytes(), Longs.toByteArray(fileLength));
    Long blockSize = getBlockSize(fileLength);
    for (int i = 0; i < blockSize; i++) {
        pipelined.hset(fileDataKey.getBytes(), getBlockName(fileName, i), compressFilter(values.get(i)));
        if (i % Constants.SYNC_COUNT == 0) {
            pipelined.sync();
            pipelined = shardedJedis.pipelined();
        }
    }
    pipelined.sync();
    shardedJedis.close();
    values.clear();
}

项目：cachecloud 文件：ShardedJedisPipelineTest.java

@Test
public void testSyncWithNoCommandQueued() {
  JedisShardInfo shardInfo1 = new JedisShardInfo(redis1.getHost(), redis1.getPort());
  JedisShardInfo shardInfo2 = new JedisShardInfo(redis2.getHost(), redis2.getPort());
  shardInfo1.setPassword("foobared");
  shardInfo2.setPassword("foobared");
  List<JedisShardInfo> shards = new ArrayList<JedisShardInfo>();
  shards.add(shardInfo1);
  shards.add(shardInfo2);

  ShardedJedis jedis2 = new ShardedJedis(shards);

  ShardedJedisPipeline pipeline = jedis2.pipelined();
  pipeline.sync();

  jedis2.close();

  jedis2 = new ShardedJedis(shards);
  pipeline = jedis2.pipelined();
  List<Object> resp = pipeline.syncAndReturnAll();
  assertTrue(resp.isEmpty());

  jedis2.close();
}

项目：pagecache-parent 文件：JedisManager.java

/**
 * 在hash中存入单个键值对
 * 
 * @param key
 * @param field
 * @param value
 * @param expireTime
 */
public void hsetByte(String key, String field, Object value, int expireTime) {
    ShardedJedis shardedJedis = shardedJedisPool.getResource();
    if (shardedJedis == null) {
        return;
    }
    try {
        ShardedJedisPipeline pipelined = shardedJedis.pipelined();
        pipelined.hset(serializer.serialize(key), serializer.serialize(field),
                serializer.serialize(value));
        pipelined.expire(serializer.serialize(key), expireTime);
    } catch (Exception e) {
        log.error(e.getMessage(), e);
    } finally {
        shardedJedis.close();
    }
}

项目：java-platform 文件：RedisExample.java

public void testShardPipelined() {// 0.127秒
    JedisShardInfo jedis = new JedisShardInfo("120.25.241.144", 6379);
    jedis.setPassword("b840fc02d52404542994");

    List<JedisShardInfo> shards = Arrays.asList(jedis);
    ShardedJedis sharding = new ShardedJedis(shards);
    ShardedJedisPipeline pipeline = sharding.pipelined();

    long start = System.currentTimeMillis();
    for (int i = 0; i < 1000; i++) {
        pipeline.set("n" + i, "n" + i);
        System.out.println(i);
    }
    pipeline.syncAndReturnAll();
    long end = System.currentTimeMillis();
    System.out.println("共花费：" + (end - start) / 1000.0 + "秒");

    sharding.disconnect();
    try {
        Closeables.close(sharding, true);
    } catch (IOException e) {
        e.printStackTrace();
    }
}

项目：Jedis 文件：ShardedJedisPipelineTest.java

@Test
public void testSyncWithNoCommandQueued() {
  JedisShardInfo shardInfo1 = new JedisShardInfo(redis1.getHost(), redis1.getPort());
  JedisShardInfo shardInfo2 = new JedisShardInfo(redis2.getHost(), redis2.getPort());
  shardInfo1.setPassword("foobared");
  shardInfo2.setPassword("foobared");
  List<JedisShardInfo> shards = new ArrayList<JedisShardInfo>();
  shards.add(shardInfo1);
  shards.add(shardInfo2);

  ShardedJedis jedis2 = new ShardedJedis(shards);

  ShardedJedisPipeline pipeline = jedis2.pipelined();
  pipeline.sync();

  jedis2.close();

  jedis2 = new ShardedJedis(shards);
  pipeline = jedis2.pipelined();
  List<Object> resp = pipeline.syncAndReturnAll();
  assertTrue(resp.isEmpty());

  jedis2.close();
}

项目：nano-framework 文件：RedisClientImpl.java

@Override
public Map<String, String> get(final String... keys) {
    Assert.notEmpty(keys);

    ShardedJedis jedis = null;
    try {
        jedis = POOL.getJedis(config.getRedisType());
        final ShardedJedisPipeline pipelined = jedis.pipelined();
        final Map<String, Response<String>> values = Maps.newHashMap();
        for (String key : keys) {
            values.put(key, pipelined.get(key));
        }

        pipelined.sync();
        final Map<String, String> valueMap = Maps.newHashMap();
        values.forEach((key, response) -> valueMap.put(key, response.get()));
        return valueMap;
    } catch (final Throwable e) {
        throw new RedisClientException(e.getMessage(), e);
    } finally {
        POOL.close(jedis);
    }
}

项目：nano-framework 文件：RedisClientImpl.java

@Override
public Map<String, Boolean> set(final Map<String, Object> map) {
    Assert.notEmpty(map);

    ShardedJedis jedis = null;
    try {
        jedis = POOL.getJedis(config.getRedisType());
        final ShardedJedisPipeline pipelined = jedis.pipelined();
        final Map<String, Response<String>> responses = Maps.newHashMap();
        map.forEach((key, value) -> responses.put(key, pipelined.set(key, toJSONString(value))));
        pipelined.sync();

        final Map<String, Boolean> values = Maps.newHashMap();
        responses.forEach((key, response) -> values.put(key, isOK(response.get())));
        return values;
    } catch (final Throwable e) {
        throw new RedisClientException(e.getMessage(), e);
    } finally {
        POOL.close(jedis);
    }
}

项目：nano-framework 文件：RedisClientImpl.java

@Override
public boolean setByNX(final String key, final String value, final int timeout) {
    Assert.hasText(key);
    Assert.hasText(value);
    ShardedJedis jedis = null;
    try {
        jedis = POOL.getJedis(config.getRedisType());
        final ShardedJedisPipeline pipeline = jedis.pipelined();
        final Response<Long> set = pipeline.setnx(key, value);
        final Response<Long> expire = pipeline.expire(key, timeout);
        pipeline.sync();
        return isSuccess(set.get()) && isSuccess(expire.get());
    } catch (final Throwable e) {
        throw new RedisClientException(e.getMessage(), e);
    } finally {
        POOL.close(jedis);
    }
}

项目：nano-framework 文件：RedisClientImpl.java

@Override
public Map<String, Boolean> setByNX(final Map<String, Object> map) {
    Assert.notEmpty(map);

    ShardedJedis jedis = null;
    try {
        jedis = POOL.getJedis(config.getRedisType());
        final ShardedJedisPipeline pipelined = jedis.pipelined();
        final Map<String, Response<Long>> responses = Maps.newHashMap();
        for (Entry<String, Object> entry : map.entrySet()) {
            responses.put(entry.getKey(), pipelined.setnx(entry.getKey(), toJSONString(entry.getValue())));
        }

        pipelined.sync();
        final Map<String, Boolean> values = Maps.newHashMap();
        responses.forEach((key, response) -> values.put(key, isSuccess(response.get())));
        return values;
    } catch (final Throwable e) {
        throw new RedisClientException(e.getMessage(), e);
    } finally {
        POOL.close(jedis);
    }
}

项目：nano-framework 文件：RedisClientImpl.java

@Override
public Map<String, Boolean> hsetByNX(final String key, final Map<String, Object> map) {
    Assert.hasText(key);
    Assert.notEmpty(map);

    ShardedJedis jedis = null;
    try {
        jedis = POOL.getJedis(config.getRedisType());
        final ShardedJedisPipeline pipeline = jedis.pipelined();
        final Map<String, Response<Long>> responses = Maps.newHashMap();
        for (Entry<String, Object> entry : map.entrySet()) {
            responses.put(entry.getKey(), pipeline.hsetnx(key, entry.getKey(), toJSONString(entry.getValue())));
        }

        pipeline.sync();
        final Map<String, Boolean> values = Maps.newHashMap();
        responses.forEach((field, response) -> values.put(field, isSuccess(response.get())));
        return values;
    } catch (final Throwable e) {
        throw new RedisClientException(e.getMessage(), e);
    } finally {
        POOL.close(jedis);
    }
}

项目：nano-framework 文件：RedisClientImpl.java

@Override
public List<String> lrangeltrim(final String key, final int count) {
    Assert.hasText(key);
    ShardedJedis jedis = null;
    try {
        jedis = POOL.getJedis(config.getRedisType());
        final ShardedJedisPipeline pipeline = jedis.pipelined();
        final Response<List<String>> values = pipeline.lrange(key, 0, count);
        pipeline.ltrim(key, count + 1, -1);
        pipeline.sync();
        return values.get();
    } catch (final Throwable e) {
        throw new RedisClientException(e.getMessage(), e);
    } finally {
        POOL.close(jedis);
    }
}

项目：nano-framework 文件：RedisClientImpl.java

@Override
public Map<String, Long> scard(final String... keys) {
    Assert.notEmpty(keys);
    ShardedJedis jedis = null;
    try {
        jedis = POOL.getJedis(config.getRedisType());
        final ShardedJedisPipeline pipeline = jedis.pipelined();
        final Map<String, Response<Long>> responses = Maps.newHashMap();
        for (String key : keys) {
            responses.put(key, pipeline.scard(key));
        }

        pipeline.sync();
        final Map<String, Long> values = Maps.newHashMap();
        responses.forEach((key, response) -> values.put(key, response.get()));
        return values;
    } catch (final Throwable e) {
        throw new RedisClientException(e.getMessage(), e);
    } finally {
        POOL.close(jedis);
    }
}

项目：easyooo-framework 文件：ShardedJedisOperation.java

@Override
public String sets(final String... keyvalues) {
    return exec0(new ShardedJedisCallback<String>() {
        @Override
        public String doCallback(ShardedJedis jedis) {
            Object[] args = Arrays.asList(keyvalues).toArray();
            final Map<String, Object> map = MapUtil.gmap(args);
            pipelined(new PiplineCallbackAdapter() {
                @Override
                public void doCallback(ShardedJedisPipeline pipeline) {
                    for (Entry<String, Object> entry : map.entrySet()) {
                        pipeline.set(entry.getKey(), entry.getValue().toString());
                    }
                }
            });
            return new String();
        }
    });
}

项目：easyooo-framework 文件：ShardedJedisOperation.java

@Override
public Long del(final String... keys) {
    return exec0(new ShardedJedisCallback<Long>() {
        @Override
        public Long doCallback(ShardedJedis jedis) {
            pipelined(new PiplineCallbackAdapter() {
                @Override
                public void doCallback(ShardedJedisPipeline pipeline) {
                    for (String key : keys) {
                        pipeline.del(key);
                    }
                }
            });
            return new Long(keys.length);
        }
    });
}

项目：unique-web 文件：RedisUtil.java

/**
 * 批量的 {@link #getString(String)}
 * 
 * @param keys key数组
 * @return value的集合
 */
public List<String> batchGetString(final String[] keys) {
    return new Executor<List<String>>(shardedJedisPool) {

        @Override
        List<String> execute() {
            ShardedJedisPipeline pipeline = jedis.pipelined();
            List<String> result = new ArrayList<String>(keys.length);
            List<Response<String>> responses = new ArrayList<Response<String>>(keys.length);
            for (String key : keys) {
                responses.add(pipeline.get(key));
            }
            pipeline.sync();
            for (Response<String> resp : responses) {
                result.add(resp.get());
            }
            return result;
        }
    }.getResult();
}

项目：unique-web 文件：RedisUtil.java

/**
 * 批量的{@link #hashMultipleGet(String, String...)}，在管道中执行
 * 
 * @param pairs 多个hash的多个field
 * @return 执行结果的集合
 */
public List<List<String>> batchHashMultipleGet(final List<Pair<String, String[]>> pairs) {
    return new Executor<List<List<String>>>(shardedJedisPool) {

        @Override
        List<List<String>> execute() {
            ShardedJedisPipeline pipeline = jedis.pipelined();
            List<List<String>> result = new ArrayList<List<String>>(pairs.size());
            List<Response<List<String>>> responses = new ArrayList<Response<List<String>>>(pairs.size());
            for (Pair<String, String[]> pair : pairs) {
                responses.add(pipeline.hmget(pair.getKey(), pair.getValue()));
            }
            pipeline.sync();
            for (Response<List<String>> resp : responses) {
                result.add(resp.get());
            }
            return result;
        }
    }.getResult();

}

项目：unique-web 文件：RedisUtil.java

/**
 * 批量的{@link #hashGetAll(String)}
 * 
 * @param keys key的数组
 * @return 执行结果的集合
 */
public List<Map<String, String>> batchHashGetAll(final String... keys) {
    return new Executor<List<Map<String, String>>>(shardedJedisPool) {

        @Override
        List<Map<String, String>> execute() {
            ShardedJedisPipeline pipeline = jedis.pipelined();
            List<Map<String, String>> result = new ArrayList<Map<String, String>>(keys.length);
            List<Response<Map<String, String>>> responses = new ArrayList<Response<Map<String, String>>>(keys.length);
            for (String key : keys) {
                responses.add(pipeline.hgetAll(key));
            }
            pipeline.sync();
            for (Response<Map<String, String>> resp : responses) {
                result.add(resp.get());
            }
            return result;
        }
    }.getResult();
}

项目：unique-web 文件：RedisUtil.java

/**
 * 批量的{@link #hashMultipleGet(String, String...)}，与{@link #batchHashGetAll(String...)}不同的是，返回值为Map类型
 * 
 * @param keys key的数组
 * @return 多个hash的所有filed和value
 */
public Map<String, Map<String, String>> batchHashGetAllForMap(final String... keys) {
    return new Executor<Map<String, Map<String, String>>>(shardedJedisPool) {

        @Override
        Map<String, Map<String, String>> execute() {
            ShardedJedisPipeline pipeline = jedis.pipelined();

            // 设置map容量防止rehash
            int capacity = 1;
            while ((int) (capacity * 0.75) <= keys.length) {
                capacity <<= 1;
            }
            Map<String, Map<String, String>> result = new HashMap<String, Map<String, String>>(capacity);
            List<Response<Map<String, String>>> responses = new ArrayList<Response<Map<String, String>>>(keys.length);
            for (String key : keys) {
                responses.add(pipeline.hgetAll(key));
            }
            pipeline.sync();
            for (int i = 0; i < keys.length; ++i) {
                result.put(keys[i], responses.get(i).get());
            }
            return result;
        }
    }.getResult();
}

项目：unique-web 文件：RedisUtil.java

/**
 * 一次获得多个链表的数据
 * 
 * @param keys key的数组
 * @return 执行结果
 */
public Map<String, List<String>> batchGetAllList(final List<String> keys) {
    return new Executor<Map<String, List<String>>>(shardedJedisPool) {

        @Override
        Map<String, List<String>> execute() {
            ShardedJedisPipeline pipeline = jedis.pipelined();
            Map<String, List<String>> result = new HashMap<String, List<String>>();
            List<Response<List<String>>> responses = new ArrayList<Response<List<String>>>(keys.size());
            for (String key : keys) {
                responses.add(pipeline.lrange(key, 0, -1));
            }
            pipeline.sync();
            for (int i = 0; i < keys.size(); ++i) {
                result.put(keys.get(i), responses.get(i).get());
            }
            return result;
        }
    }.getResult();
}

项目：RedisDirectory 文件：ShardedJedisPoolStream.java

/**
 * Use transactions to delete index file
 *
 * @param fileLengthKey the key using for hash file length
 * @param fileDataKey   the key using for hash file data
 * @param field         the hash field
 * @param blockSize     the index file data block size
 */
@Override
public void deleteFile(String fileLengthKey, String fileDataKey, String field, long blockSize) {
    ShardedJedis shardedJedis = getShardedJedis();
    ShardedJedisPipeline pipelined = shardedJedis.pipelined();
    //delete file length
    pipelined.hdel(fileLengthKey.getBytes(), field.getBytes());
    //delete file content
    for (int i = 0; i < blockSize; i++) {
        byte[] blockName = getBlockName(field, i);
        pipelined.hdel(fileDataKey.getBytes(), blockName);
    }
    pipelined.sync();
    shardedJedis.close();
}

项目：RedisDirectory 文件：ShardedJedisPoolStream.java

@Override
public List<byte[]> loadFileOnce(String fileDataKey, String fileName, long blockSize) {
    ShardedJedis shardedJedis = getShardedJedis();
    ShardedJedisPipeline pipelined = shardedJedis.pipelined();
    List<byte[]> res = new ArrayList<>();
    List<Response<byte[]>> temps = new ArrayList<>();
    int temp = 0;
    //如果不分批次sync容易read time out和Java heap space
    while (temp < blockSize) {
        Response<byte[]> data = pipelined.hget(fileDataKey.getBytes(), getBlockName(fileName, temp));
        temps.add(data);
        if (temp % Constants.SYNC_COUNT == 0) {
            pipelined.sync();
            res.addAll(temps.stream().map(response -> uncompressFilter(response.get())).collect(Collectors.toList()));
            temps.clear();
            pipelined = shardedJedis.pipelined();
        }
        temp++;
    }
    try {
        pipelined.sync();
    } catch (JedisConnectionException e) {
        log.error("pipelined = {}, blockSize = {}!", pipelined.toString(), blockSize);
        log.error("", e);
    } finally {
        shardedJedis.close();
    }
    res.addAll(temps.stream().map(response -> uncompressFilter(response.get())).collect(Collectors.toList()));
    temps.clear();
    return res;
}

项目：JRediClients 文件：ShardedJedisTest.java

@Test
public void shardedPipeline() {
  List<JedisShardInfo> shards = new ArrayList<JedisShardInfo>();
  shards.add(new JedisShardInfo(redis1.getHost(), redis1.getPort()));
  shards.add(new JedisShardInfo(redis2.getHost(), redis2.getPort()));
  shards.get(0).setPassword("foobared");
  shards.get(1).setPassword("foobared");
  ShardedJedis jedis = new ShardedJedis(shards);

  final List<String> keys = getKeysDifferentShard(jedis);
  jedis.set(keys.get(0), "a");
  jedis.set(keys.get(1), "b");

  assertNotSame(jedis.getShard(keys.get(0)), jedis.getShard(keys.get(1)));

  List<Object> results = jedis.pipelined(new ShardedJedisPipeline() {
    public void execute() {
      get(keys.get(0));
      get(keys.get(1));
    }
  });

  List<Object> expected = new ArrayList<Object>(2);
  expected.add(SafeEncoder.encode("a"));
  expected.add(SafeEncoder.encode("b"));

  assertEquals(2, results.size());
  assertArrayEquals(SafeEncoder.encode("a"), (byte[]) results.get(0));
  assertArrayEquals(SafeEncoder.encode("b"), (byte[]) results.get(1));
}

项目：JRediClients 文件：ShardedJedisPipelineTest.java

@Test
public void pipeline() throws UnsupportedEncodingException {
  ShardedJedisPipeline p = jedis.pipelined();
  p.set("foo", "bar");
  p.get("foo");
  List<Object> results = p.syncAndReturnAll();

  assertEquals(2, results.size());
  assertEquals("OK", results.get(0));
  assertEquals("bar", results.get(1));
}

项目：JRediClients 文件：ShardedJedisPipelineTest.java

@Test(expected = JedisDataException.class)
public void pipelineResponseWithinPipeline() {
  jedis.set("string", "foo");

  ShardedJedisPipeline p = jedis.pipelined();
  Response<String> string = p.get("string");
  string.get();
  p.sync();
}

项目：JRediClients 文件：ShardedJedisPipelineTest.java

@Test
public void canRetrieveUnsetKey() {
  ShardedJedisPipeline p = jedis.pipelined();
  Response<String> shouldNotExist = p.get(UUID.randomUUID().toString());
  p.sync();
  assertNull(shouldNotExist.get());
}

项目：jigsaw-payment 文件：RedisTemplate.java

public List<Object> pAddSet(String key, int seconds, String... values) {
    try (ShardedJedis shardedJedis = shardedJedisPool.getResource()) {
        ShardedJedisPipeline pipeline = shardedJedis.pipelined();
        for(String value : values) {
            pipeline.sadd(key, value);
        }
        pipeline.expire(key, seconds);
        List<Object> result = pipeline.syncAndReturnAll();
        result.remove(result.size() -1);
        return result;
    }
}

项目：RedisDirectory 文件：ShardedJedisPoolStream.java

/**
 * Use transactions to delete index file
 *
 * @param fileLengthKey the key using for hash file length
 * @param fileDataKey   the key using for hash file data
 * @param field         the hash field
 * @param blockSize     the index file data block size
 */
@Override
public void deleteFile(String fileLengthKey, String fileDataKey, String field, long blockSize) {
    ShardedJedis shardedJedis = getShardedJedis();
    ShardedJedisPipeline pipelined = shardedJedis.pipelined();
    //delete file length
    pipelined.hdel(fileLengthKey.getBytes(), field.getBytes());
    //delete file content
    for (int i = 0; i < blockSize; i++) {
        byte[] blockName = getBlockName(field, i);
        pipelined.hdel(fileDataKey.getBytes(), blockName);
    }
    pipelined.sync();
    shardedJedis.close();
}

项目：RedisDirectory 文件：ShardedJedisPoolStream.java

@Override
public List<byte[]> loadFileOnce(String fileDataKey, String fileName, long blockSize) {
    ShardedJedis shardedJedis = getShardedJedis();
    ShardedJedisPipeline pipelined = shardedJedis.pipelined();
    List<byte[]> res = new ArrayList<>();
    List<Response<byte[]>> temps = new ArrayList<>();
    int temp = 0;
    //如果不分批次sync容易read time out和Java heap space
    while (temp < blockSize) {
        Response<byte[]> data = pipelined.hget(fileDataKey.getBytes(), getBlockName(fileName, temp));
        temps.add(data);
        if (temp % Constants.SYNC_COUNT == 0) {
            pipelined.sync();
            res.addAll(temps.stream().map(response -> uncompressFilter(response.get())).collect(Collectors.toList()));
            temps.clear();
            pipelined = shardedJedis.pipelined();
        }
        temp++;
    }
    try {
        pipelined.sync();
    } catch (JedisConnectionException e) {
        log.error("pipelined = {}, blockSize = {}!", pipelined.toString(), blockSize);
        log.error("", e);
    } finally {
        shardedJedis.close();
    }
    res.addAll(temps.stream().map(response -> uncompressFilter(response.get())).collect(Collectors.toList()));
    temps.clear();
    return res;
}

项目：cachecloud 文件：ShardedJedisPipelineTest.java

@Test
public void pipeline() throws UnsupportedEncodingException {
  ShardedJedisPipeline p = jedis.pipelined();
  p.set("foo", "bar");
  p.get("foo");
  List<Object> results = p.syncAndReturnAll();

  assertEquals(2, results.size());
  assertEquals("OK", results.get(0));
  assertEquals("bar", results.get(1));
}

项目：cachecloud 文件：ShardedJedisPipelineTest.java

@Test(expected = JedisDataException.class)
public void pipelineResponseWithinPipeline() {
  jedis.set("string", "foo");

  ShardedJedisPipeline p = jedis.pipelined();
  Response<String> string = p.get("string");
  string.get();
  p.sync();
}

项目：cachecloud 文件：ShardedJedisPipelineTest.java

@Test
public void canRetrieveUnsetKey() {
  ShardedJedisPipeline p = jedis.pipelined();
  Response<String> shouldNotExist = p.get(UUID.randomUUID().toString());
  p.sync();
  assertNull(shouldNotExist.get());
}

项目：java-platform 文件：RedisExample.java

public void testShardPipelinnedPool() {// 0.124秒
    JedisShardInfo jedis = new JedisShardInfo("120.25.241.144", 6379);
    jedis.setPassword("b840fc02d52404542994");

    List<JedisShardInfo> shards = Arrays.asList(jedis);
    ShardedJedisPool pool = new ShardedJedisPool(new JedisPoolConfig(), shards);

    ShardedJedis sharding = pool.getResource();
    ShardedJedisPipeline pipeline = sharding.pipelined();

    long start = System.currentTimeMillis();
    for (int i = 0; i < 1000; i++) {
        pipeline.set("n" + i, "n" + i);
        System.out.println(i);
    }
    pipeline.syncAndReturnAll();
    long end = System.currentTimeMillis();
    System.out.println("共花费：" + (end - start) / 1000.0 + "秒");

    sharding.disconnect();
    pool.destroy();

    try {
        Closeables.close(sharding, true);
        Closeables.close(pool, true);
    } catch (IOException e) {
        e.printStackTrace();
    }

}